import json import uuid from collections import OrderedDict from pathlib import Path import duckdb import webview from flask import Flask, jsonify, request, send_from_directory _PARQUET_DIR = ( Path(__file__).resolve().parent.parent.parent.parent / "PYTHON" / "MyICR_Suite" / "local_db" / "parquet" ) PARQUET_PATH = _PARQUET_DIR / "prodotti.parquet" EMESSO_PARQUET_PATH = _PARQUET_DIR / "emesso.parquet" BOXOFFICE_PARQUET_PATH = _PARQUET_DIR / "boxoffice.parquet" DIRITTI_PARQUET_PATH = _PARQUET_DIR / "diritti.parquet" # Tabella "sidecar" 1:1 su id_diritto (verificato: 51.402 righe = 51.402, # nessuna orfana in nessuna delle due direzioni) — esposta come se fosse # parte di diritti.parquet stesso (vedi _build_diritti_join), non come # blocco/tabella separata: due campi in piu' nello stesso "Tabella Diritti". DIRITTI_ENRICH_PARQUET_PATH = _PARQUET_DIR / "diritti_enrich.parquet" GEMMA_PARQUET_PATH = _PARQUET_DIR / "gemma.parquet" # Tabella ponte codice interno <-> id IMDb, usata SOLO per risolvere il join # Gemma->Anagrafica (Gemma non ha "prodotto"/"edizione" propri, solo un # riferimento IMDb — vedi _build_gemma_join). Non e' un dominio/blocco a se'. IMDB_PARQUET_PATH = _PARQUET_DIR / "imdb.parquet" CAST_PARQUET_PATH = _PARQUET_DIR / "cast.parquet" HOST = "127.0.0.1" PORT = 5050 # Componente griglia riusabile, indipendente da PowerBricks (sandbox/datagrid/) # — servito qui via una seconda static route, non copiato/duplicato. DATAGRID_SRC_DIR = Path(__file__).resolve().parent.parent.parent / "datagrid" / "src" # Risultati query in attesa di essere mostrati nella finestra griglia (vedi # /api/open_grid) — in memoria, nessuna persistenza: un token vale finche' il # processo resta vivo. Cap a 20 per evitare crescita illimitata se si aprono # molte griglie in una sessione lunga (OrderedDict + pop del piu' vecchio). _RESULTS_STORE = OrderedDict() _RESULTS_STORE_MAX = 20 def _store_results(columns, rows): token = uuid.uuid4().hex _RESULTS_STORE[token] = {"columns": columns, "rows": rows} while len(_RESULTS_STORE) > _RESULTS_STORE_MAX: _RESULTS_STORE.popitem(last=False) return token # Tutti i campi di prodotti.parquet — vedi CLAUDE.md FIELDS = { "prodotto": "int", "edizione": "int", "program_id": "string", "titolo_italiano": "string", "titolo_originale": "string", "descr_edizione": "string", "unico_seriale": "string", "superserie_descr": "string", "superserie_id": "int", "superserie_extkey": "string", "stagione": "int", "durata": "int", "cromia": "string", "veg": "string", "num_episodi": "int", "paesi_produzione1": "string", "paesi_produzione2": "string", "paesi_produzione3": "string", "anno_produzione": "int", "genere1": "string", "genere2": "string", "genere3": "string", "provenienza": "string", "vm": "string", "tipologia": "string", # Campi derivati (non in prodotti.parquet), aggiunti dentro la subquery di # base Anagrafica (vedi _ANAGRAFICA_BASE_SQL) — stessa logica di # v_custom_anagr in Linker (PYTHON/MyICR_Suite/app/modules/linker/src/ # database/parquet_db.py). "regista"/"attori" sono gia' un riassunto # pre-aggregato (rispettivamente: primo FRE per progr_cast, primi 3 C001 # concatenati) su cast.parquet, che ha una relazione 1:molti pesante con # "prodotto" (fino a 1.369 righe) — esporlo come tabella satellite a se' # avrebbe richiesto la stessa complessita' di aggregazione di Emesso/ # Diritti senza un bisogno reale oggi. Idea per il futuro (Mauro, # 07/08/2026): un domani questi due campi potrebbero fare "drill" verso # il dettaglio cast grezzo filtrato per ruolo — non implementato, solo # annotato qui perche' non vada perso. "imdb_codice": "string", "regista": "string", "attori": "string", } # Label leggibili stile Linker (column_defs.py) per i campi dove la # corrispondenza e' certa — mostrate nelle tendine/griglia al posto del nome # colonna grezzo. Solo un sottoinsieme: dove il mapping non e' inequivocabile # (es. valutazioni Gemma) si preferisce il nome grezzo a un'etichetta # indovinata male. Fallback sempre al nome campo se assente da qui. ANAGRAFICA_LABELS = { "prodotto": "RIFER", "tipologia": "TIPOL", "titolo_italiano": "TI", "titolo_originale": "TO", "anno_produzione": "ANNO", "paesi_produzione1": "PAESE", "genere1": "GENERE", "num_episodi": "EPIS", "durata": "DURATA", "superserie_descr": "SUPERSERIE", "veg": "VEG", "imdb_codice": "IMDB", "regista": "REGISTA", "attori": "ATTORI", } # Larghezze fisse stile Linker (column_defs.py) — stesso principio delle # label sopra: solo dove la corrispondenza campo e' certa, fallback a una # larghezza di default (vedi results.html) altrimenti. Colonne dinamiche # (dimensionate sul contenuto) erano il comportamento di default del # componente datagrid — cambiato su richiesta di Mauro (07/08/2026). # I valori px di Linker (RIFER/IMDB/ANNO/EPIS/VEG) sono tarati sul font Qt # Calibri 8, piu' compatto del rendering browser — con l'header sort arrow # di Tabulator il testo troncava. Allargati leggermente (non i valori Linker # esatti) solo per queste 5, segnalato da Mauro (07/08/2026). ANAGRAFICA_WIDTHS = { "prodotto": 75, "imdb_codice": 75, "tipologia": 75, "titolo_italiano": 200, "titolo_originale": 200, "anno_produzione": 75, "paesi_produzione1": 61, "genere1": 100, "num_episodi": 70, "durata": 60, "superserie_descr": 110, "veg": 65, "regista": 131, "attori": 131, } # Campi esposti di emesso.parquet — sola lettura, join su prodotto+edizione. # prodotto/edizione/ncparpgm esclusi: chiavi di join / codice interno, non # utili in output. Filtrabili anche loro (vedi _build_filtro generico sotto). EMESSO_FIELDS = { "rete": "string", "data_emissione": "date", "ora_inizio": "int", "ora_fine": "int", "durata_netta": "int", "prima_visione": "string", "episodio": "int", "audience": "int", "share": "float", "durata_lorda": "int", "tipologia": "string", "fascia": "string", } # Label leggibili stile Linker (EM_* in column_defs.py) — sottoinsieme certo. EMESSO_LABELS = { "rete": "RETE", "data_emissione": "DATA", "ora_inizio": "INIZIO", "fascia": "FASCIA", "audience": "AUD", "share": "SHA", } # Larghezze fisse stile Linker (EM_* in column_defs.py). EMESSO_WIDTHS = { "rete": 45, "data_emissione": 68, "ora_inizio": 55, "fascia": 47, "audience": 55, "share": 50, } # Campi esposti di boxoffice.parquet — sola lettura, join su "prodotto" (unica # chiave, relazione 1:1 a differenza di Emesso: un prodotto ha al massimo un # debutto in sala). "prodotto" escluso: chiave di join, non utile in output. # "data_debutto" e' DATE nel parquet, castato a VARCHAR ISO (cast in # _build_boxoffice_join) — necessario per la serializzazione JSON (jsonify # non serializza date.date). Tipo "date" (non "string"): abilita i confronti # (gt/lt/gte/lte/between), non solo equals/contains — vedi OPS_DATE. BOXOFFICE_FIELDS = { "stagione": "string", "data_debutto": "date", "distributore": "string", "incasso": "float", "spettatori": "int", } # Label leggibili stile Linker (BO_* in column_defs.py). BOXOFFICE_LABELS = { "data_debutto": "DEBUTTO", "incasso": "INCASSO", "spettatori": "SPETTATORI", "distributore": "DIST. CINEMA", } # Larghezze fisse stile Linker (BO_* in column_defs.py). BOXOFFICE_WIDTHS = { "data_debutto": 68, "incasso": 61, "spettatori": 61, "distributore": 120, } # Campi esposti di diritti.parquet — sola lettura, join su prod+ediz (diritti) # = prodotto+edizione (Anagrafica), relazione 1:molti come Emesso (un # prodotto/edizione puo' avere piu' contratti/righe diritto nel tempo). # "id_diritto"/"prod"/"ediz" esclusi: chiave tecnica/chiavi di join. Le # colonne CONTRAENTE_WIN_*/DECR_WIN_*/SCAD_WIN_*/CAUSALE_WIN_*/NOTE_WIN_* # (9 finestre storiche per riga) restano fuori da questo primo giro: sono # proprio il materiale della logica derivata (first-run/re-run, inibizioni) # che la specifica ufficiale segnala come layer di business a se', # esplicitamente rimandato (vedi PowerBricks.md) — qui si espongono solo i # campi "grezzi" del contratto, stesso approccio minimale con cui e' partito # Emesso prima di aggiungere filtri. decr/scad/decr_inib/scad_inib sono DATE # nel parquet, cast a VARCHAR nella subquery di join (vedi # _build_diritti_join) per coerenza con "data_debutto" di Boxoffice. DIRITTI_FIELDS = { "tipologia": "string", "d_t": "string", "ragsoc_distr": "string", "perc": "float", "pass_cons": "float", "pass_cons_tot": "float", "pass_eff": "float", "pass_eff_tot": "float", "decr": "date", "scad": "date", "causale": "string", "flag_inib": "string", "decr_inib": "date", "scad_inib": "date", "contratto": "int", "riga": "int", "situazione": "int", "fr_rr": "string", # Da diritti_enrich.parquet (tabella sidecar 1:1 su id_diritto, non un # dominio a se' — vedi DIRITTI_ENRICH_PARQUET_PATH e _build_diritti_join). # ESCLUSIVO_TEM e' BOOLEAN nel parquet, esposto come stringa # ("true"/"false") per restare nel sistema di filtri generico esistente # senza aggiungere un tipo "bool" solo per un campo. "fornitore_cluster": "string", "esclusivo_tem": "string", } # Label leggibili stile Linker (DIRITTI in column_defs.py) — sottoinsieme # certo: 'd_t', 'contratto', 'riga', 'situazione', 'fr_rr' ecc. non hanno un # corrispondente inequivocabile nel dizionario Linker, restano col nome # grezzo. DIRITTI_LABELS = { "ragsoc_distr": "DISTRIBUTORE", "perc": "%", "pass_cons": "P.CONS.", "pass_eff": "P.EFF.", "decr": "DECOR.", "scad": "SCAD.", "causale": "CAUS.", } # Larghezze fisse stile Linker (DIRITTI in column_defs.py) — stesso sottoinsieme # certo di DIRITTI_LABELS sopra. DIRITTI_WIDTHS = { "ragsoc_distr": 120, "perc": 40, "pass_cons": 40, "pass_eff": 40, "decr": 68, "scad": 68, "causale": 40, } # Gemma non ha un mapping label — le valutazioni VAL_* di Linker (column_defs.py) # non hanno una corrispondenza certa coi campi v_*/a_* esposti qui senza # verificare il codice che le costruisce (fuori scope oggi, non fatto): # meglio nessuna label che una indovinata male. Stesso principio per le # larghezze: nessuna qui, fallback al default generico in results.html. GEMMA_LABELS = {} GEMMA_WIDTHS = {} # Campi esposti di gemma.parquet — dataset di tracking acquisizioni/screener # (non da OnAir come gli altri, fonte diversa), livello "pre-prodotto": la # maggior parte delle righe non e' ancora (o non sara' mai) un prodotto # Mediaset anagrafato. **Join su codice IMDb** (richiesto esplicitamente da # Mauro, non su A_CODICE_PRODOTTO): gemma non ha "prodotto"/"edizione" # propri, solo A_COD_IMDB — risolto in Anagrafica tramite imdb.parquet # (A_COD_IMDB -> riferimento_imdb -> codice = prodotti.prodotto), vedi # _build_gemma_join. "A_COD_IMDB" escluso da qui: e' la chiave di join, non # un campo di output (A_CODICE_PRODOTTO invece resta, e' solo informativo # ora che non e' piu' la chiave). Nomi campo tutti minuscolo per coerenza # col resto (colonne reali gemma.parquet sono MAIUSCOLE, es. "A_STATO" -> # "a_stato" — la corrispondenza e' sempre .upper(), vedi _build_gemma_join). # "a_data"/"a_deadline" sono date in formato italiano GG/MM/AAAA **come # stringa grezza nel parquet** (non un tipo DATE nativo come decr/scad di # Diritti) — convertite a ISO nella subquery di join per usare lo stesso # tipo "date" del resto del sistema. GEMMA_FIELDS = { "a_stato": "string", "a_id": "string", "a_titolo": "string", "a_data": "date", "a_distributore": "string", "a_tipologia": "string", "a_tipo": "string", "a_autore_regista": "string", "a_cast": "string", "a_codice_prodotto": "string", "a_episodio": "string", "a_supporto": "string", "a_note_supporto": "string", "a_deadline": "date", "a_note_pubbliche": "string", "a_note_private": "string", "a_acquistato_venduto": "string", "a_c5": "string", "a_i1": "string", "a_r4": "string", "a_la5": "string", "a_i2": "string", "a_iris": "string", "a_top": "string", "a_foc": "string", "a_c20": "string", "a_ci34": "string", "a_c27": "string", "a_cin": "string", "a_inf": "string", "a_emo": "string", "a_ene": "string", "a_com": "string", "a_sto": "string", "a_cri": "string", "a_act": "string", "v_stato": "string", "v_rda": "string", "v_c5": "string", "v_i1": "string", "v_r4": "string", "v_la5": "string", "v_i2": "string", "v_iris": "string", "v_top": "string", "v_foc": "string", "v_c20": "string", "v_ci34": "string", "v_c27": "string", "v_cin": "string", "v_inf": "string", "v_emo": "string", "v_ene": "string", "v_com": "string", "v_sto": "string", "v_cri": "string", "v_act": "string", } OPS_STRING = { "equals": "=", "not_equals": "<>", "contains": "ILIKE", "is_null": "IS NULL", "is_not_null": "IS NOT NULL", } OPS_INT = { "equals": "=", "not_equals": "<>", "gt": ">", "lt": "<", "gte": ">=", "lte": "<=", "between": "BETWEEN", "is_null": "IS NULL", "is_not_null": "IS NOT NULL", } # Campi "date" (data_emissione/data_debutto/decr/scad/...): tutti VARCHAR in # formato ISO (YYYY-MM-DD) nel punto in cui arrivano a _build_filtro — cast # gia' fatto a monte (vedi _build_emesso_join/_build_boxoffice_join/ # _build_diritti_join). Uniscono gli operatori di confronto (utili per # "dopo il...", range) a "contiene" (utile per "contiene 2026" senza dover # scrivere una data completa) — unico tipo con l'unione di entrambi i set. OPS_DATE = {**OPS_STRING, **OPS_INT} def _parse_date_value(value): """Converte una data in formato italiano GG/MM/AAAA (quello che l'utente scrive nel blocco) nel formato ISO YYYY-MM-DD della colonna gia' castata a VARCHAR — necessario perche' il confronto (>, <, BETWEEN...) e' lessicografico sulla stringa: funziona correttamente solo se entrambi i lati sono nello stesso formato ISO (l'ordine lessicografico coincide con l'ordine cronologico solo in quel formato).""" parts = str(value).strip().split("/") if len(parts) != 3: raise ValueError(f"data non valida: {value!r} (atteso GG/MM/AAAA)") day, month, year = parts return f"{int(year):04d}-{int(month):02d}-{int(day):02d}" app = Flask(__name__, static_folder="../frontend", static_url_path="") # Prototipo in evoluzione rapida: mai cache sui file statici (index.html/app.js), # altrimenti pywebview puo' servire una versione vecchia dopo un git pull senza # che sia visibile alcun errore — solo un comportamento "sbagliato" silenzioso. app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0 @app.after_request def _no_cache(response): response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" return response @app.get("/") def index(): return app.send_static_file("index.html") @app.get("/vendor/datagrid/") def datagrid_static(filename): return send_from_directory(DATAGRID_SRC_DIR, filename) @app.get("/api/results/") def api_results(token): result = _RESULTS_STORE.get(token) if result is None: return jsonify({"error": "risultati non trovati (scaduti o mai esistiti)"}), 404 return jsonify(result) # Stato del workspace Blockly (i blocchi sulla lavagna) — persistito su disco per # non dover ricostruire la query ad ogni riavvio durante i test. File locale, # mai in git (dati di sessione, non sorgente) — un solo workspace alla volta, # nessuna storia/versioning: sovrascritto ad ogni salvataggio. WORKSPACE_STATE_PATH = Path(__file__).resolve().parent.parent / ".workspace_state.json" @app.get("/api/workspace") def api_workspace_get(): if not WORKSPACE_STATE_PATH.exists(): return jsonify(None) return jsonify(json.loads(WORKSPACE_STATE_PATH.read_text(encoding="utf-8"))) @app.post("/api/workspace") def api_workspace_save(): state = request.get_json(force=True) WORKSPACE_STATE_PATH.write_text(json.dumps(state), encoding="utf-8") return jsonify({"ok": True}) def _fields_response(fields_dict, labels_dict): return jsonify([ {"name": name, "type": ftype, "label": labels_dict.get(name, name)} for name, ftype in fields_dict.items() ]) @app.get("/api/fields") def api_fields(): return _fields_response(FIELDS, ANAGRAFICA_LABELS) @app.get("/api/fields_emesso") def api_fields_emesso(): return _fields_response(EMESSO_FIELDS, EMESSO_LABELS) @app.get("/api/fields_boxoffice") def api_fields_boxoffice(): return _fields_response(BOXOFFICE_FIELDS, BOXOFFICE_LABELS) @app.get("/api/fields_diritti") def api_fields_diritti(): return _fields_response(DIRITTI_FIELDS, DIRITTI_LABELS) @app.get("/api/fields_gemma") def api_fields_gemma(): return _fields_response(GEMMA_FIELDS, GEMMA_LABELS) # Label "decodificate" per l'header della griglia risultati (results.html) — # a differenza di ANAGRAFICA_LABELS/EMESSO_LABELS/... (nomi campo grezzi, usati # nelle tendine per dominio), qui le chiavi sono i nomi di COLONNA IN OUTPUT # cosi' come escono da /api/query (con prefisso dominio per i satelliti, es. # "emesso_rete") — costruito una volta sola all'avvio, non ricalcolato ad ogni # richiesta. _SATELLITE_LABELS = { "emesso": EMESSO_LABELS, "boxoffice": BOXOFFICE_LABELS, "diritti": DIRITTI_LABELS, "gemma": GEMMA_LABELS, } def _build_output_labels(): labels = dict(ANAGRAFICA_LABELS) for domain, domain_labels in _SATELLITE_LABELS.items(): for field, label in domain_labels.items(): labels[f"{domain}_{field}"] = label return labels OUTPUT_LABELS = _build_output_labels() @app.get("/api/labels") def api_labels(): return jsonify(OUTPUT_LABELS) # Stesso principio di OUTPUT_LABELS sopra, ma per le larghezze fisse stile # Linker — chiavi = nomi di colonna in output (con prefisso dominio per i # satelliti). _SATELLITE_WIDTHS = { "emesso": EMESSO_WIDTHS, "boxoffice": BOXOFFICE_WIDTHS, "diritti": DIRITTI_WIDTHS, "gemma": GEMMA_WIDTHS, } def _build_output_widths(): widths = dict(ANAGRAFICA_WIDTHS) for domain, domain_widths in _SATELLITE_WIDTHS.items(): for field, width in domain_widths.items(): widths[f"{domain}_{field}"] = width return widths OUTPUT_WIDTHS = _build_output_widths() @app.get("/api/widths") def api_widths(): return jsonify(OUTPUT_WIDTHS) def _build_filtro(node, fields_dict, table_alias): field = node.get("field") op = node.get("op") value = node.get("value") if field not in fields_dict: raise ValueError(f"campo non valido: {field}") ftype = fields_dict[field] if ftype == "string": ops = OPS_STRING elif ftype == "date": ops = OPS_DATE else: ops = OPS_INT if op not in ops: raise ValueError(f"operatore non valido per {field}: {op}") sql_op = ops[op] # Prefisso tabella obbligatorio quando e' presente un JOIN (passo 2): # alcuni nomi di campo (es. "tipologia") esistono in entrambe le tabelle # — senza prefisso DuckDB solleva un errore di colonna ambigua. Dentro la # subquery Emesso (una sola tabella in scope) il prefisso non serve. prefix = f'{table_alias}.' if table_alias else "" caster = float if ftype == "float" else int if op in ("is_null", "is_not_null"): return f'{prefix}"{field}" {sql_op}', [] if op == "contains": # "contiene" resta sulla stringa grezza anche per le date (es. # "contiene 2026" senza scrivere una data completa) — nessuna # conversione di formato del valore. Ma per un campo "date" la # colonna nel WHERE e' ancora quella grezza del parquet sorgente # (spesso DATE, non VARCHAR: il CAST a VARCHAR avviene solo nella # SELECT a valle, non visibile qui) — ILIKE non fa cast implicito # DATE->VARCHAR come invece fanno gli operatori di confronto, quindi # serve un CAST esplicito solo per questo operatore. col = f'CAST({prefix}"{field}" AS VARCHAR)' if ftype == "date" else f'{prefix}"{field}"' return f'{col} {sql_op} ?', [f"%{value}%"] if op == "between": parts = [p.strip() for p in str(value).split(",")] if len(parts) != 2: raise ValueError(f"'{field}': between richiede due valori separati da virgola") conv = _parse_date_value if ftype == "date" else caster return f'{prefix}"{field}" BETWEEN ? AND ?', [conv(parts[0]), conv(parts[1])] if ftype == "string": return f'{prefix}"{field}" {sql_op} ?', [value] if ftype == "date": return f'{prefix}"{field}" {sql_op} ?', [_parse_date_value(value)] return f'{prefix}"{field}" {sql_op} ?', [caster(value)] def _build_node(node, fields_dict, table_alias): kind = node.get("kind") if kind == "filtro": return _build_filtro(node, fields_dict, table_alias) if kind == "gruppo": op = node.get("op") if op not in ("AND", "OR"): raise ValueError(f"operatore di gruppo non valido: {op}") parts, params = _build_where(node.get("children", []), fields_dict, table_alias) if not parts: return None return "(" + f" {op} ".join(parts) + ")", params raise ValueError(f"nodo non valido: {kind}") def _build_where(nodes, fields_dict, table_alias): clauses = [] params = [] for node in nodes: built = _build_node(node, fields_dict, table_alias) if built is None: continue clause, node_params = built clauses.append(clause) params.extend(node_params) return clauses, params def _parse_query_body(body, max_limit): filters = body.get("filters", []) limit = min(int(body.get("limit", 200)), max_limit) output_fields = body.get("fields") if output_fields is None: output_fields = list(FIELDS) invalid = [f for f in output_fields if f not in FIELDS] if invalid: raise ValueError(f"campi in output non validi: {invalid}") if not output_fields: raise ValueError("nessun campo selezionato per l'output") emesso = body.get("emesso") if emesso is not None: # "aggregation" e' concetto rimandato a un blocco a se' (step # successivo, non ancora nel frontend) — default "tutte" (una riga # per emissione) quando assente, cosi' il backend resta compatibile # sia con la versione odierna del blocco sia con quella futura. aggregation = emesso.get("aggregation", "tutte") if aggregation not in ("prima", "ultima", "tutte"): raise ValueError(f"aggregazione Emesso non valida: {aggregation}") emesso_fields = emesso.get("fields") if emesso_fields is None: emesso_fields = list(EMESSO_FIELDS) invalid_e = [f for f in emesso_fields if f not in EMESSO_FIELDS] if invalid_e: raise ValueError(f"campi Emesso non validi: {invalid_e}") if not emesso_fields: raise ValueError("nessun campo Emesso selezionato per l'output") join_type = emesso.get("join_type", "auto") if join_type not in ("auto", "left", "inner"): raise ValueError(f"tipo di join non valido: {join_type}") emesso = { "aggregation": aggregation, "fields": emesso_fields, "filters": emesso.get("filters", []), "join_type": join_type, } boxoffice = body.get("boxoffice") if boxoffice is not None: # Nessuna "aggregation" qui a differenza di Emesso: relazione 1:1 su # prodotto, non serve scegliere prima/ultima/tutte. boxoffice_fields = boxoffice.get("fields") if boxoffice_fields is None: boxoffice_fields = list(BOXOFFICE_FIELDS) invalid_b = [f for f in boxoffice_fields if f not in BOXOFFICE_FIELDS] if invalid_b: raise ValueError(f"campi Boxoffice non validi: {invalid_b}") if not boxoffice_fields: raise ValueError("nessun campo Boxoffice selezionato per l'output") join_type_b = boxoffice.get("join_type", "auto") if join_type_b not in ("auto", "left", "inner"): raise ValueError(f"tipo di join Boxoffice non valido: {join_type_b}") boxoffice = { "fields": boxoffice_fields, "filters": boxoffice.get("filters", []), "join_type": join_type_b, } diritti = body.get("diritti") if diritti is not None: # Nessuna "aggregation" qui, a differenza di Emesso: la specifica # ufficiale segnala che Diritti non e' esprimibile come semplice JOIN # (titolarita' %, passaggi residui, first-run/re-run, inibizioni sono # logica di business derivata) — questo primo giro espone solo i # campi grezzi con un JOIN semplice (relazione 1:molti, "tutte" le # righe diritto), la logica derivata resta un layer a se' futuro. diritti_fields = diritti.get("fields") if diritti_fields is None: diritti_fields = list(DIRITTI_FIELDS) invalid_d = [f for f in diritti_fields if f not in DIRITTI_FIELDS] if invalid_d: raise ValueError(f"campi Diritti non validi: {invalid_d}") if not diritti_fields: raise ValueError("nessun campo Diritti selezionato per l'output") join_type_d = diritti.get("join_type", "auto") if join_type_d not in ("auto", "left", "inner"): raise ValueError(f"tipo di join Diritti non valido: {join_type_d}") diritti = { "fields": diritti_fields, "filters": diritti.get("filters", []), "join_type": join_type_d, } gemma = body.get("gemma") if gemma is not None: # Nessuna "aggregation": relazione gia' fan-out per costruzione (vedi # _build_gemma_join, imdb.riferimento_imdb non univoco) — si accetta # la cardinalita' naturale del join a catena, stesso principio gia' # usato per "tutte" le righe di Emesso/Diritti. gemma_fields = gemma.get("fields") if gemma_fields is None: gemma_fields = list(GEMMA_FIELDS) invalid_g = [f for f in gemma_fields if f not in GEMMA_FIELDS] if invalid_g: raise ValueError(f"campi Gemma non validi: {invalid_g}") if not gemma_fields: raise ValueError("nessun campo Gemma selezionato per l'output") join_type_g = gemma.get("join_type", "auto") if join_type_g not in ("auto", "left", "inner"): raise ValueError(f"tipo di join Gemma non valido: {join_type_g}") gemma = { "fields": gemma_fields, "filters": gemma.get("filters", []), "join_type": join_type_g, } return filters, output_fields, limit, emesso, boxoffice, diritti, gemma def _build_emesso_join(emesso): """JOIN su Emesso: 'prima'/'ultima' usano una window function per prendere la riga INTERA della prima/ultima emissione (non solo la data), 'tutte' e' un JOIN semplice (una riga per emissione, cambia cardinalita'). LEFT JOIN sempre: un prodotto senza emissioni resta comunque in output. I filtri Emesso (es. rete/tipologia) si applicano DENTRO la subquery, prima di scegliere prima/ultima — filtrare "prima emissione su Canale 5" deve restringere alle emissioni Canale 5 e poi prendere la piu' vecchia, non il contrario. """ emesso_clauses, emesso_params = _build_where( emesso.get("filters", []), EMESSO_FIELDS, table_alias=None ) emesso_where_sql = f"WHERE {' AND '.join(emesso_clauses)}" if emesso_clauses else "" emesso_posix = EMESSO_PARQUET_PATH.as_posix() aggregation = emesso["aggregation"] if aggregation == "tutte": emesso_from = f"(SELECT * FROM read_parquet('{emesso_posix}') {emesso_where_sql})" else: order = "ASC" if aggregation == "prima" else "DESC" emesso_from = f"""( SELECT * FROM ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY prodotto, edizione ORDER BY data_emissione {order} ) AS rn FROM read_parquet('{emesso_posix}') {emesso_where_sql} ) WHERE rn = 1 )""" # Scelta del tipo di JOIN: "auto" (default) sceglie INNER se ci sono # filtri Emesso — un filtro deve restringere i prodotti ("solo chi ha # un'emissione che soddisfa questo criterio"), non limitarsi a mostrare # NULL per chi non corrisponde — altrimenti LEFT (Emesso arricchisce # l'output senza restringere le righe Anagrafica). "left"/"inner" forzano # esplicitamente la scelta (blocco "ponte" nel frontend), a prescindere # dalla presenza di filtri. join_type_choice = emesso.get("join_type", "auto") if join_type_choice == "left": join_type = "LEFT JOIN" elif join_type_choice == "inner": join_type = "INNER JOIN" else: join_type = "INNER JOIN" if emesso_clauses else "LEFT JOIN" join_sql = ( f'{join_type} {emesso_from} e ' f'ON a."prodotto" = e."prodotto" AND a."edizione" = e."edizione"' ) cols = ", ".join(f'e."{f}" AS "emesso_{f}"' for f in emesso["fields"]) return join_sql, cols, emesso_params def _build_boxoffice_join(boxoffice): """JOIN su Boxoffice: relazione 1:1 su 'prodotto' (verificato sui dati reali, 39.771 righe = 39.771 prodotti distinti) — a differenza di Emesso non serve ne' window function ne' scelta di aggregazione, un JOIN diretto basta e non cambia mai la cardinalita' del risultato. Stessa regola automatica LEFT/INNER di Emesso: con filtri Boxoffice presenti un prodotto senza corrispondenza deve uscire dal risultato (INNER), altrimenti Boxoffice arricchisce senza restringere (LEFT). """ bo_clauses, bo_params = _build_where( boxoffice.get("filters", []), BOXOFFICE_FIELDS, table_alias=None ) bo_where_sql = f"WHERE {' AND '.join(bo_clauses)}" if bo_clauses else "" bo_posix = BOXOFFICE_PARQUET_PATH.as_posix() # CAST esplicito di data_debutto a VARCHAR: la colonna e' DATE nel # parquet, va normalizzata a stringa sia per il filtro sia per l'output # (jsonify non serializza datetime.date). bo_from = f"""( SELECT prodotto, stagione, CAST(data_debutto AS VARCHAR) AS data_debutto, distributore, incasso, spettatori FROM read_parquet('{bo_posix}') {bo_where_sql} )""" join_type_choice = boxoffice.get("join_type", "auto") if join_type_choice == "left": join_type = "LEFT JOIN" elif join_type_choice == "inner": join_type = "INNER JOIN" else: join_type = "INNER JOIN" if bo_clauses else "LEFT JOIN" join_sql = f'{join_type} {bo_from} bo ON a."prodotto" = bo."prodotto"' cols = ", ".join(f'bo."{f}" AS "boxoffice_{f}"' for f in boxoffice["fields"]) return join_sql, cols, bo_params def _build_diritti_join(diritti): """JOIN su Diritti: relazione 1:molti su prod+ediz (diritti) = prodotto+ edizione (Anagrafica) — nomi di colonna diversi tra le due tabelle, a differenza di Emesso/Boxoffice che condividono gia' "prodotto"/ "edizione". Nessuna window function/aggregazione (a differenza di Emesso): questo primo giro restituisce sempre tutte le righe diritto che soddisfano i filtri, la scelta di eventuale aggregazione e' rimandata insieme al resto della logica di business derivata (vedi DIRITTI_FIELDS). diritti_enrich.parquet e' agganciata qui dentro come tabella "sidecar" (LEFT JOIN 1:1 su id_diritto, verificato zero righe orfane in entrambe le direzioni) — non e' un dominio/blocco a se', i suoi due campi (fornitore_cluster, esclusivo_tem) sono semplicemente altri due campi di "Tabella Diritti" agli occhi dell'utente e del resto del backend. Stessa regola automatica LEFT/INNER di Emesso/Boxoffice (qui riferita all'insieme diritti+diritti_enrich): con filtri Diritti presenti un prodotto senza corrispondenza deve uscire dal risultato (INNER), altrimenti Diritti arricchisce senza restringere (LEFT). """ d_clauses, d_params = _build_where( diritti.get("filters", []), DIRITTI_FIELDS, table_alias=None ) d_where_sql = f"WHERE {' AND '.join(d_clauses)}" if d_clauses else "" d_posix = DIRITTI_PARQUET_PATH.as_posix() enrich_posix = DIRITTI_ENRICH_PARQUET_PATH.as_posix() # decr/scad/decr_inib/scad_inib sono DATE nel parquet — CAST esplicito a # VARCHAR per coerenza con "data_debutto" di Boxoffice (jsonify non # serializza datetime.date, e i filtri stringa li trattano gia' come tali). date_cols = {"decr", "scad", "decr_inib", "scad_inib"} enrich_cols = {"fornitore_cluster", "esclusivo_tem"} raw_cols = [f for f in DIRITTI_FIELDS if f not in enrich_cols] select_cols_sql = ", ".join( f'CAST(dr."{f}" AS VARCHAR) AS "{f}"' if f in date_cols else f'dr."{f}"' for f in raw_cols ) d_from = f"""( SELECT dr.prod, dr.ediz, {select_cols_sql}, en."fornitore_cluster" AS "fornitore_cluster", CAST(en."ESCLUSIVO_TEM" AS VARCHAR) AS "esclusivo_tem" FROM read_parquet('{d_posix}') dr LEFT JOIN read_parquet('{enrich_posix}') en ON dr."id_diritto" = en."id_diritto" {d_where_sql} )""" join_type_choice = diritti.get("join_type", "auto") if join_type_choice == "left": join_type = "LEFT JOIN" elif join_type_choice == "inner": join_type = "INNER JOIN" else: join_type = "INNER JOIN" if d_clauses else "LEFT JOIN" join_sql = ( f'{join_type} {d_from} d ' f'ON a."prodotto" = d."prod" AND a."edizione" = d."ediz"' ) cols = ", ".join(f'd."{f}" AS "diritti_{f}"' for f in diritti["fields"]) return join_sql, cols, d_params def _build_gemma_join(gemma): """JOIN su Gemma: a differenza degli altri domini, Gemma non ha "prodotto"/"edizione" propri — solo un riferimento IMDb (A_COD_IMDB). Su richiesta esplicita di Mauro il join non passa per A_CODICE_PRODOTTO (presente solo sul 9,5% delle righe) ma per IMDb: A_COD_IMDB -> imdb.riferimento_imdb -> imdb.codice = prodotti.prodotto. Il JOIN verso imdb dentro la subquery e' sempre INNER: una riga Gemma senza un codice IMDb valido/risolvibile non ha comunque modo di agganciarsi ad Anagrafica. ATTENZIONE fan-out: imdb.riferimento_imdb NON e' univoco (uno stesso id IMDb puo' corrispondere a piu' "codice" interni, fino a 41 osservati) — il join a catena puo' quindi restituire piu' righe di quante ce ne sarebbero con un match diretto 1:1. Accettato come caratteristica nota del dato (stesso principio gia' usato per il fan-out di Emesso/Diritti), non filtrato/deduplicato in questo primo giro. "a_data"/"a_deadline" sono stringhe in formato italiano GG/MM/AAAA nel parquet sorgente (non un tipo DATE nativo) — convertite a ISO qui via strptime, per usare lo stesso tipo "date" del resto del sistema. """ g_clauses, g_params = _build_where( gemma.get("filters", []), GEMMA_FIELDS, table_alias=None ) g_where_sql = f"WHERE {' AND '.join(g_clauses)}" if g_clauses else "" g_posix = GEMMA_PARQUET_PATH.as_posix() imdb_posix = IMDB_PARQUET_PATH.as_posix() date_cols = {"a_data", "a_deadline"} select_cols_sql = ", ".join( f'CAST(CAST(strptime(g."{f.upper()}", \'%d/%m/%Y\') AS DATE) AS VARCHAR) AS "{f}"' if f in date_cols else f'g."{f.upper()}" AS "{f}"' for f in GEMMA_FIELDS ) g_from = f"""( SELECT im."codice" AS "prodotto", {select_cols_sql} FROM read_parquet('{g_posix}') g INNER JOIN read_parquet('{imdb_posix}') im ON g."A_COD_IMDB" = im."riferimento_imdb" {g_where_sql} )""" join_type_choice = gemma.get("join_type", "auto") if join_type_choice == "left": join_type = "LEFT JOIN" elif join_type_choice == "inner": join_type = "INNER JOIN" else: join_type = "INNER JOIN" if g_clauses else "LEFT JOIN" join_sql = f'{join_type} {g_from} gem ON a."prodotto" = gem."prodotto"' cols = ", ".join(f'gem."{f}" AS "gemma_{f}"' for f in gemma["fields"]) return join_sql, cols, g_params def _anagrafica_base_sql(): """FROM per Anagrafica, con i tre campi derivati (imdb_codice/regista/ attori) gia' agganciati — stessa logica di v_custom_anagr in Linker (PYTHON/MyICR_Suite/app/modules/linker/src/database/parquet_db.py), riletta 1:1 qui invece di reinventarla. Nessuna deduplicazione per "prodotto" (a differenza di v_custom_anagr, che tiene una sola edizione): PowerBricks espone il grano prodotto+edizione intero, i tre campi derivati sono comunque calcolati per "prodotto" quindi si ripetono identici su tutte le edizioni dello stesso prodotto — nessun fan-out, "codice" di imdb.parquet e' univoco (verificato: 110.773 righe = 110.773), "regista"/"attori" sono gia' aggregati a 1 riga per prodotto dalle subquery sotto. """ prod_posix = PARQUET_PATH.as_posix() imdb_posix = IMDB_PARQUET_PATH.as_posix() cast_posix = CAST_PARQUET_PATH.as_posix() return f"""( SELECT p.*, i.riferimento_imdb AS imdb_codice, r.regista, c.attori FROM read_parquet('{prod_posix}') p LEFT JOIN ( SELECT codice, MIN(riferimento_imdb) AS riferimento_imdb FROM read_parquet('{imdb_posix}') GROUP BY codice ) i ON i.codice = p.prodotto LEFT JOIN ( SELECT prodotto, TRIM(COALESCE(nome, '') || ' ' || COALESCE(cognome, '')) AS regista FROM read_parquet('{cast_posix}') WHERE ruolo = 'FRE' QUALIFY ROW_NUMBER() OVER (PARTITION BY prodotto ORDER BY progr_cast) = 1 ) r ON r.prodotto = p.prodotto LEFT JOIN ( SELECT prodotto, STRING_AGG(TRIM(COALESCE(nome, '') || ' ' || COALESCE(cognome, '')), ', ' ORDER BY progr_cast) AS attori FROM read_parquet('{cast_posix}') WHERE ruolo = 'C001' AND progr_cast <= 3 GROUP BY prodotto ) c ON c.prodotto = p.prodotto )""" def _execute_query(filters, output_fields, limit, emesso=None, boxoffice=None, diritti=None, gemma=None): clauses, anagrafica_params = _build_where(filters, FIELDS, table_alias="a") where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else "" select_cols = ", ".join(f'a."{c}"' for c in output_fields) join_sql = "" join_params = [] if emesso: emesso_join_sql, emesso_cols, emesso_params = _build_emesso_join(emesso) join_sql += f" {emesso_join_sql}" join_params += emesso_params select_cols = f"{select_cols}, {emesso_cols}" if boxoffice: bo_join_sql, bo_cols, bo_params = _build_boxoffice_join(boxoffice) join_sql += f" {bo_join_sql}" join_params += bo_params select_cols = f"{select_cols}, {bo_cols}" if diritti: d_join_sql, d_cols, d_params = _build_diritti_join(diritti) join_sql += f" {d_join_sql}" join_params += d_params select_cols = f"{select_cols}, {d_cols}" if gemma: g_join_sql, g_cols, g_params = _build_gemma_join(gemma) join_sql += f" {g_join_sql}" join_params += g_params select_cols = f"{select_cols}, {g_cols}" # Il path e' una costante di configurazione (non input utente): sicuro da interpolare. sql = f""" SELECT {select_cols} FROM {_anagrafica_base_sql()} a {join_sql} {where_sql} LIMIT {limit} """ # Ordine parametri: i JOIN (con gli eventuali placeholder dei filtri # Emesso/Boxoffice, in quest'ordine nel testo SQL) compaiono prima della # WHERE Anagrafica. con = duckdb.connect(":memory:") rows = con.execute(sql, join_params + anagrafica_params).fetchall() columns = [c[0] for c in con.description] return columns, rows # Mappa dominio -> FIELDS dict, usata sia per validare i campi in # consultazione standalone sia per costruire il WHERE (_build_where e' # sempre lo stesso motore generico, cambia solo il dizionario di riferimento). STANDALONE_FIELDS = { "emesso": EMESSO_FIELDS, "boxoffice": BOXOFFICE_FIELDS, "diritti": DIRITTI_FIELDS, "gemma": GEMMA_FIELDS, } def _standalone_base_sql(domain): """FROM per la consultazione di un singolo dominio satellite SENZA passare da Anagrafica ("consultazione come tabella singola", Mauro, 06/08/2026: per una verifica mirata su un dominio non serve sempre comporre con Anagrafica). Ogni dominio riusa lo stesso cast dei campi data della modalita' unita a Anagrafica (necessario indipendentemente dal join, e' la serializzazione JSON che lo richiede), ma **senza** la logica che esiste solo in funzione del legame con Anagrafica: niente aggregazione prima/ultima per Emesso (concetto legato al confronto con piu' emissioni dello stesso prodotto/edizione, qui non c'e' un'edizione di riferimento), niente risoluzione IMDb per Gemma (serviva solo a calcolare "prodotto" per il join, qui gemma.parquet si legge cosi' com'e'). diritti_enrich resta agganciata (e' "trasparente", parte della stessa Tabella Diritti indipendentemente dal contesto). """ if domain == "emesso": # data_emissione e' gia' VARCHAR nel parquet sorgente (a differenza # di data_debutto/decr/scad), nessun cast necessario. return f"read_parquet('{EMESSO_PARQUET_PATH.as_posix()}')" if domain == "boxoffice": cols_sql = ", ".join( f'CAST("{f}" AS VARCHAR) AS "{f}"' if f == "data_debutto" else f'"{f}"' for f in BOXOFFICE_FIELDS ) return f"(SELECT {cols_sql} FROM read_parquet('{BOXOFFICE_PARQUET_PATH.as_posix()}'))" if domain == "diritti": date_cols = {"decr", "scad", "decr_inib", "scad_inib"} enrich_cols = {"fornitore_cluster", "esclusivo_tem"} raw_cols = [f for f in DIRITTI_FIELDS if f not in enrich_cols] cols_sql = ", ".join( f'CAST(dr."{f}" AS VARCHAR) AS "{f}"' if f in date_cols else f'dr."{f}"' for f in raw_cols ) return f"""( SELECT {cols_sql}, en."fornitore_cluster" AS "fornitore_cluster", CAST(en."ESCLUSIVO_TEM" AS VARCHAR) AS "esclusivo_tem" FROM read_parquet('{DIRITTI_PARQUET_PATH.as_posix()}') dr LEFT JOIN read_parquet('{DIRITTI_ENRICH_PARQUET_PATH.as_posix()}') en ON dr."id_diritto" = en."id_diritto" )""" if domain == "gemma": date_cols = {"a_data", "a_deadline"} cols_sql = ", ".join( f'CAST(CAST(strptime(g."{f.upper()}", \'%d/%m/%Y\') AS DATE) AS VARCHAR) AS "{f}"' if f in date_cols else f'g."{f.upper()}" AS "{f}"' for f in GEMMA_FIELDS ) return f"(SELECT {cols_sql} FROM read_parquet('{GEMMA_PARQUET_PATH.as_posix()}') g)" raise ValueError(f"dominio non valido per consultazione singola: {domain}") def _parse_standalone_body(body, max_limit): standalone = body.get("standalone") or {} domain = standalone.get("domain") if domain not in STANDALONE_FIELDS: raise ValueError(f"dominio non valido per consultazione singola: {domain}") fields_dict = STANDALONE_FIELDS[domain] output_fields = standalone.get("fields") if output_fields is None: output_fields = list(fields_dict) invalid = [f for f in output_fields if f not in fields_dict] if invalid: raise ValueError(f"campi non validi per {domain}: {invalid}") if not output_fields: raise ValueError("nessun campo selezionato per l'output") filters = standalone.get("filters", []) limit = min(int(body.get("limit", 200)), max_limit) return domain, filters, output_fields, limit def _execute_standalone_query(domain, filters, output_fields, limit): fields_dict = STANDALONE_FIELDS[domain] clauses, params = _build_where(filters, fields_dict, table_alias=None) where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else "" select_cols = ", ".join(f'"{c}"' for c in output_fields) base_sql = _standalone_base_sql(domain) sql = f""" SELECT {select_cols} FROM {base_sql} {where_sql} LIMIT {limit} """ con = duckdb.connect(":memory:") rows = con.execute(sql, params).fetchall() columns = [c[0] for c in con.description] return columns, rows @app.post("/api/query") def api_query(): body = request.get_json(force=True) or {} try: if "standalone" in body: domain, filters, output_fields, limit = _parse_standalone_body(body, max_limit=1000) columns, rows = _execute_standalone_query(domain, filters, output_fields, limit) else: filters, output_fields, limit, emesso, boxoffice, diritti, gemma = _parse_query_body(body, max_limit=1000) columns, rows = _execute_query(filters, output_fields, limit, emesso, boxoffice, diritti, gemma) except ValueError as e: return jsonify({"error": str(e)}), 400 return jsonify({"columns": columns, "rows": rows}) @app.post("/api/open_grid") def api_open_grid(): """Esegue la query e apre i risultati in una finestra pywebview separata, con la griglia web (componente datagrid, sandbox/datagrid/) — nessun conflitto di loop eventi come con PyQt6: e' la stessa tecnologia (pywebview) della finestra principale, solo una seconda istanza.""" body = request.get_json(force=True) or {} try: if "standalone" in body: domain, filters, output_fields, limit = _parse_standalone_body(body, max_limit=20000) columns, rows = _execute_standalone_query(domain, filters, output_fields, limit) else: filters, output_fields, limit, emesso, boxoffice, diritti, gemma = _parse_query_body(body, max_limit=20000) columns, rows = _execute_query(filters, output_fields, limit, emesso, boxoffice, diritti, gemma) except ValueError as e: return jsonify({"error": str(e)}), 400 token = _store_results(columns, rows) webview.create_window( f"PowerBricks — Risultati ({len(rows)} righe)", f"http://{HOST}:{PORT}/results.html?token={token}", width=1100, height=700, ) return jsonify({"ok": True, "rows": len(rows), "token": token}) if __name__ == "__main__": app.run(host="127.0.0.1", port=5050, debug=True, use_reloader=False, threaded=True)