#!/usr/bin/env python3 """Backfill avanzato delle attività di miglioramento dai DVR storici. Strategia: 1. Per ogni `dvr_piano_miglioramento` con `attivita_lookup_id` valorizzato (= aggangiata alla famiglia v2 tramite codice variante): - Determina lo STATO dalla cella data: - contiene 'ESEGUIT' → risolta (data_risoluzione = data conclusione parsabile o data inizio o data del DVR) - data_conclusione futura → aperta (in corso) - data_conclusione passata → aperta (scaduta) - date assenti → aperta con r_score MEDIO di default - Crea record `dvr_attivita_miglioramento` con source_type='piano_storico' source_id=id del piano storico. 2. Cross-check coi pericoli problematici: per ogni `dvr_pericoli` con col_d/col_e in ('X','N') e codice attività del punto: - Se esiste già un'attività APERTA della stessa famiglia per quel DVR (creata in step 1) → linka source: aggiunge un riferimento (skip, evita doppio). - Altrimenti → crea nuova attività source_type='pericolo'. Idempotente: se rilanciato, salta i source_id già presenti. """ import re import sqlite3 from datetime import datetime, timedelta from pathlib import Path import os as _os DB = Path(_os.environ["IMPORT_DB"]) if _os.environ.get("IMPORT_DB") else Path(__file__).resolve().parents[1] / "db" / "dev.db" def famiglia_di(codice: str) -> str: head = codice.split("-")[0] m = re.match(r"^([A-Z]+)", head) return m.group(1) if m else head def is_eseguita(s: str | None) -> bool: return s is not None and "ESEGUIT" in s.upper() def parse_date(s: str | None) -> datetime | None: if not s: return None try: return datetime.strptime(s, "%Y-%m-%d") except ValueError: return None def stato_e_date(p_data_avvio, p_data_conclusione, data_emiss, oggi): """Ritorna (stato, data_inizio, data_scadenza, data_risoluzione, r_score, livello).""" eseguita = is_eseguita(p_data_avvio) or is_eseguita(p_data_conclusione) d_avvio = parse_date(p_data_avvio) d_concl = parse_date(p_data_conclusione) d_emiss = parse_date(data_emiss) or oggi # data_inizio: usa data avvio se parsabile, altrimenti data emissione data_inizio = d_avvio or d_emiss # data_scadenza: usa data conclusione se parsabile, altrimenti +6 mesi if d_concl: data_scadenza = d_concl else: data_scadenza = data_inizio + timedelta(days=180) # R_score / livello dipendono dal "tempo a disposizione": # se la finestra avvio→conclusione era >= 9 mesi → MOLTO_BASSO/BASSO (R=2) # se 3-9 mesi → MEDIO (R=4) # < 3 mesi → ALTO (R=8) if d_avvio and d_concl: finestra_gg = (d_concl - d_avvio).days if finestra_gg >= 270: r, livello = 2, "BASSO" elif finestra_gg >= 90: r, livello = 4, "MEDIO" else: r, livello = 8, "ALTO" else: # nessuna finestra parsabile → MEDIO di default r, livello = 4, "MEDIO" if eseguita: # Data risoluzione: data conclusione se parsabile, altrimenti data avvio, # altrimenti data emissione DVR data_ris = d_concl or d_avvio or d_emiss return ("risolta", data_inizio, data_scadenza, data_ris, r, livello) return ("aperta", data_inizio, data_scadenza, None, r, livello) def main(): conn = sqlite3.connect(DB) cur = conn.cursor() # Mappa: lookup_azioni_piano_miglioramento.id → famiglia_id cur.execute( "SELECT id, famiglia_id FROM lookup_azioni_piano_miglioramento WHERE famiglia_id IS NOT NULL" ) fam_da_lookup = dict(cur.fetchall()) # Mappa: famiglia_codice → famiglia_id cur.execute("SELECT codice, id FROM lookup_attivita_famiglie") famiglie = dict(cur.fetchall()) # Già backfilled? cur.execute( "SELECT source_id FROM dvr_attivita_miglioramento WHERE source_type='piano_storico'" ) gia_piano = {r[0] for r in cur.fetchall()} cur.execute( "SELECT source_id FROM dvr_attivita_miglioramento WHERE source_type='pericolo'" ) gia_pericolo = {r[0] for r in cur.fetchall()} print(f"Già importati: {len(gia_piano)} (piano_storico), {len(gia_pericolo)} (pericolo)") oggi = datetime.now() # ───────────────────────────────────────────────────────────────────── # STEP 1: import piano storico → attività # ───────────────────────────────────────────────────────────────────── cur.execute( """SELECT p.id, p.dvr_id, p.attivita_lookup_id, p.data_avvio, p.data_conclusione, d.data_emissione FROM dvr_piano_miglioramento p JOIN dvr d ON d.id = p.dvr_id WHERE p.attivita_lookup_id IS NOT NULL""" ) rows = cur.fetchall() print(f"\n[1] Piano storico → attività: {len(rows)} righe candidate") creati_storico = 0 skipped = 0 for pid, dvr_id, lookup_id, davvio, dconcl, demiss in rows: if pid in gia_piano: skipped += 1 continue fam_id = fam_da_lookup.get(lookup_id) if not fam_id: continue stato, d_ini, d_sca, d_ris, r, livello = stato_e_date(davvio, dconcl, demiss, oggi) cur.execute( """INSERT INTO dvr_attivita_miglioramento (dvr_id, famiglia_id, source_type, source_id, r_score, livello, data_inizio, data_scadenza, stato, data_risoluzione, variante_scelta_id) VALUES (?, ?, 'piano_storico', ?, ?, ?, ?, ?, ?, ?, ?)""", ( dvr_id, fam_id, pid, r, livello, d_ini.date().isoformat(), d_sca.date().isoformat(), stato, d_ris.date().isoformat() if d_ris else None, lookup_id, # variante scelta dal piano storico ), ) creati_storico += 1 print(f" Creati: {creati_storico} (skip già fatti: {skipped})") # ───────────────────────────────────────────────────────────────────── # STEP 2: cross-check coi pericoli problematici # Per ogni pericolo X/N, se NON esiste già un'attività APERTA della stessa # famiglia per quel DVR → la creo (source_type='pericolo'). # Se esiste già un'attività aperta (anche da piano_storico) della stessa # famiglia → considero coperto, salto. # ───────────────────────────────────────────────────────────────────── cur.execute( """SELECT p.id, p.dvr_id, p.col_d, p.col_e, lp.codice_attivita, d.data_emissione FROM dvr_pericoli p JOIN lookup_pericoli_punti lp ON lp.id = p.punto_id JOIN dvr d ON d.id = p.dvr_id WHERE lp.codice_attivita IS NOT NULL AND (p.col_d IN ('X','N') OR p.col_e IN ('X','N'))""" ) rows = cur.fetchall() print(f"\n[2] Cross-check pericoli problematici: {len(rows)} righe") creati_pericolo = 0 coperti = 0 skipped2 = 0 for pid, dvr_id, cd, ce, codice, demiss in rows: if pid in gia_pericolo: skipped2 += 1 continue fam_cod = famiglia_di(codice) fam_id = famiglie.get(fam_cod) if not fam_id: continue # Esiste già un'attività APERTA della stessa famiglia per questo DVR? cur.execute( """SELECT id FROM dvr_attivita_miglioramento WHERE dvr_id = ? AND famiglia_id = ? AND stato = 'aperta'""", (dvr_id, fam_id), ) if cur.fetchone(): coperti += 1 continue # Livello: X=MEDIO, N=ALTO; max tra U e P def score(v): if v == "N": return 8 if v == "X": return 4 return 0 r = max(score(cd), score(ce)) if r == 0: continue livello = "ALTO" if r == 8 else "MEDIO" d_emiss = parse_date(demiss) or oggi scad_giorni = 0 if r == 8 else 180 d_scad = d_emiss + timedelta(days=scad_giorni) cur.execute( """INSERT INTO dvr_attivita_miglioramento (dvr_id, famiglia_id, source_type, source_id, r_score, livello, data_inizio, data_scadenza, stato) VALUES (?, ?, 'pericolo', ?, ?, ?, ?, ?, 'aperta')""", (dvr_id, fam_id, pid, r, livello, d_emiss.date().isoformat(), d_scad.date().isoformat()), ) creati_pericolo += 1 print(f" Creati: {creati_pericolo} (coperti da piano_storico: {coperti}, skip già fatti: {skipped2})") conn.commit() # Summary finale print("\n=== Riepilogo finale ===") cur.execute( "SELECT source_type, stato, COUNT(*) FROM dvr_attivita_miglioramento GROUP BY source_type, stato ORDER BY source_type, stato" ) for r in cur.fetchall(): print(f" {r[0]:18s} {r[1]:10s} {r[2]}") cur.execute("SELECT COUNT(*), COUNT(DISTINCT dvr_id) FROM dvr_attivita_miglioramento") tot, dvr_count = cur.fetchone() print(f"\n Totali: {tot} attività su {dvr_count} DVR") conn.close() if __name__ == "__main__": main()