from __future__ import annotations import re import unicodedata from dataclasses import dataclass from typing import Any from django.db.models import Q from rest_framework.exceptions import ValidationError from istat.models import IstatMunicipality from istat.xml_export.mappings.provinces import PROVINCE_MAPPINGS from services.country_utils import normalize_country ITALY_ISO2 = "IT" @dataclass(frozen=True) class ResolvedResidence: is_italian_resident: bool municipality: IstatMunicipality | None municipality_code: str | None municipality_name: str | None province_code: str | None province_matches: bool = True def normalize_municipality_name(value: Any) -> str: text = str(value or "").strip().lower() if not text: return "" text = unicodedata.normalize("NFKD", text) text = "".join(char for char in text if not unicodedata.combining(char)) text = re.sub(r"[\W_]+", " ", text, flags=re.UNICODE) return re.sub(r"\s+", " ", text).strip() def normalize_province_code(value: Any) -> str: return str(value or "").strip().upper() def is_italian_residence_country(country: Any) -> bool: normalized = normalize_country(country) return str(normalized or country or "").strip().upper() == ITALY_ISO2 def get_residence_municipality_source(*, city: Any = None, extra_data: dict | None = None) -> Any: extra = extra_data if isinstance(extra_data, dict) else {} return ( city or extra.get("city") or extra.get("residence_city") or extra.get("residence_municipality") ) def get_residence_province_source(*, province: Any = None, region: Any = None, extra_data: dict | None = None) -> Any: extra = extra_data if isinstance(extra_data, dict) else {} return ( province or extra.get("province") or extra.get("residence_province") or extra.get("state") or region ) def resolve_municipality( value: Any, *, province: Any = None, active_only: bool = True, ) -> IstatMunicipality | None: raw_value = str(value or "").strip() if not raw_value: return None queryset = IstatMunicipality.objects.all() if active_only: queryset = queryset.filter(is_active=True) province_code = normalize_province_code(province) if province_code: queryset = queryset.filter(province=province_code) if raw_value.isdigit(): code = raw_value.zfill(9) municipality = queryset.filter(code=code).first() if municipality is not None: return municipality normalized_name = normalize_municipality_name(raw_value) if not normalized_name: return None return queryset.filter(normalized_name=normalized_name).order_by("name", "code").first() def search_municipalities(*, province: Any = None, search: Any = None, limit: int = 25): province_code = normalize_province_code(province) queryset = IstatMunicipality.objects.filter(is_active=True) if province_code: queryset = queryset.filter(province=province_code) raw_search = str(search or "").strip() if raw_search: normalized_search = normalize_municipality_name(raw_search) query = Q(name__icontains=raw_search) if normalized_search: query |= Q(normalized_name__icontains=normalized_search) if raw_search.isdigit(): query |= Q(code__startswith=raw_search.zfill(9)[: len(raw_search)]) queryset = queryset.filter(query) return queryset.order_by("name", "province", "code")[: max(min(limit, 100), 1)] def serialize_municipality(municipality: IstatMunicipality) -> dict[str, str]: label_province = f" ({municipality.province})" if municipality.province else "" return { "code": municipality.code, "name": municipality.name, "province": municipality.province or "", "value": municipality.name, "label": f"{municipality.name}{label_province}", } def resolve_residence( *, country: Any, city: Any = None, province: Any = None, region: Any = None, extra_data: dict | None = None, require_for_italy: bool = False, ) -> ResolvedResidence: is_italian = is_italian_residence_country(country) province_code = normalize_province_code( get_residence_province_source( province=province, region=region, extra_data=extra_data, ) ) municipality_value = get_residence_municipality_source( city=city, extra_data=extra_data, ) if not is_italian: return ResolvedResidence( is_italian_resident=False, municipality=None, municipality_code=None, municipality_name=str(municipality_value).strip() if municipality_value else None, province_code=province_code or None, ) municipality = resolve_municipality(municipality_value, province=province_code) municipality_without_province = None if not municipality and municipality_value: municipality_without_province = resolve_municipality(municipality_value) if not province_code: municipality = municipality_without_province province_matches = not ( municipality_without_province is not None and province_code and municipality_without_province.province != province_code ) if not require_for_italy: return ResolvedResidence( is_italian_resident=True, municipality=municipality, municipality_code=municipality.code if municipality else None, municipality_name=municipality.name if municipality else None, province_code=province_code or (municipality.province if municipality else None), province_matches=province_matches, ) errors: dict[str, str] = {} if not province_code: errors["province"] = "Province is required for Italian residents." elif province_code not in PROVINCE_MAPPINGS: errors["province"] = "Province must be a valid Italian province code." if not municipality_value: errors["city"] = "Municipality is required for Italian residents." elif municipality is None: errors["city"] = "Municipality must be selected from the official ISTAT list." if not province_matches: errors.setdefault("province", "Province must match the selected municipality.") if errors: raise ValidationError(errors) return ResolvedResidence( is_italian_resident=True, municipality=municipality, municipality_code=municipality.code, municipality_name=municipality.name, province_code=province_code, province_matches=True, ) def validate_italian_residence( country: Any, province: Any = None, municipality: Any = None, *, region: Any = None, extra_data: dict | None = None, ) -> ResolvedResidence: return resolve_residence( country=country, city=municipality, province=province, region=region, extra_data=extra_data, require_for_italy=True, )