from __future__ import annotations import csv import io from pathlib import Path from urllib.request import urlopen from zipfile import ZipFile from django.core.management.base import BaseCommand from istat.municipalities import normalize_municipality_name from istat.models import ( IstatCountry, IstatDocumentType, IstatMunicipality, IstatProvince, ) def _normalized_row(row: dict) -> dict: return {str(k).strip().lower(): (v.strip() if isinstance(v, str) else v) for k, v in row.items()} def _decode_text(raw: bytes) -> str: for encoding in ("utf-8-sig", "cp1252", "latin1"): try: return raw.decode(encoding) except UnicodeDecodeError: continue return raw.decode("latin1", errors="replace") def _read_rows_from_text(text: str): sample = text[:2048] try: dialect = csv.Sniffer().sniff(sample, delimiters=";,|\t") except csv.Error: dialect = csv.excel reader = csv.DictReader(io.StringIO(text), dialect=dialect) for row in reader: if not row: continue yield _normalized_row(row) def _read_rows(path: Path): if path.suffix.lower() == ".zip": with ZipFile(path) as archive: csv_name = next( (name for name in archive.namelist() if name.lower().endswith(".csv")), None, ) if not csv_name: raise ValueError(f"No CSV file found inside ZIP archive: {path}") yield from _read_rows_from_text(_decode_text(archive.read(csv_name))) return yield from _read_rows_from_text(_decode_text(path.read_bytes())) def _read_rows_from_url(url: str): with urlopen(url) as response: data = response.read() if url.lower().endswith(".zip"): with ZipFile(io.BytesIO(data)) as archive: csv_name = next( (name for name in archive.namelist() if name.lower().endswith(".csv")), None, ) if not csv_name: raise ValueError(f"No CSV file found inside ZIP archive: {url}") yield from _read_rows_from_text(_decode_text(archive.read(csv_name))) return yield from _read_rows_from_text(_decode_text(data)) def _country_payload_from_row(row: dict) -> dict | None: code = row.get("codice istat") or row.get("codice_istat") name = row.get("denominazione en") or row.get("denominazione it") iso_code = row.get("codice iso 3166 alpha2") or row.get("codice iso 3166 alpha3") if not code or not name: code = row.get("code") or row.get("codice") or row.get("codice_nazione") name = row.get("name") or row.get("nazione") or row.get("country") or row.get("denominazione") iso_code = row.get("iso_code") or row.get("iso") or row.get("iso2") or row.get("iso3") if not code or not name: return None if iso_code and len(str(iso_code).strip()) > 3: iso_code = None return { "code": str(code).zfill(9), "name": name, "iso_code": iso_code or None, } class Command(BaseCommand): help = "Import ISTAT/Alloggiati reference tables from CSV files." def add_arguments(self, parser): parser.add_argument("--countries", type=str, help="CSV path for countries") parser.add_argument( "--countries-url", type=str, help="Optional URL to a ZIP or CSV with ISTAT countries", ) parser.add_argument("--municipalities", type=str, help="CSV path for municipalities") parser.add_argument("--provinces", type=str, help="CSV path for provinces") parser.add_argument("--document-types", type=str, help="CSV path for document types") parser.add_argument( "--truncate", action="store_true", help="Clear existing tables before import", ) def handle(self, *args, **options): if options.get("truncate"): IstatCountry.objects.all().delete() IstatMunicipality.objects.all().delete() IstatProvince.objects.all().delete() IstatDocumentType.objects.all().delete() countries_path = options.get("countries") municipalities_path = options.get("municipalities") provinces_path = options.get("provinces") documents_path = options.get("document_types") if countries_path: path = Path(countries_path) if not path.exists(): self.stderr.write(f"Countries file not found: {path}") else: rows = [] for row in _read_rows(path): payload = _country_payload_from_row(row) if not payload: continue rows.append(IstatCountry(**payload)) IstatCountry.objects.bulk_create(rows, ignore_conflicts=True) self.stdout.write(f"Imported countries: {len(rows)}") elif options.get("countries_url"): rows = [] for row in _read_rows_from_url(options["countries_url"]): payload = _country_payload_from_row(row) if not payload: continue rows.append(IstatCountry(**payload)) IstatCountry.objects.bulk_create(rows, ignore_conflicts=True) self.stdout.write(f"Imported countries: {len(rows)}") if provinces_path: path = Path(provinces_path) if not path.exists(): self.stderr.write(f"Provinces file not found: {path}") else: rows = [] for row in _read_rows(path): code = row.get("code") or row.get("codice") or row.get("sigla") or row.get("sigla_provincia") name = row.get("name") or row.get("provincia") or row.get("denominazione") region = row.get("region") or row.get("regione") if not code or not name: continue rows.append(IstatProvince(code=str(code).upper(), name=name, region=region or None)) IstatProvince.objects.bulk_create(rows, ignore_conflicts=True) self.stdout.write(f"Imported provinces: {len(rows)}") if municipalities_path: path = Path(municipalities_path) if not path.exists(): self.stderr.write(f"Municipalities file not found: {path}") else: rows = [] for row in _read_rows(path): code = row.get("code") or row.get("codice") or row.get("codice_istat") or row.get("codice_comune") name = row.get("name") or row.get("comune") or row.get("denominazione") province = row.get("province") or row.get("provincia") or row.get("sigla_provincia") region = row.get("region") or row.get("regione") if not code or not name: continue rows.append( IstatMunicipality( code=str(code).zfill(9), name=name, normalized_name=normalize_municipality_name(name), province=(str(province).upper() if province else None), region=region or None, is_active=True, ) ) IstatMunicipality.objects.bulk_create(rows, ignore_conflicts=True) self.stdout.write(f"Imported municipalities: {len(rows)}") if documents_path: path = Path(documents_path) if not path.exists(): self.stderr.write(f"Document types file not found: {path}") else: rows = [] for row in _read_rows(path): code = row.get("code") or row.get("codice") description = row.get("description") or row.get("descrizione") or row.get("name") if not code or not description: continue rows.append(IstatDocumentType(code=str(code), description=description)) IstatDocumentType.objects.bulk_create(rows, ignore_conflicts=True) self.stdout.write(f"Imported document types: {len(rows)}")