from __future__ import annotations import csv import hashlib import json from collections import Counter, defaultdict from datetime import date, datetime, timedelta from functools import lru_cache from io import BytesIO from typing import Iterable from urllib.request import urlopen from zipfile import ZipFile from django.db import IntegrityError, transaction from django.utils import timezone from bookings.models import Booking from guests.models import Guest from guests.guest_defaults import ( ISTAT_NOT_SPECIFIED, ISTAT_NOT_SPECIFIED_DISPLAY, istat_not_specified_export_value, normalize_guest_for_istat, ) from istat.municipalities import ( normalize_municipality_name, resolve_residence, ) from guests.istat_utils import derive_guest_type_codes from properties.models import Property from istat.models import ( IstatCountry, IstatMunicipality, IstatProvince, IstatReservationPosition, IstatTourismType, IstatTransportType, ) from istat.xml_export.mappings.provinces import PROVINCE_MAPPINGS from services.country_utils import normalize_country, resolve_istat_country_code from services.guest_night_service import generate_guest_nights, with_guest_night_prefetch from structures.models import Structure FIELD_LENGTHS = { "guest_type": 2, "arrival_date": 10, "surname": 50, "name": 30, "gender": 1, "birth_date": 10, "birth_municipality_code": 9, "birth_province_code": 2, "country_of_birth_code": 9, "nationality_code": 9, "residence_municipality_code": 9, "residence_province_code": 2, "country_code": 9, "address": 50, "document_type": 5, "document_number": 20, "document_issue_place": 9, "departure_date": 10, "tourism_type": 30, "transport_type": 30, "rooms_occupied": 3, "rooms_available": 3, "beds_available": 4, "city_tax": 1, "position_code": 10, "mode": 1, } ISTAT_COUNTRY_REFERENCE_URL = ( "https://www.istat.it/wp-content/uploads/2024/03/" "Elenco-codici-e-denominazioni-unita-territoriali-estere.zip" ) TOURISM_VALUE_PAIRS = [ ("Cultural", "Culturale"), ("Beach", "Balneare"), ("Conference/Business", "Congressuale/Affari"), ("Trade Fair", "Fieristico"), ("Sport/Fitness", "Sportivo/Fitness"), ("School", "Scolastico"), ("Religious", "Religioso"), ("Social", "Sociale"), ("Theme Parks", "Parchi Tematici"), ("Spa/Health Treatments", "Termale/Trattamenti salute"), ("Food and Wine", "Enogastronomico"), ("Cycle Tourism", "Cicloturismo"), ("Excursion/Nature", "Escursionistico/Naturalistico"), ("Other reason", "Altro motivo"), ("Not Specified", "Non Specificato"), ] TRANSPORT_VALUE_PAIRS = [ ("Car", "Auto"), ("Plane", "Aereo"), ("Plane + Bus", "Aereo+Pullman"), ("Plane + Shuttle/Taxi/Car", "Aereo+Navetta/Taxi/Auto"), ("Plane + Train", "Aereo+Treno"), ("Train", "Treno"), ("Bus", "Pullman"), ("Caravan", "Caravan/Autocaravan"), ("Boat", "Barca/Nave/Traghetto"), ("Motorcycle", "Moto"), ("Bicycle", "Bicicletta"), ("Walking", "A piedi"), ("Other", "Altro mezzo"), ("Not Specified", "Non Specificato"), ] NOT_SPECIFIED_VALUE_PAIRS = [ (ISTAT_NOT_SPECIFIED, "Non Specificato"), ] ERROR_KEY_ALIASES = { "birth_country": "country_of_birth", "citizenship": "nationality", "residence_country": "country", } def _ascii(value: str) -> str: if value is None: return "" return str(value).replace("\r", " ").replace("\n", " ").encode("ascii", "ignore").decode() def _pad_right(value: str, length: int) -> str: raw = _ascii(value) if len(raw) > length: raw = raw[:length] return raw.ljust(length) def _pad_left(value: str, length: int, pad: str = "0") -> str: raw = _ascii(value) if len(raw) > length: raw = raw[-length:] return raw.rjust(length, pad) def _format_date(value) -> str: if not value: return "" if isinstance(value, datetime): value = value.date() if isinstance(value, date): return value.strftime("%d/%m/%Y") if isinstance(value, str): raw = value.strip() if not raw: return "" for fmt in ("%Y-%m-%d", "%d/%m/%Y"): try: return datetime.strptime(raw, fmt).strftime("%d/%m/%Y") except ValueError: continue return "" def _split_name(guest: Guest) -> tuple[str, str]: extra = guest.extra_data if isinstance(guest.extra_data, dict) else {} first_name = (extra.get("first_name") or "").strip() last_name = (extra.get("last_name") or "").strip() if first_name or last_name: return last_name, first_name full = (guest.full_name or "").strip() if not full: return "", "" parts = full.split() if len(parts) == 1: return parts[0], "" return " ".join(parts[1:]), parts[0] def _coerce_extra_data(value) -> dict: if isinstance(value, dict): return value if isinstance(value, str): raw = value.strip() if not raw: return {} try: parsed = json.loads(raw) except (TypeError, ValueError, json.JSONDecodeError): return {} return parsed if isinstance(parsed, dict) else {} return {} def _normalize(value: str) -> str: return _ascii(value).strip().lower() def _build_value_map(pairs: list[tuple[str, str]]) -> dict[str, str]: mapping: dict[str, str] = {} for english_value, italian_value in pairs: mapping[_normalize(english_value)] = italian_value mapping[_normalize(italian_value)] = italian_value return mapping def _build_istat_choice_value_map(pairs: list[tuple[str, str]]) -> dict[str, str]: return _build_value_map(NOT_SPECIFIED_VALUE_PAIRS + pairs) def _normalize_error_key(key: str) -> str: return ERROR_KEY_ALIASES.get(key, key) def _register_country_entry( *, code: str | None, names: Iterable[str | None], iso_codes: Iterable[str | None], by_code: dict[str, str], by_iso: dict[str, str], by_name: dict[str, str], name_by_code: dict[str, str], ): if not code: return canonical_code = str(code).strip() if not canonical_code: return if canonical_code.isdigit(): canonical_code = canonical_code.zfill(9) normalized_code = _normalize(canonical_code) by_code.setdefault(normalized_code, canonical_code) if normalized_code.isdigit(): by_code.setdefault(normalized_code.lstrip("0") or "0", canonical_code) for iso_code in iso_codes: normalized_iso = _normalize(iso_code or "") if normalized_iso: by_iso.setdefault(normalized_iso, canonical_code) display_name = "" for name in names: normalized_name = _normalize(name or "") if not normalized_name: continue by_name.setdefault(normalized_name, canonical_code) if not display_name: display_name = name or "" if display_name: name_by_code.setdefault(normalized_code, display_name) def _build_country_lookup(): by_code = {} by_iso = {} by_name = {} name_by_code = {} for country in IstatCountry.objects.all(): _register_country_entry( code=country.code, names=(country.name,), iso_codes=(country.iso_code,), by_code=by_code, by_iso=by_iso, by_name=by_name, name_by_code=name_by_code, ) return by_code, by_iso, by_name, name_by_code @lru_cache(maxsize=1) def _load_official_country_lookup(): try: with urlopen(ISTAT_COUNTRY_REFERENCE_URL, timeout=10) as response: archive = ZipFile(BytesIO(response.read())) except Exception: return {}, {}, {}, {} csv_name = next( (name for name in archive.namelist() if name.lower().endswith(".csv")), None, ) if not csv_name: return {}, {}, {}, {} try: text = archive.read(csv_name).decode("utf-8-sig") except UnicodeDecodeError: text = archive.read(csv_name).decode("latin1", errors="replace") by_code = {} by_iso = {} by_name = {} name_by_code = {} reader = csv.DictReader(text.splitlines(), delimiter=";") for row in reader: code = (row.get("Codice ISTAT") or "").strip() name_it = (row.get("Denominazione IT") or "").strip() name_en = (row.get("Denominazione EN") or "").strip() iso_alpha2 = (row.get("Codice ISO 3166 alpha2") or "").strip() iso_alpha3 = (row.get("Codice ISO 3166 alpha3") or "").strip() if not code or not (name_it or name_en): continue _register_country_entry( code=code, names=(name_en, name_it), iso_codes=(iso_alpha2, iso_alpha3), by_code=by_code, by_iso=by_iso, by_name=by_name, name_by_code=name_by_code, ) return by_code, by_iso, by_name, name_by_code def _resolve_country_code(value: str | None, lookups) -> str | None: if not value: return None normalized_istat_code = resolve_istat_country_code(value) if normalized_istat_code: return normalized_istat_code raw = _normalize(value) if not raw: return None by_code, by_iso, by_name, _ = lookups if raw.isdigit(): resolved = by_code.get(raw) or by_code.get(raw.zfill(9)) if resolved: return resolved else: resolved = by_iso.get(raw) or by_name.get(raw) if resolved: return resolved fallback = _load_official_country_lookup() if not fallback: return None fallback_by_code, fallback_by_iso, fallback_by_name, _ = fallback if raw.isdigit(): return fallback_by_code.get(raw) or fallback_by_code.get(raw.zfill(9)) return fallback_by_iso.get(raw) or fallback_by_name.get(raw) def _build_municipality_lookup(): by_code = {} by_name = {} province_by_code = {} for municipality in IstatMunicipality.objects.filter(is_active=True): if municipality.code: by_code[_normalize(municipality.code)] = municipality if len(municipality.code) == 9 and municipality.code.endswith("000"): by_code[_normalize(municipality.code[:6])] = municipality if municipality.name: by_name[normalize_municipality_name(municipality.name)] = municipality if municipality.normalized_name: by_name[municipality.normalized_name] = municipality if municipality.code and municipality.province: province_by_code[_normalize(municipality.code)] = municipality.province return by_code, by_name, province_by_code def _resolve_municipality(value: str | None, lookups) -> tuple[str | None, str | None]: if not value: return None, None raw = _normalize(value) if not raw: return None, None by_code, by_name, province_by_code = lookups code_key = raw.zfill(9) if raw.isdigit() and len(raw) < 9 else raw short_code_key = raw.zfill(6) if raw.isdigit() and len(raw) < 6 else raw if raw.isdigit() and (code_key in by_code or short_code_key in by_code): municipality = by_code.get(code_key) or by_code[short_code_key] return municipality.code, municipality.province normalized_name = normalize_municipality_name(value) if normalized_name in by_name: municipality = by_name[normalized_name] return municipality.code, municipality.province return None, None def _build_province_lookup(): by_code = {} by_name = {} for province in IstatProvince.objects.all(): if province.code: by_code[_normalize(province.code)] = province.code if province.name: by_name[_normalize(province.name)] = province.code return by_code, by_name def _resolve_province_code(value: str | None, lookups) -> str | None: if not value: return None raw = _normalize(value) if not raw: return None by_code, by_name = lookups if raw in by_code: return by_code[raw] return by_name.get(raw) def _country_name_for_code(code: str | None, lookups) -> str: if not code: return "" raw = _normalize(code) if not raw: return "" _, _, _, name_by_code = lookups if raw in name_by_code: return name_by_code[raw] or "" fallback = _load_official_country_lookup() if not fallback: return "" return fallback[3].get(raw, "") or "" def _gender_code(value: str | None) -> str | None: if not value: return None raw = _normalize(value) if raw in {"m", "male", "maschio"}: return "1" if raw in {"f", "female", "femmina"}: return "2" return None def _ordered_guests(guests: Iterable[Guest]) -> list[Guest]: guests = list(guests) if not guests: return [] guests.sort(key=lambda g: (not g.is_main_guest, g.id)) return guests def _guest_type_codes(group_type: str, count: int) -> list[str]: return derive_guest_type_codes(group_type, count) def _format_numeric(value: int | str | None, length: int) -> str: if value is None: return _pad_right("", length) raw = str(value).strip() if not raw: return _pad_right("", length) return _pad_left(raw, length, pad="0") def _rooms_available(structure: Structure, properties: list[Property]) -> int: if structure.total_units and structure.total_units > 0: return int(structure.total_units) return len(properties) def _beds_available(properties: list[Property]) -> int: total = 0 for prop in properties: if not prop.property_type: continue total += int(prop.property_type.num_beds or 0) total += int(prop.property_type.num_sofa_beds or 0) return total def _statistical_bookings(*, structure: Structure, start_date: date, end_date: date): end_date_exclusive = end_date + timedelta(days=1) return list( with_guest_night_prefetch( Booking.objects.filter( structure=structure, is_checked_in=True, check_in_date__lt=end_date_exclusive, check_out_date__gt=start_date, ).select_related("structure") ).order_by("check_in_date", "id") ) def _exportable_bookings(*, structure: Structure, start_date: date, end_date: date): today = timezone.localdate() completed_end_date = min(end_date, today) completed_end_date_exclusive = completed_end_date + timedelta(days=1) end_date_exclusive = end_date + timedelta(days=1) return list( Booking.objects.filter( structure=structure, is_checked_in=True, check_in_date__lt=end_date_exclusive, check_out_date__gt=start_date, check_out_date__lte=completed_end_date_exclusive, guests__isnull=False, ) .select_related("property", "property_type", "structure") .prefetch_related("guests") .distinct() .order_by("check_in_date", "id") ) def _build_guest_night_statistics(*, guest_nights): month_data = defaultdict( lambda: { "guest_occurrences": set(), "nights_stayed": 0, "arrivals": 0, "departures": 0, "properties_included": set(), "nationality_counts": Counter(), } ) properties_included = set() for guest_night in guest_nights: month_key = (guest_night.date.year, guest_night.date.month) payload = month_data[month_key] payload["guest_occurrences"].add((guest_night.booking_id, guest_night.guest_key)) payload["nights_stayed"] += 1 if guest_night.is_arrival_night: payload["arrivals"] += 1 if guest_night.is_departure_night: payload["departures"] += 1 if guest_night.property_id: payload["properties_included"].add(guest_night.property_id) properties_included.add(guest_night.property_id) if guest_night.nationality: payload["nationality_counts"][guest_night.nationality] += 1 monthly_stats = [] total_guests = 0 total_nights = 0 total_arrivals = 0 total_departures = 0 for (year, month), payload in sorted(month_data.items()): month_guest_count = len(payload["guest_occurrences"]) total_guests += month_guest_count total_nights += payload["nights_stayed"] total_arrivals += payload["arrivals"] total_departures += payload["departures"] monthly_stats.append( { "year": year, "month": month, "label": date(year, month, 1).strftime("%B %Y"), "total_guests": month_guest_count, "nights_stayed": payload["nights_stayed"], "arrivals": payload["arrivals"], "departures": payload["departures"], "properties_included": len(payload["properties_included"]), "nationalities": dict( sorted(payload["nationality_counts"].items(), key=lambda item: item[0]) ), } ) avg_stay_length = round(total_nights / total_guests, 1) if total_guests else 0 return { "total_guests": total_guests, "nights_stayed": total_nights, "avg_stay_length": avg_stay_length, "arrivals": total_arrivals, "departures": total_departures, "properties_included": len(properties_included), "monthly": monthly_stats, } def _ensure_position_code(booking: Booking, guest: Guest, seq_index: int) -> str: existing = IstatReservationPosition.objects.filter( reservation=booking, guest=guest ).first() if existing: return existing.istat_position_code base = f"{booking.id}-{seq_index:02d}" if len(base) > FIELD_LENGTHS["position_code"]: digest = hashlib.sha1(f"{booking.id}-{guest.id}".encode("utf-8")).hexdigest() base = digest[: FIELD_LENGTHS["position_code"]].upper() for _ in range(3): try: with transaction.atomic(): created = IstatReservationPosition.objects.create( reservation=booking, guest=guest, istat_position_code=base, ) return created.istat_position_code except IntegrityError: digest = hashlib.sha1( f"{booking.id}-{guest.id}-{timezone.now().timestamp()}".encode("utf-8") ).hexdigest() base = digest[: FIELD_LENGTHS["position_code"]].upper() return base[: FIELD_LENGTHS["position_code"]] def _birth_country_source(guest: Guest, extra: dict, country_lookups) -> str | None: explicit_value = guest.country_of_birth or extra.get("country_of_birth") if explicit_value: return explicit_value legacy_place_of_birth = extra.get("place_of_birth") if legacy_place_of_birth and _resolve_country_code(legacy_place_of_birth, country_lookups): return legacy_place_of_birth return None def _birth_city_source(guest: Guest, extra: dict) -> str | None: return ( extra.get("birth_city") or extra.get("birth_municipality") or extra.get("place_of_birth") or guest.city ) def _residence_municipality_source(guest: Guest, extra: dict) -> str | None: return ( extra.get("residence_municipality") or extra.get("residence_city") or extra.get("city") or guest.city or extra.get("place_of_birth") ) def _residence_province_source(guest: Guest, extra: dict) -> str | None: return ( extra.get("residence_province") or extra.get("province") or extra.get("state") or guest.region ) def _resolve_residence_municipality( guest: Guest, extra: dict, municipality_lookups, ) -> tuple[str | None, str | None]: return _resolve_municipality( _residence_municipality_source(guest, extra), municipality_lookups, ) def _resolve_residence_province( guest: Guest, extra: dict, province_lookups, municipality_province_code: str | None = None, ) -> str | None: return municipality_province_code or _resolve_province_code( _residence_province_source(guest, extra), province_lookups, ) def generate_istat_export(*, structure_id: int, start_date: date, end_date: date): export_data = build_istat_preview( structure_id=structure_id, start_date=start_date, end_date=end_date, ) records = export_data["records"] content = "\r\n".join(records) if records: content += "\r\n" filename = f"{export_data['structure_code']}_{timezone.localdate().strftime('%Y%m%d')}.txt" return { "content": content.encode("ascii"), "filename": filename, "valid_count": len(records), "invalid_records": export_data["invalid_records"], } def build_istat_preview(*, structure_id: int, start_date: date, end_date: date): structure = Structure.objects.filter(id=structure_id).first() if not structure: raise ValueError("Structure not found.") if not structure.istat_code: raise ValueError("Structure.istat_code is required for ISTAT export.") statistical_bookings = _statistical_bookings( structure=structure, start_date=start_date, end_date=end_date, ) guest_nights = generate_guest_nights( statistical_bookings, period_start=start_date, period_end_exclusive=end_date + timedelta(days=1), ) statistics = _build_guest_night_statistics(guest_nights=guest_nights) exportable_bookings = _exportable_bookings( structure=structure, start_date=start_date, end_date=end_date, ) exportable_booking_ids = {booking.id for booking in exportable_bookings} properties = list( Property.objects.filter(structure=structure).select_related("property_type") ) rooms_available = _rooms_available(structure, properties) beds_available = _beds_available(properties) country_lookups = _build_country_lookup() municipality_lookups = _build_municipality_lookup() province_lookups = _build_province_lookup() tourism_allowed = set( _normalize(name) for name in IstatTourismType.objects.values_list("name", flat=True) ) transport_allowed = set( _normalize(name) for name in IstatTransportType.objects.values_list("name", flat=True) ) tourism_map = _build_istat_choice_value_map(TOURISM_VALUE_PAIRS) transport_map = _build_istat_choice_value_map(TRANSPORT_VALUE_PAIRS) italy_code = ( _resolve_country_code("IT", country_lookups) or _resolve_country_code("Italy", country_lookups) or _resolve_country_code("Italia", country_lookups) ) records = [] invalid_records = [] record_payloads = [] for booking in statistical_bookings: guests = _ordered_guests(booking.guests.all()) if not guests: continue guest_type_codes = _guest_type_codes(booking.guest_group_type, len(guests)) for idx, guest in enumerate(guests, start=1): errors = [] guest = normalize_guest_for_istat(guest) extra = _coerce_extra_data(guest.extra_data) raw_guest_type = str( guest.guest_type or extra.get("guest_type") or "" ).strip() guest_type = raw_guest_type or ( guest_type_codes[idx - 1] if idx - 1 < len(guest_type_codes) else None ) gender_code = _gender_code(extra.get("gender") or guest.gender) birth_date = _format_date(extra.get("date_of_birth") or guest.date_of_birth) country_of_birth_code = _resolve_country_code( _birth_country_source(guest, extra, country_lookups), country_lookups, ) nationality_code = _resolve_country_code( extra.get("nationality") or guest.nationality, country_lookups, ) country_code = _resolve_country_code( extra.get("country") or guest.country, country_lookups, ) birth_city = _birth_city_source(guest, extra) birth_municipality_code, birth_province_code = _resolve_municipality( birth_city, municipality_lookups ) if not birth_province_code: birth_province_code = _resolve_province_code( extra.get("state") or guest.region, province_lookups ) residence = resolve_residence( country=extra.get("country") or guest.country, city=guest.city, province=extra.get("province"), region=guest.region, extra_data=extra, ) residence_municipality_code = residence.municipality_code residence_province_code = residence.province_code if not booking.check_in_date: errors.append("arrival_date") if not gender_code: errors.append("gender") if not birth_date: errors.append("birth_date") if not country_of_birth_code: errors.append("country_of_birth") if not nationality_code: errors.append("nationality") if not country_code: errors.append("country") if not guest_type or guest_type not in {"16", "17", "18", "19", "20"}: errors.append("guest_type") if italy_code and country_code == italy_code: if not residence_municipality_code: errors.append("residence_municipality") if ( not residence_province_code or residence_province_code not in PROVINCE_MAPPINGS or not residence.province_matches ): errors.append("residence_province") tourism_value = ( guest.tourism_type or extra.get("tourism_type") or booking.tourism_type or "" ) transport_value = ( guest.transport_type or extra.get("transport_type") or booking.transport_type or "" ) tourism_allowed_value = istat_not_specified_export_value(tourism_value) transport_allowed_value = istat_not_specified_export_value(transport_value) tourism_export_value = tourism_map.get(_normalize(tourism_value)) transport_export_value = transport_map.get(_normalize(transport_value)) if not tourism_value or not tourism_export_value: errors.append("tourism_type") elif tourism_allowed and ( _normalize(tourism_allowed_value) not in tourism_allowed and _normalize(tourism_export_value) not in tourism_allowed ): errors.append("tourism_type") if not transport_value or not transport_export_value: errors.append("transport_type") elif transport_allowed and ( _normalize(transport_allowed_value) not in transport_allowed and _normalize(transport_export_value) not in transport_allowed ): errors.append("transport_type") position_code = _ensure_position_code(booking, guest, idx) if not position_code: errors.append("position_code") surname, name = _split_name(guest) surname = surname.title() name = name.title() normalized_errors = [_normalize_error_key(error) for error in errors] guest_type_labels = { "16": "Single", "17": "Head fam.", "18": "Group leader", "19": "Fam. member", "20": "Group member", } guest_type_label = ( f"{guest_type} {guest_type_labels.get(guest_type, '')}".strip() if guest_type else "" ) gender_label = "M" if gender_code == "1" else "F" if gender_code == "2" else "" birth_country_iso2 = normalize_country( guest.country_of_birth or extra.get("country_of_birth") ) nationality_iso2 = normalize_country( extra.get("nationality") or guest.nationality ) residence_country_iso2 = normalize_country( extra.get("country") or guest.country ) status = "valid" if normalized_errors: status = "excluded" invalid_records.append( { "booking_id": booking.id, "guest_id": guest.id, "errors": normalized_errors, } ) record_payloads.append( { "booking_id": booking.id, "guest_id": guest.id, "arrival_date": _format_date(booking.check_in_date), "departure_date": _format_date(booking.check_out_date), "surname": surname, "name": name, "property_name": booking.property.name if booking.property else "", "guest_type": guest_type_label, "gender": gender_label, "birth_date": birth_date, "birth_country": birth_country_iso2 or "", "nationality": nationality_iso2 or "", "residence_country": residence_country_iso2 or "", "tourism_type": tourism_value, "transport_type": transport_value, "position_id": position_code, "mode": "1 New", "status": status, "errors": normalized_errors, } ) if normalized_errors: continue if booking.id not in exportable_booking_ids: continue rooms_occupied = 1 if guest_type in {"19", "20"}: rooms_occupied_value = "" rooms_available_value = "" beds_available_value = "" else: rooms_occupied_value = rooms_occupied rooms_available_value = rooms_available beds_available_value = beds_available fields = [ _format_numeric(guest_type, FIELD_LENGTHS["guest_type"]), _pad_right(_format_date(booking.check_in_date), FIELD_LENGTHS["arrival_date"]), _pad_right(surname, FIELD_LENGTHS["surname"]), _pad_right(name, FIELD_LENGTHS["name"]), _format_numeric(gender_code, FIELD_LENGTHS["gender"]), _pad_right(birth_date, FIELD_LENGTHS["birth_date"]), _format_numeric( birth_municipality_code, FIELD_LENGTHS["birth_municipality_code"] ), _pad_right( birth_province_code or "", FIELD_LENGTHS["birth_province_code"] ), _format_numeric( country_of_birth_code, FIELD_LENGTHS["country_of_birth_code"] ), _format_numeric(nationality_code, FIELD_LENGTHS["nationality_code"]), _format_numeric( residence_municipality_code, FIELD_LENGTHS["residence_municipality_code"] ), _pad_right( residence_province_code or "", FIELD_LENGTHS["residence_province_code"], ), _format_numeric(country_code, FIELD_LENGTHS["country_code"]), _pad_right("", FIELD_LENGTHS["address"]), _pad_right("", FIELD_LENGTHS["document_type"]), _pad_right("", FIELD_LENGTHS["document_number"]), _pad_right("", FIELD_LENGTHS["document_issue_place"]), _pad_right(_format_date(booking.check_out_date), FIELD_LENGTHS["departure_date"]), _pad_right(tourism_export_value or "", FIELD_LENGTHS["tourism_type"]), _pad_right(transport_export_value or "", FIELD_LENGTHS["transport_type"]), _format_numeric(rooms_occupied_value, FIELD_LENGTHS["rooms_occupied"]), _format_numeric(rooms_available_value, FIELD_LENGTHS["rooms_available"]), _format_numeric(beds_available_value, FIELD_LENGTHS["beds_available"]), _format_numeric(1 if (booking.city_tax or 0) > 0 else 0, FIELD_LENGTHS["city_tax"]), _pad_right(position_code, FIELD_LENGTHS["position_code"]), _format_numeric("1", FIELD_LENGTHS["mode"]), ] record = "".join(fields) records.append(record) return { "structure_code": structure.istat_code, "records": records, "record_payloads": record_payloads, "invalid_records": invalid_records, "summary": { "total_guests": statistics["total_guests"], "nights_stayed": statistics["nights_stayed"], "avg_stay_length": statistics["avg_stay_length"], "arrivals": statistics["arrivals"], "departures": statistics["departures"], "properties_included": statistics["properties_included"], "monthly": statistics["monthly"], "valid_rows": sum( 1 for payload in record_payloads if payload.get("status") == "valid" ), "excluded_rows": len(invalid_records), "warnings": len(invalid_records), }, }