from __future__ import annotations from __future__ import annotations import json from dataclasses import dataclass from datetime import date, datetime, time, timedelta from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path from typing import Any, TypedDict from django.conf import settings from django.contrib.auth.models import User from django.core.management.base import BaseCommand, CommandError from django.db import transaction from properties.models import Property, PropertyType, PropertyTypeBed from rates.models import Rate from structures.models import ChannelSettings, Structure class BedSeed(TypedDict, total=False): bed_type: str quantity: int class RateDefaultsSeed(TypedDict, total=False): start_date: str days: int min_nights: int base_price: float | int | str booking_pct: float | int | str airbnb_pct: float | int | str expedia_pct: float | int | str class PropertySeed(TypedDict, total=False): name: str internal_property_id: str floor_number: int status: int amenities: str | list[str] rate_defaults: RateDefaultsSeed class PropertyTypeSeed(TypedDict, total=False): name: str internal_property_type_id: str image_url: str property_size_sqm: float | int | str max_guests: int num_beds: int num_sofa_beds: int num_bedrooms: int num_bathrooms: int amenities: str | list[str] status: int beds: list[BedSeed] properties: list[PropertySeed] class ChannelSettingsSeed(TypedDict, total=False): default_booking_type: str default_booking_value: int default_booking_until_date: str | None booking_percentage: float | int | str airbnb_percentage: float | int | str expedia_percentage: float | int | str class StructureSeed(TypedDict, total=False): name: str structure_type: str internal_reference_code: str image_url: str base_price: int occupancy: int rating: float | int total_units: int status: str street_address: str zip_code: str country: str legal_entity_name: str tax_id_vat_number: str default_currency: str default_language: str time_zone: str default_check_in_time: str default_check_out_time: str default_tax_rate: float | int | str channel_settings: ChannelSettingsSeed rate_defaults: RateDefaultsSeed property_types: list[PropertyTypeSeed] class SeedPayload(TypedDict, total=False): rate_defaults: RateDefaultsSeed structures: list[StructureSeed] class _DryRunRollback(Exception): """Internal sentinel for dry-run rollback.""" @dataclass class SeedSummary: structures_created: int = 0 structures_updated: int = 0 channel_settings_created: int = 0 channel_settings_updated: int = 0 property_types_created: int = 0 property_types_updated: int = 0 properties_created: int = 0 properties_updated: int = 0 beds_upserted: int = 0 rates_created: int = 0 rates_updated: int = 0 rates_skipped_booked: int = 0 rates_skipped_invalid: int = 0 class Command(BaseCommand): help = ( "One-time bootstrap for structures, property types, properties, channel settings, " "and initial rates from a JSON file." ) def add_arguments(self, parser): parser.add_argument( "--owner-email", required=True, help="Email of the owner user for all created structures.", ) parser.add_argument( "--input", required=True, help="Absolute or project-relative path to the JSON seed payload.", ) parser.add_argument( "--run-key", default="structures_bootstrap_v1", help="Unique key used for one-time execution marker.", ) parser.add_argument( "--dry-run", action="store_true", help="Validate and simulate writes, then rollback.", ) parser.add_argument( "--force", action="store_true", help="Bypass one-time marker checks.", ) def handle(self, *args, **options): owner_email = options["owner_email"] input_path = self._resolve_input_path(options["input"]) run_key = self._sanitize_run_key(options["run_key"]) dry_run = options["dry_run"] force = options["force"] marker_path = self._marker_path(run_key) if marker_path.exists() and not force: raise CommandError( f"Run key '{run_key}' already executed. " f"Marker file exists at {marker_path}. Use --force to rerun." ) payload = self._load_payload(input_path) structures = payload.get("structures") or [] if not structures: raise CommandError("JSON payload must include a non-empty 'structures' array.") owner = self._resolve_owner(owner_email) summary = SeedSummary() try: with transaction.atomic(): global_rate_defaults = payload.get("rate_defaults") or {} for index, structure_seed in enumerate(structures, start=1): self.stdout.write(f"Processing structure {index}/{len(structures)}") self._process_structure( owner=owner, structure_seed=structure_seed, global_rate_defaults=global_rate_defaults, summary=summary, ) if dry_run: raise _DryRunRollback() except _DryRunRollback: self.stdout.write(self.style.WARNING("Dry-run complete. All changes rolled back.")) self._print_summary(summary) return if not dry_run: self._write_marker( marker_path=marker_path, input_path=input_path, run_key=run_key, owner=owner, ) self.stdout.write(self.style.SUCCESS("Bootstrap completed successfully.")) self._print_summary(summary) def _resolve_input_path(self, raw_path: str) -> Path: candidate = Path(raw_path) if not candidate.is_absolute(): candidate = Path(settings.BASE_DIR) / candidate if not candidate.exists(): raise CommandError(f"Input JSON file not found: {candidate}") return candidate def _sanitize_run_key(self, run_key: str) -> str: cleaned = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in run_key).strip("_") if not cleaned: raise CommandError("run-key must contain at least one alphanumeric character.") return cleaned def _marker_path(self, run_key: str) -> Path: marker_dir = Path(settings.BASE_DIR) / ".bootstrap_markers" marker_dir.mkdir(parents=True, exist_ok=True) return marker_dir / f"{run_key}.done.json" def _write_marker(self, marker_path: Path, input_path: Path, run_key: str, owner: User) -> None: marker_payload = { "run_key": run_key, "input_file": str(input_path), "owner_id": owner.id, "owner_email": owner.email, "executed_at_utc": datetime.utcnow().isoformat() + "Z", } marker_path.write_text(json.dumps(marker_payload, indent=2), encoding="utf-8") def _load_payload(self, input_path: Path) -> SeedPayload: try: payload = json.loads(input_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise CommandError(f"Invalid JSON in {input_path}: {exc}") from exc if not isinstance(payload, dict): raise CommandError("Seed payload must be a JSON object.") return payload # type: ignore[return-value] def _resolve_owner(self, owner_email: str) -> User: matches = User.objects.filter(email__iexact=owner_email).order_by("id") owner = matches.first() if owner is None: raise CommandError(f"No user found with email '{owner_email}'.") if matches.count() > 1: self.stdout.write( self.style.WARNING( f"Multiple users found for email '{owner_email}'. Using user id={owner.id}." ) ) return owner def _process_structure( self, *, owner: User, structure_seed: StructureSeed, global_rate_defaults: RateDefaultsSeed, summary: SeedSummary, ) -> None: name = self._required_text(structure_seed.get("name"), "structure.name") structure_type = self._required_text( structure_seed.get("structure_type"), f"{name}.structure_type" ) zip_code = self._required_text(structure_seed.get("zip_code"), f"{name}.zip_code") lookup = {"user": owner} internal_ref = self._optional_text(structure_seed.get("internal_reference_code")) if internal_ref: lookup["internal_reference_code"] = internal_ref else: lookup["name"] = name lookup["zip_code"] = zip_code structure = Structure.objects.filter(**lookup).order_by("id").first() structure_created = structure is None structure_fields = { "name": name, "structure_type": structure_type, "internal_reference_code": internal_ref, "image_url": self._optional_text(structure_seed.get("image_url")), "base_price": self._to_int(structure_seed.get("base_price"), f"{name}.base_price", default=0), "occupancy": self._to_int(structure_seed.get("occupancy"), f"{name}.occupancy", default=0), "rating": self._to_float(structure_seed.get("rating"), f"{name}.rating", default=0.0), "total_units": self._to_int(structure_seed.get("total_units"), f"{name}.total_units", default=0), "status": self._normalize_structure_status(structure_seed.get("status"), name), "street_address": self._optional_text(structure_seed.get("street_address")), "zip_code": zip_code, "country": self._optional_text(structure_seed.get("country")), "legal_entity_name": self._optional_text(structure_seed.get("legal_entity_name")), "tax_id_vat_number": self._optional_text(structure_seed.get("tax_id_vat_number")), "default_currency": self._optional_text(structure_seed.get("default_currency")), "default_language": self._optional_text(structure_seed.get("default_language")), "time_zone": self._optional_text(structure_seed.get("time_zone")), "default_check_in_time": self._to_time( structure_seed.get("default_check_in_time"), f"{name}.default_check_in_time", ), "default_check_out_time": self._to_time( structure_seed.get("default_check_out_time"), f"{name}.default_check_out_time", ), "default_tax_rate": self._to_decimal( structure_seed.get("default_tax_rate"), f"{name}.default_tax_rate", default=None, ), } if structure_created: structure = Structure.objects.create(user=owner, **structure_fields) summary.structures_created += 1 else: assert structure is not None for field, value in structure_fields.items(): setattr(structure, field, value) structure.save() summary.structures_updated += 1 assert structure is not None channel_settings = self._upsert_channel_settings( structure=structure, owner=owner, seed=structure_seed.get("channel_settings") or {}, summary=summary, ) structure_rate_defaults = structure_seed.get("rate_defaults") or {} property_types = structure_seed.get("property_types") or [] for property_type_seed in property_types: property_type = self._upsert_property_type( structure=structure, seed=property_type_seed, summary=summary, ) self._upsert_beds( property_type=property_type, beds=property_type_seed.get("beds") or [], summary=summary, context=f"{structure.name}/{property_type.name}", ) for property_seed in property_type_seed.get("properties") or []: property_obj = self._upsert_property( structure=structure, property_type=property_type, seed=property_seed, summary=summary, ) self._upsert_rates( property_obj=property_obj, structure=structure, channel_settings=channel_settings, global_rate_defaults=global_rate_defaults, structure_rate_defaults=structure_rate_defaults, property_rate_defaults=property_seed.get("rate_defaults") or {}, summary=summary, ) def _upsert_channel_settings( self, *, structure: Structure, owner: User, seed: ChannelSettingsSeed, summary: SeedSummary, ) -> ChannelSettings: channel_settings, created = ChannelSettings.objects.get_or_create( structure=structure, defaults={ "created_by": owner, "updated_by": owner, "default_booking_type": "relative", "default_booking_value": 6, "booking_percentage": Decimal("0"), "airbnb_percentage": Decimal("0"), "expedia_percentage": Decimal("0"), }, ) if created: summary.channel_settings_created += 1 if seed: default_booking_type = seed.get("default_booking_type") if default_booking_type is not None: normalized_type = str(default_booking_type).strip().lower() if normalized_type not in {"relative", "absolute"}: raise CommandError( f"{structure.name}.channel_settings.default_booking_type " "must be 'relative' or 'absolute'." ) channel_settings.default_booking_type = normalized_type if "default_booking_value" in seed: channel_settings.default_booking_value = self._to_int( seed.get("default_booking_value"), f"{structure.name}.channel_settings.default_booking_value", default=6, minimum=1, ) if "default_booking_until_date" in seed: channel_settings.default_booking_until_date = self._to_date( seed.get("default_booking_until_date"), f"{structure.name}.channel_settings.default_booking_until_date", default=None, ) if channel_settings.default_booking_type == "absolute" and not channel_settings.default_booking_until_date: raise CommandError( f"{structure.name}.channel_settings.default_booking_until_date is required " "when default_booking_type is 'absolute'." ) if "booking_percentage" in seed: channel_settings.booking_percentage = self._to_decimal( seed.get("booking_percentage"), f"{structure.name}.channel_settings.booking_percentage", default=Decimal("0"), ) or Decimal("0") if "airbnb_percentage" in seed: channel_settings.airbnb_percentage = self._to_decimal( seed.get("airbnb_percentage"), f"{structure.name}.channel_settings.airbnb_percentage", default=Decimal("0"), ) or Decimal("0") if "expedia_percentage" in seed: channel_settings.expedia_percentage = self._to_decimal( seed.get("expedia_percentage"), f"{structure.name}.channel_settings.expedia_percentage", default=Decimal("0"), ) or Decimal("0") channel_settings.updated_by = owner channel_settings.save() if not created: summary.channel_settings_updated += 1 return channel_settings def _upsert_property_type( self, *, structure: Structure, seed: PropertyTypeSeed, summary: SeedSummary, ) -> PropertyType: pt_name = self._required_text(seed.get("name"), f"{structure.name}.property_types[].name") lookup = {"structure": structure} internal_type_id = self._optional_text(seed.get("internal_property_type_id")) if internal_type_id: lookup["internal_property_type_id"] = internal_type_id else: lookup["name"] = pt_name property_type = PropertyType.objects.filter(**lookup).order_by("id").first() created = property_type is None beds = seed.get("beds") or [] computed_beds = sum(self._to_int(item.get("quantity"), f"{pt_name}.beds[].quantity", default=0) for item in beds) explicit_num_beds = seed.get("num_beds") num_beds = self._to_int( explicit_num_beds, f"{pt_name}.num_beds", default=computed_beds, minimum=0, ) fields = { "name": pt_name, "internal_property_type_id": internal_type_id, "image_url": self._optional_text(seed.get("image_url")), "property_size_sqm": self._to_decimal( seed.get("property_size_sqm"), f"{pt_name}.property_size_sqm", default=None, ), "max_guests": self._to_int(seed.get("max_guests"), f"{pt_name}.max_guests", default=0, minimum=0), "num_beds": num_beds, "num_sofa_beds": self._to_int( seed.get("num_sofa_beds"), f"{pt_name}.num_sofa_beds", default=0, minimum=0, ), "num_bedrooms": self._to_int( seed.get("num_bedrooms"), f"{pt_name}.num_bedrooms", default=0, minimum=0, ), "num_bathrooms": self._to_int( seed.get("num_bathrooms"), f"{pt_name}.num_bathrooms", default=0, minimum=0, ), "amenities": self._normalize_amenities(seed.get("amenities")), "status": self._to_int(seed.get("status"), f"{pt_name}.status", default=PropertyType.Status.UNMAPPED), } if created: property_type = PropertyType.objects.create(structure=structure, **fields) summary.property_types_created += 1 else: assert property_type is not None for field, value in fields.items(): setattr(property_type, field, value) property_type.save() summary.property_types_updated += 1 assert property_type is not None return property_type def _upsert_beds( self, *, property_type: PropertyType, beds: list[BedSeed], summary: SeedSummary, context: str, ) -> None: for bed in beds: bed_type = self._required_text(bed.get("bed_type"), f"{context}.beds[].bed_type") quantity = self._to_int(bed.get("quantity"), f"{context}.beds[{bed_type}].quantity", default=1, minimum=1) PropertyTypeBed.objects.update_or_create( property_type=property_type, bed_type=bed_type, defaults={"quantity": quantity}, ) summary.beds_upserted += 1 def _upsert_property( self, *, structure: Structure, property_type: PropertyType, seed: PropertySeed, summary: SeedSummary, ) -> Property: property_name = self._required_text( seed.get("name"), f"{structure.name}/{property_type.name}.properties[].name", ) lookup = {"structure": structure} internal_property_id = self._optional_text(seed.get("internal_property_id")) if internal_property_id: lookup["internal_property_id"] = internal_property_id else: lookup["name"] = property_name property_obj = Property.objects.filter(**lookup).order_by("id").first() created = property_obj is None fields = { "property_type": property_type, "name": property_name, "internal_property_id": internal_property_id, "floor_number": self._to_int( seed.get("floor_number"), f"{property_name}.floor_number", default=0, minimum=0, ), "amenities": self._normalize_amenities(seed.get("amenities")), "status": self._to_int( seed.get("status"), f"{property_name}.status", default=Property.PropertyStatus.UNMAPPED, ), } if created: property_obj = Property.objects.create(structure=structure, **fields) summary.properties_created += 1 else: assert property_obj is not None for field, value in fields.items(): setattr(property_obj, field, value) property_obj.save() summary.properties_updated += 1 assert property_obj is not None return property_obj def _upsert_rates( self, *, property_obj: Property, structure: Structure, channel_settings: ChannelSettings, global_rate_defaults: RateDefaultsSeed, structure_rate_defaults: RateDefaultsSeed, property_rate_defaults: RateDefaultsSeed, summary: SeedSummary, ) -> None: rate_defaults: RateDefaultsSeed = { **(global_rate_defaults or {}), **(structure_rate_defaults or {}), **(property_rate_defaults or {}), } days = self._to_int( rate_defaults.get("days"), f"{structure.name}/{property_obj.name}.rate_defaults.days", default=0, minimum=0, ) if days <= 0: return start_date = self._to_date( rate_defaults.get("start_date"), f"{structure.name}/{property_obj.name}.rate_defaults.start_date", default=date.today(), ) min_nights = self._to_int( rate_defaults.get("min_nights"), f"{structure.name}/{property_obj.name}.rate_defaults.min_nights", default=1, minimum=1, ) default_base = Decimal(str(structure.base_price or 0)) base_price = self._to_decimal( rate_defaults.get("base_price"), f"{structure.name}/{property_obj.name}.rate_defaults.base_price", default=default_base, ) or default_base if base_price <= 0: summary.rates_skipped_invalid += days self.stdout.write( self.style.WARNING( f"Skipping rates for {property_obj.name}: base_price must be > 0." ) ) return booking_pct = self._to_decimal( rate_defaults.get("booking_pct"), f"{structure.name}/{property_obj.name}.rate_defaults.booking_pct", default=channel_settings.booking_percentage or Decimal("0"), ) or Decimal("0") airbnb_pct = self._to_decimal( rate_defaults.get("airbnb_pct"), f"{structure.name}/{property_obj.name}.rate_defaults.airbnb_pct", default=channel_settings.airbnb_percentage or Decimal("0"), ) or Decimal("0") expedia_pct = self._to_decimal( rate_defaults.get("expedia_pct"), f"{structure.name}/{property_obj.name}.rate_defaults.expedia_pct", default=channel_settings.expedia_percentage or Decimal("0"), ) or Decimal("0") for offset in range(days): rate_date = start_date + timedelta(days=offset) defaults = { "base_price": self._money(base_price), "min_nights": min_nights, "booking": self._money(base_price * (Decimal("1") + booking_pct / Decimal("100"))), "airbnb": self._money(base_price * (Decimal("1") + airbnb_pct / Decimal("100"))), "experia": self._money(base_price * (Decimal("1") + expedia_pct / Decimal("100"))), } rate, created = Rate.objects.get_or_create( property=property_obj, date=rate_date, defaults=defaults, ) if created: summary.rates_created += 1 continue if rate.is_booked or rate.booking_ref_id: summary.rates_skipped_booked += 1 continue for field, value in defaults.items(): setattr(rate, field, value) rate.save() summary.rates_updated += 1 def _normalize_amenities(self, value: Any) -> str: if value in (None, ""): return "" if isinstance(value, str): return value.strip() if isinstance(value, list): normalized = [str(item).strip() for item in value if str(item).strip()] return ", ".join(normalized) raise CommandError("amenities must be a string, an array of strings, or null.") def _normalize_structure_status(self, value: Any, structure_name: str) -> str: if value in (None, ""): return "active" normalized = str(value).strip().lower() if normalized not in {"active", "inactive"}: raise CommandError( f"{structure_name}.status must be either 'active' or 'inactive'." ) return normalized def _required_text(self, value: Any, field_name: str) -> str: parsed = self._optional_text(value) if not parsed: raise CommandError(f"{field_name} is required.") return parsed def _optional_text(self, value: Any) -> str | None: if value is None: return None parsed = str(value).strip() return parsed or None def _to_int(self, value: Any, field_name: str, *, default: int = 0, minimum: int | None = None) -> int: if value in (None, ""): parsed = default else: try: parsed = int(value) except (TypeError, ValueError) as exc: raise CommandError(f"{field_name} must be an integer.") from exc if minimum is not None and parsed < minimum: raise CommandError(f"{field_name} must be >= {minimum}.") return parsed def _to_float(self, value: Any, field_name: str, *, default: float = 0.0) -> float: if value in (None, ""): return default try: return float(value) except (TypeError, ValueError) as exc: raise CommandError(f"{field_name} must be a number.") from exc def _to_decimal( self, value: Any, field_name: str, *, default: Decimal | None, ) -> Decimal | None: if value in (None, ""): return default try: return Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise CommandError(f"{field_name} must be a valid decimal number.") from exc def _to_time(self, value: Any, field_name: str) -> time | None: if value in (None, ""): return None if isinstance(value, time): return value if isinstance(value, str): parsed = value.strip() for fmt in ("%H:%M", "%H:%M:%S"): try: return datetime.strptime(parsed, fmt).time() except ValueError: continue raise CommandError(f"{field_name} must be in 'HH:MM' or 'HH:MM:SS' format.") def _to_date(self, value: Any, field_name: str, *, default: date | None) -> date | None: if value in (None, ""): return default if isinstance(value, date): return value if isinstance(value, str): try: return date.fromisoformat(value.strip()) except ValueError as exc: raise CommandError(f"{field_name} must be in YYYY-MM-DD format.") from exc raise CommandError(f"{field_name} must be a date string in YYYY-MM-DD format.") def _money(self, value: Decimal) -> Decimal: return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def _print_summary(self, summary: SeedSummary) -> None: self.stdout.write("Summary:") self.stdout.write(f" Structures created: {summary.structures_created}") self.stdout.write(f" Structures updated: {summary.structures_updated}") self.stdout.write(f" Channel settings created: {summary.channel_settings_created}") self.stdout.write(f" Channel settings updated: {summary.channel_settings_updated}") self.stdout.write(f" Property types created: {summary.property_types_created}") self.stdout.write(f" Property types updated: {summary.property_types_updated}") self.stdout.write(f" Properties created: {summary.properties_created}") self.stdout.write(f" Properties updated: {summary.properties_updated}") self.stdout.write(f" Beds upserted: {summary.beds_upserted}") self.stdout.write(f" Rates created: {summary.rates_created}") self.stdout.write(f" Rates updated: {summary.rates_updated}") self.stdout.write(f" Rates skipped (booked): {summary.rates_skipped_booked}") self.stdout.write(f" Rates skipped (invalid base): {summary.rates_skipped_invalid}")