""" gemma_unified_builder.py ======================== Costruisce/aggiorna gemma_unified.db — DB SQLite che fonde: - gemma.parquet (sorgente Gemma) - LINKER_VALUTAZIONE in linker.db (sinottico utente) Usato da: - app/main.py → build in background all'avvio dell'app - database_logic.py → build lazy al primo accesso MediaTrack - linker_db.py → invalida il file dopo salva_valutazioni() Struttura di gemma_unified.db: - tabella `gemma` : copia 1:1 di gemma.parquet - tabella `valutazioni_unified`: una riga per IMDB, con GEM_SIN='G'|'S' """ import threading from collections import defaultdict from pathlib import Path try: from loguru import logger except ImportError: import logging logger = logging.getLogger(__name__) # Cache in-process: evita rigenerazioni ripetute nello stesso processo _built_mtime: float = 0.0 _build_lock = threading.Lock() # un solo build alla volta per processo def _to_iso(date_str: str) -> str: """Normalizza data a YYYY-MM-DD per confronto. Gestisce DD/MM/YYYY e YYYY-MM-DD.""" d = (date_str or '').strip() if len(d) == 10 and d[2] == '/' and d[5] == '/': return f"{d[6:10]}-{d[3:5]}-{d[0:2]}" return d def _build_valutazioni_unified(gemma_db: Path, linker_db: Path) -> None: """ Crea la tabella valutazioni_unified dentro gemma_db. Per ogni IMDB: coalesce i campi non-NULL tra tutte le righe Gemma (ordinate per A_DATA DESC), poi confronta con la riga sinottica più recente in LINKER_VALUTAZIONE. La sinottica vince solo se la sua DATA è più recente di A_DATA Gemma per lo stesso IMDB. """ import sqlite3 as _sl con = _sl.connect(str(gemma_db)) try: # --- Gemma: coalesce campi non-NULL tra tutte le righe per IMDB --- all_gemma = con.execute( "SELECT A_COD_IMDB, A_DISTRIBUTORE, V_RDA, V_C5, V_I1, V_R4, V_LA5, V_I2, " "V_IRIS, V_TOP, V_FOC, V_C20, V_CI34, V_C27, A_SUPPORTO, A_DATA FROM gemma" ).fetchall() imdb_groups: dict = defaultdict(list) for row in all_gemma: if row[0]: imdb_groups[row[0]].append(row) gemma_best = {} for imdb, rows in imdb_groups.items(): rows_sorted = sorted(rows, key=lambda r: _to_iso(r[15]), reverse=True) merged = list(rows_sorted[0]) for row in rows_sorted[1:]: for i in range(1, 16): if (merged[i] is None or merged[i] == '') and row[i] not in (None, ''): merged[i] = row[i] gemma_best[imdb] = tuple(merged) # --- Sinottico: best row per IMDB da LINKER_VALUTAZIONE --- sinot_best = {} if linker_db and linker_db.exists(): lcon = _sl.connect(str(linker_db)) try: sinot_rows = lcon.execute( "SELECT COD_IMDB, FORNITORE, DATA, C5, I1, R4, LA5, I2, IRIS, " "TOPCRIME, FOCUS, C20, CINE34, TWENTYSEVEN " "FROM LINKER_VALUTAZIONE WHERE COD_IMDB IS NOT NULL" ).fetchall() except Exception: sinot_rows = [] finally: lcon.close() for row in sinot_rows: imdb, data = row[0], row[2] or '' if imdb not in sinot_best or data > (sinot_best[imdb][2] or ''): sinot_best[imdb] = row # --- Crea tabella e popola --- con.execute("DROP TABLE IF EXISTS valutazioni_unified") con.execute(""" CREATE TABLE valutazioni_unified ( A_COD_IMDB TEXT, A_DISTRIBUTORE TEXT, V_RDA TEXT, V_C5 TEXT, V_I1 TEXT, V_R4 TEXT, V_LA5 TEXT, V_I2 TEXT, V_IRIS TEXT, V_TOP TEXT, V_FOC TEXT, V_C20 TEXT, V_CI34 TEXT, V_C27 TEXT, A_SUPPORTO TEXT, A_DATA TEXT, GEM_SIN TEXT ) """) all_imbds = set(gemma_best) | set(sinot_best) rows_s, rows_g = [], [] for imdb in all_imbds: g = gemma_best.get(imdb) s = sinot_best.get(imdb) s_date = _to_iso(s[2]) if s else '' g_date = _to_iso(g[15]) if g else '' sinot_wins = s is not None and (g is None or s_date >= g_date) if sinot_wins: rows_s.append((imdb, s[1], None, s[3], s[4], s[5], s[6], s[7], s[8], s[9], s[10], s[11], s[12], s[13], None, s[2])) elif g is not None: rows_g.append((g[0], g[1], g[2], g[3], g[4], g[5], g[6], g[7], g[8], g[9], g[10], g[11], g[12], g[13], g[14], g[15])) con.execute("BEGIN") con.executemany("INSERT INTO valutazioni_unified VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'S')", rows_s) con.executemany("INSERT INTO valutazioni_unified VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'G')", rows_g) con.execute("COMMIT") finally: con.close() def ensure_gemma_unified(parquet_dir: Path, linker_db: Path = None) -> None: """ Assicura che gemma_unified.db sia aggiornato rispetto a gemma.parquet e LINKER_VALUTAZIONE. Operazione idempotente: non fa nulla se il file è già aggiornato. Chiamabile da qualsiasi modulo. """ global _built_mtime gemma_parquet = parquet_dir / 'gemma.parquet' gemma_unified_db = parquet_dir / 'gemma_unified.db' if not gemma_parquet.exists(): return parquet_mtime = gemma_parquet.stat().st_mtime linker_mtime = linker_db.stat().st_mtime if linker_db and linker_db.exists() else 0.0 src_mtime = max(parquet_mtime, linker_mtime) if _built_mtime == src_mtime and gemma_unified_db.exists() and gemma_unified_db.stat().st_size > 0: return if gemma_unified_db.exists() and gemma_unified_db.stat().st_size > 0 and gemma_unified_db.stat().st_mtime >= src_mtime: _built_mtime = src_mtime return with _build_lock: # Ri-controlla dopo aver acquisito il lock — un altro thread potrebbe aver già buildato if _built_mtime == src_mtime and gemma_unified_db.exists() and gemma_unified_db.stat().st_size > 0: return if gemma_unified_db.exists() and gemma_unified_db.stat().st_size > 0 and gemma_unified_db.stat().st_mtime >= src_mtime: _built_mtime = src_mtime return # Pulizia file .tmp orfani di build precedenti interrotte for _stale in parquet_dir.glob('gemma_unified_*.tmp'): try: _stale.unlink() except Exception: pass logger.info("Rigenerazione gemma_unified.db...") import tempfile as _tf, os as _os, time as _time # mkstemp garantisce unicità anche tra thread concorrenti dello stesso processo Flask _fd, _tmp_str = _tf.mkstemp(dir=gemma_unified_db.parent, suffix='.tmp', prefix='gemma_unified_') _os.close(_fd) tmp = Path(_tmp_str) try: import duckdb as _ddb import sqlite3 as _sl2 _t0 = _time.monotonic() # DuckDB legge il parquet nativamente (no pyarrow di sistema richiesto) _duck = _ddb.connect() df = _duck.execute(f"SELECT * FROM read_parquet('{gemma_parquet}')").df() _duck.close() logger.debug(f"gemma_unified: parquet letto ({len(df)} righe) in {_time.monotonic()-_t0:.1f}s") _t1 = _time.monotonic() _tcon = _sl2.connect(str(tmp)) try: _tcon.execute("PRAGMA journal_mode=OFF") _tcon.execute("PRAGMA synchronous=OFF") col_defs = ', '.join(f'"{c}" TEXT' for c in df.columns) _tcon.execute("DROP TABLE IF EXISTS gemma") _tcon.execute(f"CREATE TABLE gemma ({col_defs})") placeholders = ', '.join('?' * len(df.columns)) _tcon.execute("BEGIN") _tcon.executemany( f"INSERT INTO gemma VALUES ({placeholders})", df.where(df.notna(), other=None).values.tolist() ) _tcon.execute("COMMIT") finally: _tcon.close() logger.debug(f"gemma_unified: write SQLite completato in {_time.monotonic()-_t1:.1f}s") _build_valutazioni_unified(tmp, linker_db) # Sostituisce atomicamente — nessun unlink preventivo (evita finestra dove il file è assente) tmp.replace(gemma_unified_db) _built_mtime = src_mtime logger.info(f"gemma_unified.db pronto in {_time.monotonic()-_t0:.1f}s totali.") except Exception as e: try: tmp.unlink(missing_ok=True) except Exception: pass logger.error(f"Errore build gemma_unified.db: {e}")