#!/usr/bin/env python3 """ Rilegge dai file sorgente i dati emersi dal tracker difformità del cliente (settembre 2026) e li scrive in un JSON per app/fix-tracker-settembre.mjs. Legge le celle Word con row._tr.tc_lst (mai row.cells). Non tocca il DB. python3 import-tools/scripts/estrai_fix_tracker.py > /tmp/fix_tracker.json """ import sys, os, re, json, sqlite3 from datetime import datetime from docx import Document import openpyxl DVR_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "-mat", "DVR") db = sqlite3.connect(sys.argv[1]) def norm(s): s = re.sub(r"[ \t\r\n]+", " ", str(s or "")).strip() return s or None def key(s): return re.sub(r"[^a-z0-9]", "", (s or "").lower()) def to_int(v): try: n = int(norm(v)); return n if 0 <= n <= 9 else None except (TypeError, ValueError): return None def celltext(tc): return norm("".join(x.text for x in tc.iter() if x.tag.endswith("}t"))) def tc_rows(doc): for t in doc.tables: for row in t.rows: yield [celltext(tc) for tc in row._tr.tc_lst] def open_docx(sf): try: return Document(os.path.join(DVR_DIR, sf)) except Exception: return None out = {"piano": [], "valutazione_p": [], "valutazione_nuove": [], "valutazione_pg_vuoti": [], "piano_miglioramento": [], "luogo_sicuro": []} # ── 1. Piano (anagrafica) dai Word: riga ['piano', 'primo piano'] della tabella Dati def piano_norm(v): t = re.sub(r"\(.*?\)", "", (v or "").lower()).replace("piamo", "piano") ord_ = {"terra": "Terra", "rialzato": "Rialzato", "primo": "1", "secondo": "2", "terzo": "3", "quarto": "4", "quinto": "5", "sesto": "6", "settimo": "7", "ottavo": "8", "1": "1", "s1": "S1"} found = [] for w in re.findall(r"[a-z0-9]+", t): if w in ord_ and ord_[w] not in found: found.append(ord_[w]) order = ["S1", "Terra", "Rialzato", "1", "2", "3", "4", "5", "6", "7", "8"] return ", ".join(sorted(found, key=order.index)) or None for aid, sf in db.execute("""SELECT a.id, d.source_file FROM agenzie a JOIN dvr d ON d.agenzia_id = COALESCE(a.dvr_condiviso_da_id, a.id) WHERE a.deleted_at IS NULL AND a.piano IS NULL AND d.source_format = 'docx'"""): doc = open_docx(sf) if not doc: continue val = None for t in doc.tables[:5]: for row in t.rows: cells = [celltext(tc) for tc in row._tr.tc_lst] if len(cells) >= 2 and (cells[-2] or "").lower() == "piano": val = cells[-1] p = piano_norm(val) if p: out["piano"].append({"agenzia_id": aid, "piano": p, "sorgente": val}) # ── 2/3. Valutazione dai Word: P copiata da G (fix di agosto con row.cells), elementi mancanti, P/G vuoti elem = {key(n): i for i, n in db.execute("SELECT id, nome FROM lookup_valutazione_elementi")} for did, sf in db.execute("SELECT id, source_file FROM dvr WHERE source_format = 'docx'"): doc = open_docx(sf) if not doc: continue dbv = {e: (p, g) for e, p, g in db.execute("SELECT elemento_id, p, g FROM dvr_valutazione WHERE dvr_id = ?", (did,))} seen = set() for cells in tc_rows(doc): if not cells or not cells[0]: continue eid = elem.get(key(cells[0])) if eid is None or eid in seen or len(cells) < 7: continue seen.add(eid) p, g = to_int(cells[4]), to_int(cells[5]) if p is None or g is None or not (0 <= p <= 4 and 1 <= g <= 4): continue cur = dbv.get(eid) rec = {"dvr_id": did, "elemento_id": eid, "p": p, "g": g} if cur is None: out["valutazione_nuove"].append({**rec, "figure_esposte": cells[1], "rischi_descr": cells[2], "misure_prevenzione_override": cells[3]}) elif cur == (None, None): out["valutazione_pg_vuoti"].append(rec) elif cur == (g, g) and p != g: out["valutazione_p"].append(rec) # ── 3b. Rumore e Campi elettromagnetici di Sant'Agata (Excel): P/G vuoti a database, 1-1 nel file for did, sf in db.execute("SELECT id, source_file FROM dvr WHERE source_file LIKE 'SORRENTO_SANT''AGATA%'"): ws = openpyxl.load_workbook(os.path.join(DVR_DIR, sf), read_only=True, data_only=True)["Valutazione"] dbv = {e: (p, g) for e, p, g in db.execute("SELECT elemento_id, p, g FROM dvr_valutazione WHERE dvr_id = ?", (did,))} for row in ws.iter_rows(values_only=True): eid = elem.get(key(str(row[0] or ""))) p, g = to_int(row[4]), to_int(row[5]) if eid and dbv.get(eid) == (None, None) and p is not None and g: out["valutazione_pg_vuoti"].append({"dvr_id": did, "elemento_id": eid, "p": p, "g": g}) # ── 4. Piano di miglioramento di Cagliari Centro e Camposampiero: tabella che parte prima della riga 33 for did, sf in db.execute("""SELECT id, source_file FROM dvr d WHERE source_format = 'xlsx' AND NOT EXISTS (SELECT 1 FROM dvr_piano_miglioramento p WHERE p.dvr_id = d.id)"""): ws = openpyxl.load_workbook(os.path.join(DVR_DIR, sf), read_only=True, data_only=True)["Rischi"] start = next((r for r in range(1, 60) if str(ws.cell(r, 1).value or "").strip().upper().startswith("ATTIVITA")), None) if not start: continue last = None; ordine = 0 for r in range(start + 1, ws.max_row + 1): a, b, c = norm(ws.cell(r, 1).value), norm(ws.cell(r, 2).value), ws.cell(r, 3).value d = c.date().isoformat() if isinstance(c, datetime) else norm(c) if a: ordine += 1 last = {"dvr_id": did, "attivita_override": a, "responsabile": b or "Operation di rete - Alleanza Assicurazioni", "data_avvio": d, "data_conclusione": None, "ordine": ordine} out["piano_miglioramento"].append(last) elif last and d: last["data_conclusione"] = d # ── 5. Didascalia del luogo sicuro troncata dopo il trattino (Word) for did, sf, descr in db.execute("SELECT id, source_file, luogo_sicuro_esterno_descr FROM dvr WHERE source_format = 'docx' AND luogo_sicuro_esterno_descr IS NOT NULL"): doc = open_docx(sf) if not doc: continue for el in doc.element.body.iter(): if el.tag.endswith("}p"): t = norm("".join(x.text for x in el.iter() if x.tag.endswith("}t"))) if t and t != descr and t.endswith(descr) and re.search(r"[-–] " + re.escape(descr) + "$", t): out["luogo_sicuro"].append({"dvr_id": did, "descr": t}); break json.dump(out, sys.stdout, ensure_ascii=False, indent=1) sys.stderr.write(" ".join(f"{k}={len(v)}" for k, v in out.items()) + "\n")