""" CodiciDoppiReport — Check giornaliero codici IMDB doppi in GEMMA Logica (equivalente VBA): 1. Trova IMDB con count > 1 in GEMMA (A_DATA >= 2014-01-01, escludi 'no imdb' e NULL) 2. Recupera tutte le righe GEMMA con quegli IMDB doppi 3. Esclude KEY_COD (A_ID) presenti in doppi_ok.db (duplicati noti/approvati) 4. Output ordinato per A_COD_IMDB Sorgente dati: gemma.parquet Whitelist: doppi_ok.db (SQLite locale — aggiornare manualmente quando si approva un doppio) """ import argparse import json import sqlite3 from datetime import datetime from pathlib import Path import pandas as pd import openpyxl from openpyxl.styles import PatternFill, Font, Alignment from openpyxl.utils import get_column_letter from loguru import logger BASE_DIR = Path(__file__).parent # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- def load_config() -> dict: with open(BASE_DIR / "config.json", encoding="utf-8") as f: return json.load(f) # --------------------------------------------------------------------------- # Dati # --------------------------------------------------------------------------- def get_gemma_data(cfg: dict) -> pd.DataFrame: path = cfg["gemma_parquet_path"] df = pd.read_parquet(path) # Filtri base: data >= 2014, IMDB valido df = df[df["A_COD_IMDB"].str.startswith("tt", na=False)] df = df[pd.to_datetime(df["A_DATA"], errors="coerce", dayfirst=True) >= pd.Timestamp("2014-01-01")] logger.info(f"GEMMA parquet: {len(df)} righe dopo filtro data/IMDB") return df def get_doppi_ok(cfg: dict) -> set: path = BASE_DIR / cfg.get("doppi_ok_db", "doppi_ok.db") conn = sqlite3.connect(path) rows = conn.execute("SELECT key_cod FROM doppi_ok").fetchall() conn.close() return {str(r[0]) for r in rows} # --------------------------------------------------------------------------- # Logica # --------------------------------------------------------------------------- def build_report(df: pd.DataFrame, doppi_ok: set) -> pd.DataFrame: # Step 1: IMDB che compaiono più di una volta imdb_counts = df.groupby("A_COD_IMDB")["A_ID"].count() imdb_doppi = set(imdb_counts[imdb_counts > 1].index) logger.info(f"IMDB doppi trovati: {len(imdb_doppi)}") # Step 2: tutte le righe con quegli IMDB df_doppi = df[df["A_COD_IMDB"].isin(imdb_doppi)].copy() # Step 3: escludi KEY_COD approvati n_before = len(df_doppi) df_doppi = df_doppi[~df_doppi["A_ID"].isin(doppi_ok)] logger.info(f"Dopo esclusione doppi_ok: {len(df_doppi)} righe (rimossi {n_before - len(df_doppi)})") # Ricontrolla: escludi anche IMDB dove tutti i duplicati sono stati rimossi imdb_restanti = df_doppi.groupby("A_COD_IMDB")["A_ID"].count() imdb_ancora_doppi = set(imdb_restanti[imdb_restanti > 1].index) df_doppi = df_doppi[df_doppi["A_COD_IMDB"].isin(imdb_ancora_doppi)] logger.info(f"IMDB ancora doppi dopo whitelist: {len(imdb_ancora_doppi)}") df_doppi = df_doppi.sort_values("A_COD_IMDB").reset_index(drop=True) return df_doppi # --------------------------------------------------------------------------- # Generazione Excel # --------------------------------------------------------------------------- HEADERS = [ "KEY_COD", "STATO", "TITOLO", "DATA", "DISTRIBUTORE", "TIPOLOGIA", "TIPO", "REGISTA", "CAST", "IMDB", "SUPPORTO", "NOTE SUPPORTI", "NOTE PUBBLICHE", "NOTE PRIVATE", "V_STATO", "V_RDA", "V_C5", "V_I1", "V_R4", "V_LA5", "V_I2", "V_IRIS", "V_TOP", "V_FOC", "V_C20", "V_CI34", "V_CIN", "V_INF", "V_EMO", "V_ENE", "V_COM", "V_STO", "V_CRI", "V_ACT", ] COL_MAP = [ "A_ID", "A_STATO", "A_TITOLO", "A_DATA", "A_DISTRIBUTORE", "A_TIPOLOGIA", "A_TIPO", "A_AUTORE_REGISTA", "A_CAST", "A_COD_IMDB", "A_SUPPORTO", "A_NOTE_SUPPORTO", "A_NOTE_PUBBLICHE", "A_NOTE_PRIVATE", "V_STATO", "V_RDA", "V_C5", "V_I1", "V_R4", "V_LA5", "V_I2", "V_IRIS", "V_TOP", "V_FOC", "V_C20", "V_CI34", "V_CIN", "V_INF", "V_EMO", "V_ENE", "V_COM", "V_STO", "V_CRI", "V_ACT", ] COL_WIDTHS = [ 9, 10, 40, 12, 28, 16, 12, 22, 28, 14, 12, 18, 28, 28, 9, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, ] ORANGE = PatternFill(start_color="FCE4D6", end_color="FCE4D6", fill_type="solid") FONT_NAME = "Aptos Narrow" def generate_excel(df: pd.DataFrame) -> Path: output_dir = BASE_DIR / "output" output_dir.mkdir(exist_ok=True) output_path = output_dir / f"CODICI_DOPPI_{datetime.now().strftime('%Y%m%d')}.xlsx" wb = openpyxl.Workbook() ws = wb.active ws.title = "CODICI DOPPI" # Riga 1: titolo ws.merge_cells(f"A1:{get_column_letter(len(HEADERS))}1") title_cell = ws["A1"] title_cell.value = f"GEMMA — Codici doppi ({datetime.now().strftime('%d/%m/%Y')})" title_cell.font = Font(name=FONT_NAME, bold=True, size=14) title_cell.fill = ORANGE title_cell.alignment = Alignment(horizontal="left", vertical="center") ws.row_dimensions[1].height = 22 # Riga 2: intestazioni for col, header in enumerate(HEADERS, 1): cell = ws.cell(row=2, column=col, value=header) cell.font = Font(name=FONT_NAME, bold=True, size=9) cell.fill = ORANGE cell.alignment = Alignment(horizontal="center", vertical="center") ws.auto_filter.ref = f"A2:{get_column_letter(len(HEADERS))}2" ws.row_dimensions[2].height = 14 for i, width in enumerate(COL_WIDTHS, 1): ws.column_dimensions[get_column_letter(i)].width = width for row_idx, row in df.iterrows(): excel_row = row_idx + 3 for col_idx, col_name in enumerate(COL_MAP, 1): value = row.get(col_name) if pd.isna(value) if value is not None else False: value = None cell = ws.cell(row=excel_row, column=col_idx, value=value) cell.font = Font(name=FONT_NAME, size=9) if col_name == "A_DATA" and value is not None: cell.number_format = "DD/MM/YYYY" wb.save(output_path) logger.info(f"Excel generato: {output_path.name} ({len(df)} righe)") return output_path # --------------------------------------------------------------------------- # Invio email # --------------------------------------------------------------------------- def send_email(excel_path, cfg: dict, n_rows: int): import win32com.client email_cfg = cfg["codici_doppi_email"] subject = email_cfg["subject"].replace("{date}", datetime.now().strftime("%d/%m/%Y")) if excel_path is None: body = email_cfg.get("body_empty", "Nessun codice doppio trovato oggi.") else: body = email_cfg.get("body", f"Report codici doppi GEMMA ({n_rows} righe).") recipients = "; ".join(email_cfg["to"]) outlook = win32com.client.Dispatch("Outlook.Application") mail = outlook.CreateItem(0) mail.Subject = subject mail.Body = body mail.To = recipients if email_cfg.get("cc"): mail.CC = "; ".join(email_cfg["cc"]) if excel_path is not None: mail.Attachments.Add(str(excel_path.resolve())) mail.Send() logger.info(f"Email inviata — {'con allegato' if excel_path else 'senza allegato'} — TO: {len(email_cfg['to'])}") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def run(dry_run: bool = False): logs_dir = BASE_DIR / "logs" logs_dir.mkdir(exist_ok=True) logger.add( logs_dir / "codici_doppi_{time:YYYY-MM-DD}.log", rotation="14 days", retention="60 days", level="DEBUG", ) mode = "[DRY-RUN] " if dry_run else "" logger.info(f"=== CodiciDoppiReport avviato {mode}===") cfg = load_config() df = get_gemma_data(cfg) doppi_ok = get_doppi_ok(cfg) logger.info(f"Whitelist doppi_ok: {len(doppi_ok)} KEY_COD esclusi") df_report = build_report(df, doppi_ok) logger.info(f"Righe nel report: {len(df_report)}") if df_report.empty: logger.info("Nessun codice doppio — email NON inviata") logger.info(f"=== CodiciDoppiReport completato {mode}===") return excel_path = generate_excel(df_report) if dry_run: logger.info(f"[DRY-RUN] Email NON inviata — Excel: {excel_path}") logger.info(f"[DRY-RUN] Destinatari TO: {cfg['codici_doppi_email']['to']}") else: send_email(excel_path, cfg, len(df_report)) logger.info(f"=== CodiciDoppiReport completato {mode}===") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() run(dry_run=args.dry_run)