#!/usr/bin/env python3 """ Import anagrafica delle 796 sedi Alleanza. Sorgente: -mat/ANAGRAFICA_SEDI ALLEANZA_struttura.xlsx, foglio "ORG. SEDI FISICHE RETE ALLEANZA" Destinazione: import-tools/db/dev.db, tabelle lookup_regioni_alleanza, lookup_regioni_geografiche, agenzie. Esegui dalla radice progetto: python3 import-tools/scripts/import_anagrafica.py """ from __future__ import annotations import os import re import sqlite3 import sys import warnings from pathlib import Path warnings.filterwarnings("ignore") try: from openpyxl import load_workbook except ImportError: sys.exit("openpyxl non installato. pip3 install --user openpyxl") ROOT = Path(__file__).resolve().parents[2] # .../Alleanza/ DB = ROOT / "import-tools" / "db" / "dev.db" # Preferisce v2 (con CODICE IMMOBILE) se presente, fallback a v1 XLSX_V2 = ROOT / "-mat" / "ANAGRAFICA_SEDI ALLEANZA_struttura_v2.xlsx" XLSX_V1 = ROOT / "-mat" / "ANAGRAFICA_SEDI ALLEANZA_struttura.xlsx" XLSX = XLSX_V2 if XLSX_V2.exists() else XLSX_V1 HAS_CODICE_IMMOBILE = XLSX == XLSX_V2 if not DB.exists(): sys.exit(f"DB non trovato: {DB}. Crea prima dev.db con schema.sql.") if not XLSX.exists(): sys.exit(f"File anagrafica non trovato: {XLSX}") def norm_text(v) -> str | None: if v is None: return None s = re.sub(r"\s+", " ", str(v).strip()) return s or None def norm_cap(v) -> str | None: if v is None: return None s = str(v).strip() if not s: return None if s.isdigit(): return s.zfill(5) return s if 4 <= len(s) <= 5 else None def to_float(v) -> float | None: if v is None or v == "": return None try: return float(str(v).replace(",", ".")) except ValueError: return None def to_int(v) -> int | None: if v is None or v == "": return None try: return int(str(v).strip()) except ValueError: return None def main(): print(f"DB: {DB}") print(f"XLSX: {XLSX}") wb = load_workbook(XLSX, data_only=True, read_only=True) ws = wb["ORG. SEDI FISICHE RETE ALLEANZA"] rows = [] for r in ws.iter_rows(min_row=2, values_only=True): if r[0] is None: continue if HAS_CODICE_IMMOBILE: (cod_imm, cod_fis, cod_ag, tipo, nome_ag, nome_isp, indirizzo, cap, citta, provincia, reg_geo, area, reg_all, lat, lon, *_rest) = r else: cod_imm = None (cod_fis, cod_ag, tipo, nome_ag, nome_isp, indirizzo, cap, citta, provincia, reg_geo, area, reg_all, lat, lon, *_rest) = r rows.append({ "codice_immobile": str(cod_imm).strip() if cod_imm else None, "codice_fisico": str(cod_fis).strip(), "codice_agenzia": to_int(cod_ag), "tipologia": norm_text(tipo), "nome_agenzia": norm_text(nome_ag), "nome_sede": norm_text(nome_isp), "indirizzo": norm_text(indirizzo), "cap": norm_cap(cap), "citta": norm_text(citta), "provincia": norm_text(provincia), "regione_geografica": norm_text(reg_geo), "area_macro": norm_text(area), "regione_alleanza": norm_text(reg_all), "latitudine": to_float(lat), "longitudine": to_float(lon), }) print(f"Sedi lette dall'anagrafica: {len(rows)}") conn = sqlite3.connect(DB) conn.execute("PRAGMA foreign_keys = ON") cur = conn.cursor() cur.execute("DELETE FROM agenzie_rischi_ambientali") cur.execute("DELETE FROM agenzie_allegati") cur.execute("DELETE FROM agenzie_categorie_protette") cur.execute("DELETE FROM agenzie") cur.execute("DELETE FROM lookup_regioni_alleanza") cur.execute("DELETE FROM lookup_regioni_geografiche") cur.execute("DELETE FROM sqlite_sequence WHERE name IN " "('agenzie','lookup_regioni_alleanza','lookup_regioni_geografiche')") reg_all_distinct = sorted({r["regione_alleanza"] for r in rows if r["regione_alleanza"]}) reg_geo_distinct = sorted({r["regione_geografica"] for r in rows if r["regione_geografica"]}) cur.executemany( "INSERT INTO lookup_regioni_alleanza(codice, ordine) VALUES (?, ?)", [(c, i) for i, c in enumerate(reg_all_distinct, start=1)], ) cur.executemany( "INSERT INTO lookup_regioni_geografiche(nome, ordine) VALUES (?, ?)", [(c, i) for i, c in enumerate(reg_geo_distinct, start=1)], ) print(f"Lookup regioni Alleanza: {len(reg_all_distinct)}") print(f"Lookup regioni geografiche: {len(reg_geo_distinct)}") reg_all_id = {row[0]: row[1] for row in cur.execute("SELECT codice, id FROM lookup_regioni_alleanza")} reg_geo_id = {row[0]: row[1] for row in cur.execute("SELECT nome, id FROM lookup_regioni_geografiche")} inserted = 0 skipped = [] anomalie = [] seen_codici = set() for r in rows: notes = [] if r["codice_fisico"] in seen_codici: skipped.append({"reason": "duplicato_codice_fisico", "row": r}) continue seen_codici.add(r["codice_fisico"]) if r["tipologia"] not in ("AG. GEN.", "ISP. AG.", "ISP. REG"): skipped.append({"reason": f"tipologia_invalida ({r['tipologia']!r})", "row": r}) continue if r["regione_geografica"] == "#N/A": notes.append("regione_geografica=#N/A nel sorgente") r["regione_geografica"] = None if r["cap"] is None: notes.append("CAP mancante") if r["latitudine"] is None or r["longitudine"] is None: notes.append("lat/lon mancanti") if r["codice_agenzia"] is None: notes.append("codice_agenzia non numerico nel sorgente") prov = r["provincia"] if not prov or len(prov) != 2: skipped.append({"reason": f"provincia_invalida ({prov!r})", "row": r}) continue try: cur.execute( """ INSERT INTO agenzie( codice_immobile, codice_fisico, codice_agenzia, tipologia, nome_agenzia, nome_sede, indirizzo, cap, citta, provincia, regione_geografica_id, area_macro, regione_alleanza_id, latitudine, longitudine, note_import ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( r["codice_immobile"], r["codice_fisico"], r["codice_agenzia"], r["tipologia"], r["nome_agenzia"], r["nome_sede"], r["indirizzo"], r["cap"], r["citta"], r["provincia"], reg_geo_id.get(r["regione_geografica"]), r["area_macro"], reg_all_id.get(r["regione_alleanza"]), r["latitudine"], r["longitudine"], "; ".join(notes) if notes else None, ), ) inserted += 1 if notes: anomalie.append({"codice_fisico": r["codice_fisico"], "note": notes}) except sqlite3.Error as e: skipped.append({"reason": f"sqlite_error: {e}", "row": r}) conn.commit() print() print("=" * 60) print(f"INSERITE: {inserted} sedi") print(f"SCARTATE: {len(skipped)}") for s in skipped[:20]: print(f" - {s['reason']}: {s['row']['codice_fisico']}") print(f"ANOMALIE (importate ma con note): {len(anomalie)}") for a in anomalie[:20]: print(f" - {a['codice_fisico']}: {a['note']}") by_tipo = dict(cur.execute("SELECT tipologia, COUNT(*) FROM agenzie GROUP BY tipologia").fetchall()) by_area = dict(cur.execute("SELECT area_macro, COUNT(*) FROM agenzie GROUP BY area_macro").fetchall()) print() print(f"Per tipologia: {by_tipo}") print(f"Per area: {by_area}") conn.close() if __name__ == "__main__": main()