from __future__ import annotations from calendar import monthrange from collections import defaultdict from dataclasses import dataclass from datetime import date, timedelta from decimal import Decimal from typing import Iterable from bookings.models import Booking from django.db.models import Prefetch from services.guest_night_service import ( calculate_age_on as shared_calculate_age_on, generate_guest_nights, get_booking_guest_count, get_prefetched_guests as shared_get_prefetched_guests, with_guest_night_prefetch, ) def with_city_tax_prefetch(queryset): """ Apply required DB optimizations: - property -> FK -> select_related - guests -> reverse FK -> prefetch_related """ return with_guest_night_prefetch(queryset) def get_prefetched_guests(booking): """ Use prefetched guests if available, fallback otherwise. """ return shared_get_prefetched_guests(booking) MONTH_NAMES = { 1: "January", 2: "February", 3: "March", 4: "April", 5: "May", 6: "June", 7: "July", 8: "August", 9: "September", 10: "October", 11: "November", 12: "December", } MONTH_SHORT = { 1: "Jan", 2: "Feb", 3: "Mar", 4: "Apr", 5: "May", 6: "Jun", 7: "Jul", 8: "Aug", 9: "Sep", 10: "Oct", 11: "Nov", 12: "Dec", } EXEMPTION_REASON_LABELS = { "medical": "Medical", "student": "Student", "refugee": "Refugee", "resident": "Resident", "minor": "Minor" } OTHER_EXEMPTION_LABEL = "Other Exemption" MINORS_LABEL = "Minor" @dataclass(frozen=True) class CityTaxGuestSubject: key: str booking_id: int guest_id: int | None full_name: str email: str phone: str date_of_birth: date | None explicit_exemption_label: str | None is_synthetic_minor: bool = False is_main_guest: bool = False @dataclass(frozen=True) class CityTaxNightRecord: date: date reservation_id: int guest_key: str guest_id: int | None month: int night_index: int is_taxable: bool is_exempt: bool is_platform: bool is_overflow: bool exemption_bucket: str | None rate: Decimal @dataclass(frozen=True) class CityTaxBookingContext: booking: Booking total_guests_count: int children_count: int adults_count: int overlap_nights_in_period: int main_guest_name: str main_guest_email: str main_guest_phone: str source_label: str platform_exempt: bool @dataclass(frozen=True) class CityTaxReportResult: preview_payload: dict export_rows: list[dict] night_records: list[CityTaxNightRecord] def normalize_platform_tokens(platform_value) -> set[str]: tokens: set[str] = set() if not platform_value: return tokens raw_values = ( platform_value if isinstance(platform_value, (list, tuple, set)) else [platform_value] ) for raw in raw_values: if raw is None: continue for token in str(raw).split(","): cleaned = token.strip().lower() if cleaned: tokens.add(cleaned) return tokens def normalize_reason(reason_value) -> str: if reason_value is None: return "" return str(reason_value).strip().lower() def calculate_age_on(date_of_birth, on_date: date) -> int | None: return shared_calculate_age_on(date_of_birth, on_date) def resolve_rate_for_month(month, rates_override, monthly_rate_map, default_rate): rates_override = rates_override or {} month_str = str(month) if month_str in rates_override: return Decimal(str(rates_override[month_str])) if month in rates_override: return Decimal(str(rates_override[month])) if month in monthly_rate_map: return monthly_rate_map[month] return default_rate def build_configured_exemption_map(exemption_reasons) -> dict[str, str]: configured: dict[str, str] = {} for reason in exemption_reasons or []: normalized = normalize_reason(reason) if not normalized: continue configured[normalized] = EXEMPTION_REASON_LABELS.get( normalized, str(reason).strip().title(), ) if not configured: return EXEMPTION_REASON_LABELS.copy() return configured def get_missing_city_tax_settings_fields(settings) -> list[str]: missing = [] if settings.max_taxable_nights is None: missing.append("max_taxable_nights") if settings.default_rate is None: missing.append("default_rate") if settings.minor_age_limit is None: missing.append("minor_age_limit") return missing def build_period_bounds(period: dict) -> tuple[date, date, list[int]]: year = period["year"] from_month = period["from_month"] to_month = period["to_month"] start_date = date(year, from_month, 1) if to_month == 12: end_exclusive = date(year + 1, 1, 1) else: end_exclusive = date(year, to_month + 1, 1) months_range = list(range(from_month, to_month + 1)) return start_date, end_exclusive, months_range def build_monthly_rate_map(structure, year: int) -> dict[int, Decimal]: from .models import CityTaxMonthlyRate return { rate.month: Decimal(str(rate.rate)) for rate in CityTaxMonthlyRate.objects.filter(structure=structure, year=year) } def pick_source_label(raw_platform): tokens = [p.strip() for p in str(raw_platform or "").split(",") if p.strip()] if not tokens: return "Direct booking" lowered = [token.lower() for token in tokens] if "airbnb" in lowered: return "Airbnb" token = tokens[0] if token.lower() == "booking.com": return "Booking.com" return token.title() def fmt_date_time(date_value: date) -> str: return f"{date_value.isoformat()} 00:00:00 UTC" def fmt_datetime(dt_value) -> str: if not dt_value: return "" return dt_value.strftime("%Y-%m-%d %H:%M:%S UTC") def month_end_exclusive(year: int, month: int) -> date: if month == 12: return date(year + 1, 1, 1) return date(year, month + 1, 1) def month_last_day(year: int, month: int) -> date: return date(year, month, monthrange(year, month)[1]) def main_guest(guest_list): if not guest_list: return None for guest in guest_list: if guest.is_main_guest: return guest return guest_list[0] def explicit_exemption_label(guest, configured_exemptions: dict[str, str]) -> str | None: if not guest or not getattr(guest, "is_city_tax_exempt", False): return None normalized_reason = normalize_reason(getattr(guest, "city_tax_exemption_reason", None)) if not normalized_reason: return OTHER_EXEMPTION_LABEL return configured_exemptions.get( normalized_reason, EXEMPTION_REASON_LABELS.get(normalized_reason, str(normalized_reason).title()), ) def build_guest_subjects( booking: Booking, configured_exemptions: dict[str, str], minor_age_limit: int, ) -> list[CityTaxGuestSubject]: total_guests_count = max(int(booking.adults_count or 0), 0) + max( int(booking.children_count or 0), 0, ) if total_guests_count <= 0: return [] guests = sorted( list(get_prefetched_guests(booking)), key=lambda guest: (not guest.is_main_guest, guest.id or 0), ) considered_guests = guests[: min(len(guests), total_guests_count)] known_minors = 0 for guest in considered_guests: age_on_checkin = calculate_age_on(guest.date_of_birth, booking.check_in_date) if age_on_checkin is not None and age_on_checkin < minor_age_limit: known_minors += 1 remaining_slots = total_guests_count - len(considered_guests) synthetic_minor_count = max(int(booking.children_count or 0) - known_minors, 0) synthetic_minor_count = min(synthetic_minor_count, remaining_slots) synthetic_adult_count = max(remaining_slots - synthetic_minor_count, 0) subjects = [ CityTaxGuestSubject( key=f"booking-{booking.id}-guest-{guest.id}", booking_id=booking.id, guest_id=guest.id, full_name=guest.full_name or "", email=guest.email or "", phone=guest.phone or "", date_of_birth=guest.date_of_birth, explicit_exemption_label=explicit_exemption_label( guest, configured_exemptions, ), is_main_guest=guest.is_main_guest, ) for guest in considered_guests ] for index in range(synthetic_minor_count): subjects.append( CityTaxGuestSubject( key=f"booking-{booking.id}-synthetic-child-{index + 1}", booking_id=booking.id, guest_id=None, full_name="", email="", phone="", date_of_birth=None, explicit_exemption_label=None, is_synthetic_minor=True, ) ) for index in range(synthetic_adult_count): subjects.append( CityTaxGuestSubject( key=f"booking-{booking.id}-synthetic-adult-{index + 1}", booking_id=booking.id, guest_id=None, full_name="", email="", phone="", date_of_birth=None, explicit_exemption_label=None, ) ) return subjects[:total_guests_count] def build_night_records_for_booking( *, booking: Booking, subjects: list[CityTaxGuestSubject], period_start: date, period_end_exclusive: date, cap: int, minor_age_limit: int, rates_override, monthly_rate_map, default_rate: Decimal, is_platform_exempt: bool, ) -> list[CityTaxNightRecord]: if booking.check_out_date <= booking.check_in_date: return [] records: list[CityTaxNightRecord] = [] total_nights = (booking.check_out_date - booking.check_in_date).days for subject in subjects: # Track taxable night count per month for this guest (cap is per guest per month) monthly_taxable_count: dict[int, int] = {} for night_offset in range(total_nights): night_date = booking.check_in_date + timedelta(days=night_offset) if not (period_start <= night_date < period_end_exclusive): continue month = night_date.month is_minor = subject.is_synthetic_minor if not is_minor and subject.date_of_birth: age_on_night = calculate_age_on(subject.date_of_birth, night_date) is_minor = age_on_night is not None and age_on_night < minor_age_limit exemption_bucket = None is_taxable = False is_exempt = False is_overflow = False if is_platform_exempt: is_exempt = True elif subject.explicit_exemption_label: exemption_bucket = subject.explicit_exemption_label is_exempt = True elif is_minor: exemption_bucket = MINORS_LABEL is_exempt = True else: # Cap applied per guest per month month_count = monthly_taxable_count.get(month, 0) if cap == 0 or month_count >= cap: is_overflow = True is_exempt = True else: is_taxable = True monthly_taxable_count[month] = month_count + 1 # night_index kept for export row compatibility (1-based offset in full stay) night_index = night_offset + 1 records.append( CityTaxNightRecord( date=night_date, reservation_id=booking.id, guest_key=subject.key, guest_id=subject.guest_id, month=month, night_index=night_index, is_taxable=is_taxable, is_exempt=is_exempt, is_platform=is_platform_exempt, is_overflow=is_overflow, exemption_bucket=exemption_bucket, rate=resolve_rate_for_month( month=month, rates_override=rates_override, monthly_rate_map=monthly_rate_map, default_rate=default_rate, ), ) ) return records def calculate_city_tax_report( *, bookings: Iterable[Booking], period: dict, rates_override, default_rate: Decimal, monthly_rate_map: dict[int, Decimal], max_taxable_nights: int, minor_age_limit: int, configured_exemptions: dict[str, str], platform_exemptions: set[str], platform_exemption_labels: list[str] | None = None, ) -> CityTaxReportResult: if hasattr(bookings, "select_related"): bookings = with_city_tax_prefetch(bookings) period_start, period_end_exclusive, months_range = build_period_bounds(period) year = period["year"] allowed_labels = set(configured_exemptions.values()) | {MINORS_LABEL} section_a_monthly = { month: {"guests": set(), "nights": 0, "tax": Decimal("0")} for month in months_range } section_b_monthly = { month: {"guests": set(), "nights": 0} for month in months_range } section_d_monthly = { month: {"guests": set(), "nights": 0, "tax": Decimal("0")} for month in months_range } section_c_monthly = defaultdict( lambda: { month: {"guests": set(), "nights": 0} for month in months_range } ) taxable_guest_keys = set() overflow_guest_keys = set() platform_guest_keys = set() all_guest_keys = set() total_tax = Decimal("0") night_records: list[CityTaxNightRecord] = [] booking_contexts: dict[int, CityTaxBookingContext] = {} taxable_night_counts: dict[tuple[str, int], int] = {} for booking in bookings: adults_count = max(int(booking.adults_count or 0), 0) children_count = max(int(booking.children_count or 0), 0) total_guests_count = get_booking_guest_count(booking) if total_guests_count <= 0: continue overlap_start = max(booking.check_in_date, period_start) overlap_end_exclusive = min(booking.check_out_date, period_end_exclusive) overlap_nights_in_period = max((overlap_end_exclusive - overlap_start).days, 0) if overlap_nights_in_period <= 0: continue booking_platform_tokens = normalize_platform_tokens(booking.platform) is_platform_exempt = bool(booking_platform_tokens.intersection(platform_exemptions)) guests_list = list(get_prefetched_guests(booking)) primary_guest = main_guest(guests_list) booking_contexts[booking.id] = CityTaxBookingContext( booking=booking, total_guests_count=total_guests_count, children_count=children_count, adults_count=adults_count, overlap_nights_in_period=overlap_nights_in_period, main_guest_name=primary_guest.full_name if primary_guest else "", main_guest_email=primary_guest.email if primary_guest and primary_guest.email else "", main_guest_phone=primary_guest.phone if primary_guest and primary_guest.phone else "", source_label=pick_source_label(booking.platform), platform_exempt=is_platform_exempt, ) guest_nights = sorted( generate_guest_nights( bookings, period_start=period_start, period_end_exclusive=period_end_exclusive, minor_age_limit=minor_age_limit, ), key=lambda item: (item.date, item.booking_id, item.guest_key), ) for guest_night in guest_nights: all_guest_keys.add(guest_night.guest_key) booking_context = booking_contexts.get(guest_night.booking_id) is_platform_exempt = bool(booking_context and booking_context.platform_exempt) normalized_reason = normalize_reason(guest_night.exemption_reason) exemption_bucket = None is_taxable = False is_exempt = False is_overflow = False if is_platform_exempt: is_exempt = True elif normalized_reason: exemption_bucket = configured_exemptions.get( normalized_reason, EXEMPTION_REASON_LABELS.get( normalized_reason, str(normalized_reason).title(), ), ) is_exempt = True elif guest_night.is_minor: exemption_bucket = MINORS_LABEL is_exempt = True else: guest_month_key = (guest_night.guest_key, guest_night.month) month_count = taxable_night_counts.get(guest_month_key, 0) if max_taxable_nights == 0 or month_count >= max_taxable_nights: is_overflow = True is_exempt = True else: is_taxable = True taxable_night_counts[guest_month_key] = month_count + 1 night_records.append( CityTaxNightRecord( date=guest_night.date, reservation_id=guest_night.booking_id, guest_key=guest_night.guest_key, guest_id=guest_night.guest_id, month=guest_night.month, night_index=guest_night.night_index, is_taxable=is_taxable, is_exempt=is_exempt, is_platform=is_platform_exempt, is_overflow=is_overflow, exemption_bucket=exemption_bucket, rate=resolve_rate_for_month( month=guest_night.month, rates_override=rates_override, monthly_rate_map=monthly_rate_map, default_rate=default_rate, ), ) ) for record in night_records: if record.is_taxable: section_a_monthly[record.month]["guests"].add(record.guest_key) section_a_monthly[record.month]["nights"] += 1 section_a_monthly[record.month]["tax"] += record.rate taxable_guest_keys.add(record.guest_key) total_tax += record.rate elif record.is_platform: section_d_monthly[record.month]["guests"].add(record.guest_key) section_d_monthly[record.month]["nights"] += 1 section_d_monthly[record.month]["tax"] += record.rate platform_guest_keys.add(record.guest_key) elif record.is_overflow: section_b_monthly[record.month]["guests"].add(record.guest_key) section_b_monthly[record.month]["nights"] += 1 overflow_guest_keys.add(record.guest_key) elif record.exemption_bucket and not record.is_platform: section_c_monthly[record.exemption_bucket][record.month]["guests"].add( record.guest_key ) section_c_monthly[record.exemption_bucket][record.month]["nights"] += 1 section_a_rows = [] section_b_rows = [] for month in months_range: section_a_rows.append( { "month": MONTH_NAMES[month], "guests": len(section_a_monthly[month]["guests"]), "nights": section_a_monthly[month]["nights"], "tax": round(float(section_a_monthly[month]["tax"]), 2), } ) section_b_rows.append( { "month": MONTH_NAMES[month], "guests": len(section_b_monthly[month]["guests"]), "nights": section_b_monthly[month]["nights"], } ) section_c_output = {} for reason_label, month_bucket in section_c_monthly.items(): if reason_label not in allowed_labels: continue total_reason_nights = sum(month_bucket[month]["nights"] for month in months_range) if total_reason_nights <= 0: continue section_c_output[reason_label] = [ { "month": MONTH_NAMES[month], "guests": len(month_bucket[month]["guests"]), "nights": month_bucket[month]["nights"], } for month in months_range ] section_b_total_nights = sum(section_b_monthly[month]["nights"] for month in months_range) section_c_total_nights = sum( month_bucket[month]["nights"] for reason_label, month_bucket in section_c_monthly.items() if reason_label in allowed_labels for month in months_range ) platform_nights_total = sum( section_d_monthly[month]["nights"] for month in months_range ) total_guest_nights = len(night_records) if total_guest_nights != ( sum(section_a_monthly[month]["nights"] for month in months_range) + sum(section_b_monthly[month]["nights"] for month in months_range) + section_c_total_nights + platform_nights_total ): raise ValueError("City tax calculation mismatch") preview_payload = { "summary": { "guests_subject_to_tax": len(taxable_guest_keys), "taxable_nights": sum( section_a_monthly[month]["nights"] for month in months_range ), "exempt_nights": ( section_b_total_nights + section_c_total_nights + platform_nights_total ), "total_to_pay": round(float(total_tax), 2), }, "section_a": { "rows": section_a_rows, "total": { "guests": sum( len(section_a_monthly[month]["guests"]) for month in months_range ), "nights": sum( section_a_monthly[month]["nights"] for month in months_range ), "tax": round(float(total_tax), 2), }, }, "section_b": { "rows": section_b_rows, "total": { "guests": sum( len(section_b_monthly[month]["guests"]) for month in months_range ), "nights": section_b_total_nights, }, }, "section_c": section_c_output, "section_d": { "guests": len(platform_guest_keys), "nights": sum( section_d_monthly[month]["nights"] for month in months_range ), "tax": round(float(sum( section_d_monthly[month]["tax"] for month in months_range )), 2), "platforms": sorted(platform_exemption_labels or []), }, "totals": { "guests": len(all_guest_keys), "nights": total_guest_nights, "tax": round(float(total_tax), 2), }, } export_rows = build_export_rows( booking_contexts=booking_contexts, night_records=night_records, months_range=months_range, year=year, ) return CityTaxReportResult( preview_payload=preview_payload, export_rows=export_rows, night_records=night_records, ) def build_export_rows( *, booking_contexts: dict[int, CityTaxBookingContext], night_records: list[CityTaxNightRecord], months_range: list[int], year: int, ) -> list[dict]: records_by_booking_and_month: dict[tuple[int, int], list[CityTaxNightRecord]] = defaultdict(list) for record in night_records: records_by_booking_and_month[(record.reservation_id, record.month)].append(record) rows: list[dict] = [] sorted_keys = sorted( records_by_booking_and_month.keys(), key=lambda item: ( booking_contexts[item[0]].booking.check_in_date, item[1], item[0], ), ) for booking_id, month in sorted_keys: context = booking_contexts[booking_id] booking = context.booking monthly_records = records_by_booking_and_month[(booking_id, month)] unique_dates = sorted({record.date for record in monthly_records}) if not unique_dates: continue segment_start = unique_dates[0] segment_end_exclusive = unique_dates[-1] + timedelta(days=1) ordinary_nights = sum(1 for record in monthly_records if record.is_taxable) exempt_nights = sum( 1 for record in monthly_records if record.is_exempt and not record.is_platform and not record.is_overflow ) platform_exempt_nights = sum(1 for record in monthly_records if record.is_platform) overflow_nights = sum(1 for record in monthly_records if record.is_overflow) overflow_guest_count = len( {record.guest_key for record in monthly_records if record.is_overflow} ) total_tax = sum( (record.rate for record in monthly_records if record.is_taxable), Decimal("0"), ) month_rate = monthly_records[0].rate all_guest_nights = len(monthly_records) validation_total = ordinary_nights + exempt_nights + overflow_nights + platform_exempt_nights rows.append( { "id": booking.id, "first_guest_status": "success", "client_full_name": context.main_guest_name, "check_in": fmt_date_time(segment_start), "check_out": fmt_date_time(segment_end_exclusive), "rental_name": booking.property.name if booking.property else "", "nights_count": (segment_end_exclusive - segment_start).days, "total_guests_count": context.total_guests_count, "adults": context.adults_count, "children": context.children_count, "final_price": float(booking.total_price or 0), "created_at": fmt_datetime(booking.created_at), "source": context.source_label, "guest_city_tax_base_amount": float(month_rate), "rental_tax_amount": float(month_rate), "estimated_total_city_tax": round(float(total_tax), 2), "estimated_ordinary_nights": ordinary_nights, "estimated_exempt_nights": exempt_nights, "estimated_platform_exempt_nights": platform_exempt_nights, "overflow_guest_count": overflow_guest_count, "overflow_nights": overflow_nights, "validation_total": validation_total, "client_email": context.main_guest_email, "client_phone": context.main_guest_phone, "month": month, "month_label": MONTH_SHORT[month], "all_guest_nights": all_guest_nights, "month_last_day": month_last_day(year, month), } ) return rows def calculate_booking_city_tax_total(booking: Booking) -> Decimal: """ Reuse the existing guest-night city tax engine for a single booking so financial documents stay aligned with the city-tax/ISTAT reporting logic. """ fallback_amount = Decimal(str(booking.city_tax or 0)) if not booking.check_in_date or not booking.check_out_date: return fallback_amount if booking.check_out_date <= booking.check_in_date: return Decimal("0.00") try: settings = booking.structure.city_tax_settings except Exception: return fallback_amount if not settings.is_active: return fallback_amount yearly_segments: list[dict] = [] current_year = booking.check_in_date.year final_year = (booking.check_out_date - timedelta(days=1)).year while current_year <= final_year: from_month = booking.check_in_date.month if current_year == booking.check_in_date.year else 1 to_month = ( (booking.check_out_date - timedelta(days=1)).month if current_year == final_year else 12 ) yearly_segments.append( { "year": current_year, "from_month": from_month, "to_month": to_month, } ) current_year += 1 configured_exemptions = build_configured_exemption_map(settings.exemption_reasons) platform_exemptions = normalize_platform_tokens(settings.platform_exemptions) total_tax = Decimal("0") for period in yearly_segments: result = calculate_city_tax_report( bookings=[booking], period=period, rates_override={}, default_rate=Decimal(str(settings.default_rate or 0)), monthly_rate_map=build_monthly_rate_map(booking.structure, period["year"]), max_taxable_nights=settings.max_taxable_nights, minor_age_limit=settings.minor_age_limit, configured_exemptions=configured_exemptions, platform_exemptions=platform_exemptions, platform_exemption_labels=list(settings.platform_exemptions or []), ) total_tax += sum( (record.rate for record in result.night_records if record.is_taxable), Decimal("0"), ) return total_tax.quantize(Decimal("0.01"))