#!/usr/bin/env python3 """ Estrae i valori "tipici" dal master TWISTER per popolare le 5 tabelle dvr_template_* che servono come defaults quando si crea un nuovo DVR per un'agenzia nuova. Sorgente: -mat/07_DVR AGENZIE_TWISTER.xlsx Popola (su dev.db): - dvr_template (singleton: descrizione immobile, testo incendio, piano emergenze, ecc.) - dvr_template_pericoli (~108 righe: 1 per ogni punto compilabile, con i testi default di col C/D/E/F dal master) - dvr_template_valutazione (~30 righe: figure_esposte, rischi_descr, misure_prevenzione_override, p, g default) - dvr_template_incendio (~16 righe: stato di default per ciascun documento — "In corso di reperimento" / "Impianto non presente") - dvr_template_piano (vuoto: ogni sede aggiunge le sue azioni) I lookup_id (col_c_lookup_id, col_f_lookup_id, misure_prev_lookup_id) restano NULL: il testo del default va in *_override. Quando l'utente apre il DVR nuovo, vede subito il testo. Se vuole scegliere un altro preset, apre il dropdown ⌄ del componente EditableTesto. Esegui da app/: python3 ../import-tools/scripts/import_default_dvr.py """ from __future__ import annotations 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] DB = ROOT / "import-tools" / "db" / "dev.db" MASTER = ROOT / "-mat" / "07_DVR AGENZIE_TWISTER.xlsx" def norm(v) -> str | None: if v is None: return None s = re.sub(r"\s+", " ", str(v).strip()) return s or None def main(): if not MASTER.exists(): sys.exit(f"Master non trovato: {MASTER}") if not DB.exists(): sys.exit(f"DB non trovato: {DB}") wb = load_workbook(MASTER, data_only=True, read_only=False) conn = sqlite3.connect(DB) conn.execute("PRAGMA foreign_keys = ON") cur = conn.cursor() # Pulisci eventuali popolamenti precedenti cur.execute("DELETE FROM dvr_template_piano") cur.execute("DELETE FROM dvr_template_incendio") cur.execute("DELETE FROM dvr_template_valutazione") cur.execute("DELETE FROM dvr_template_pericoli") cur.execute("DELETE FROM dvr_template") # ---------- 1. dvr_template (singleton) ---------- ws_dati = wb["Dati"] descr = norm(ws_dati.cell(30, 1).value) ws_inc = wb["Incendio"] incendio_intro = "\n".join( norm(ws_inc.cell(r, 1).value) or "" for r in range(2, 10) if norm(ws_inc.cell(r, 1).value) ) or None cur.execute( """ INSERT INTO dvr_template ( id, descrizione_immobile, rischio_incendio_testo ) VALUES (1, ?, ?) """, (descr, incendio_intro), ) print(f"✓ dvr_template: descrizione_immobile {len(descr or '')} char, " f"rischio_incendio {len(incendio_intro or '')} char") # ---------- 2. dvr_template_pericoli ---------- # Mappa rif → punto_id dal lookup cur.execute("SELECT id, rif, livello FROM lookup_pericoli_punti") rif_to_id = {row[1]: (row[0], row[2]) for row in cur.fetchall()} ws_per = wb["Pericoli"] n_per = 0 for r in range(2, ws_per.max_row + 1): rif = norm(ws_per.cell(r, 1).value) if not rif or rif not in rif_to_id: continue punto_id, livello = rif_to_id[rif] if livello < 3: continue # solo punti compilabili col_c = norm(ws_per.cell(r, 3).value) col_d = norm(ws_per.cell(r, 4).value) col_e = norm(ws_per.cell(r, 5).value) col_f = norm(ws_per.cell(r, 6).value) # Normalizza stati U/P (col_d, col_e): tieni solo "S/X/-/N" col_d_norm = col_d if col_d in ("S", "X", "-", "N") else None col_e_norm = col_e if col_e in ("S", "X", "-", "N") else None cur.execute( """ INSERT INTO dvr_template_pericoli ( punto_id, col_c_override, col_d, col_e, col_f_override ) VALUES (?, ?, ?, ?, ?) """, (punto_id, col_c, col_d_norm, col_e_norm, col_f), ) n_per += 1 print(f"✓ dvr_template_pericoli: {n_per} righe") # ---------- 3. dvr_template_valutazione ---------- cur.execute("SELECT id, nome FROM lookup_valutazione_elementi") nome_to_elem = {norm(row[1]).lower(): row[0] for row in cur.fetchall() if norm(row[1])} ws_val = wb["Valutazione"] n_val = 0 # Header in righe 1-4, dati da riga 5 for r in range(5, ws_val.max_row + 1): elem = norm(ws_val.cell(r, 1).value) if not elem: continue elem_id = nome_to_elem.get(elem.lower()) if not elem_id: # match parziale for key, eid in nome_to_elem.items(): if elem.lower().startswith(key[:20]) or key.startswith(elem.lower()[:20]): elem_id = eid break if not elem_id: continue figure = norm(ws_val.cell(r, 2).value) rischi = norm(ws_val.cell(r, 3).value) misure = norm(ws_val.cell(r, 4).value) p_raw = ws_val.cell(r, 5).value g_raw = ws_val.cell(r, 6).value try: p = int(p_raw) if isinstance(p_raw, (int, float)) else None except (TypeError, ValueError): p = None try: g = int(g_raw) if isinstance(g_raw, (int, float)) else None except (TypeError, ValueError): g = None if p is not None and not (1 <= p <= 4): p = None if g is not None and not (1 <= g <= 4): g = None cur.execute( """ INSERT OR IGNORE INTO dvr_template_valutazione ( elemento_id, figure_esposte, rischi_descr, misure_prevenzione_override, p, g ) VALUES (?, ?, ?, ?, ?, ?) """, (elem_id, figure, rischi, misure, p, g), ) n_val += 1 print(f"✓ dvr_template_valutazione: {n_val} righe") # ---------- 4. dvr_template_incendio ---------- cur.execute("SELECT id, nome FROM lookup_documenti_incendio") doc_name_to_id = {norm(row[1]).lower(): row[0] for row in cur.fetchall() if norm(row[1])} cur.execute("SELECT id, nome FROM lookup_stati_documento_incendio") stato_name_to_id = {norm(row[1]).lower(): row[0] for row in cur.fetchall() if norm(row[1])} n_inc = 0 # Riga 11+ in foglio Incendio. Documento principale in col 1, sub in col 2 for r in range(11, ws_inc.max_row + 1): doc = norm(ws_inc.cell(r, 1).value) sub = norm(ws_inc.cell(r, 2).value) stato = norm(ws_inc.cell(r, 2).value if not sub or sub.lower() in stato_name_to_id else ws_inc.cell(r, 3).value) # Determina nome documento: se c'è sub, è il sub-documento if sub and sub.lower() in stato_name_to_id: # col 2 è uno stato, col 1 è il documento principale doc_name = doc stato_text = sub elif sub and sub.lower() not in stato_name_to_id and doc is None: # solo sub (è figlio di un parent precedente) doc_name = sub stato_text = norm(ws_inc.cell(r, 3).value) elif doc and not sub: doc_name = doc stato_text = None elif sub: doc_name = sub stato_text = norm(ws_inc.cell(r, 3).value) else: continue if not doc_name: continue doc_id = doc_name_to_id.get(doc_name.lower()) if not doc_id: # provo un match più largo for key, did in doc_name_to_id.items(): if doc_name.lower().startswith(key[:25]) or key.startswith(doc_name.lower()[:25]): doc_id = did break if not doc_id: continue stato_id = stato_name_to_id.get((stato_text or "").lower()) stato_override = None if stato_id else stato_text cur.execute( """ INSERT OR IGNORE INTO dvr_template_incendio ( documento_id, stato_lookup_id, stato_override ) VALUES (?, ?, ?) """, (doc_id, stato_id, stato_override), ) n_inc += 1 print(f"✓ dvr_template_incendio: {n_inc} righe") # ---------- 5. dvr_template_piano ---------- # Vuoto: ogni sede sceglie le sue azioni dal lookup_azioni_piano_miglioramento print("✓ dvr_template_piano: vuoto (le azioni sono per-sede)") conn.commit() conn.close() print("\nFatto. Per replicare su Turso:") print(" python3 import-tools/scripts/import_default_dvr.py --push") if __name__ == "__main__": main()