""" ProdottiAcquistatiReport — Check settimanale prodotti con diritti non marcati come Acquistati Logica: 1. Prodotti con diritti (decr IS NOT NULL) da layer2.duckdb — tabella diritti join via imdb (codice → riferimento_imdb) → A_COD_IMDB in gemma 2. Escludi se GEMMA.A_ACQUISTATO_VENDUTO = 'Acquistato' 3. Escludi IMDB presenti in eccezioni.db 4. Output: IMDB, Tipologia, Titolo, Distributore, Data (min) — ordinato per Data DESC Sorgente dati: layer2.duckdb (diritti + imdb + gemma) Whitelist eccezioni: eccezioni.db (SQLite locale) """ import argparse import json import sqlite3 from datetime import datetime from pathlib import Path import duckdb 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_prodotti_con_diritti(cfg: dict) -> pd.DataFrame: """ Prodotti con diritti (decr IS NOT NULL) — join DuckDB su parquet locali. """ base = cfg["parquet_base_path"].replace("\\", "/") con = duckdb.connect() df = con.execute(f""" SELECT DISTINCT g.* FROM '{base}/diritti.parquet' d JOIN '{base}/imdb.parquet' i ON d.prod = i.codice JOIN '{base}/gemma.parquet' g ON i.riferimento_imdb = g.A_COD_IMDB WHERE d.decr IS NOT NULL """).df() con.close() logger.info(f"GEMMA righe con diritti (parquet locale): {len(df)}") return df def get_eccezioni(cfg: dict) -> set: path = BASE_DIR / cfg.get("eccezioni_db", "eccezioni.db") conn = sqlite3.connect(path) rows = conn.execute("SELECT imdb FROM eccezioni").fetchall() conn.close() return {r[0] for r in rows} # --------------------------------------------------------------------------- # Logica # --------------------------------------------------------------------------- def build_report(df: pd.DataFrame, eccezioni: set) -> pd.DataFrame: # Escludi già marcati come Acquistato in GEMMA n0 = len(df) df = df[df["A_ACQUISTATO_VENDUTO"].isna() | (df["A_ACQUISTATO_VENDUTO"] != "Acquistato")] logger.info(f"Dopo filtro Acquistato: {len(df)} (rimossi {n0 - len(df)})") # Escludi eccezioni n1 = len(df) df = df[~df["A_COD_IMDB"].isin(eccezioni)] logger.info(f"Dopo filtro eccezioni: {len(df)} (rimossi {n1 - len(df)})") # Group by IMDB — una riga per IMDB con la data minima df["A_DATA"] = pd.to_datetime(df["A_DATA"], errors="coerce") df_grouped = ( df.groupby("A_COD_IMDB", as_index=False) .agg( Tipologia=("A_TIPOLOGIA", "first"), Titolo=("A_TITOLO", "first"), Distributore=("A_DISTRIBUTORE", "first"), Data=("A_DATA", "min"), ) .rename(columns={"A_COD_IMDB": "IMDB"}) .sort_values("Data", ascending=False) .reset_index(drop=True) ) logger.info(f"Prodotti unici nel report: {len(df_grouped)}") return df_grouped # --------------------------------------------------------------------------- # Generazione Excel # --------------------------------------------------------------------------- HEADERS = ["IMDB", "TIPOLOGIA", "TITOLO", "DISTRIBUTORE", "DATA"] COL_MAP = ["IMDB", "Tipologia", "Titolo", "Distributore", "Data"] COL_WIDTHS = [14, 20, 42, 32, 12] GREEN = PatternFill(start_color="E2EFDA", end_color="E2EFDA", 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"PRODOTTI_ACQUISTATI_{datetime.now().strftime('%Y%m%d')}.xlsx" wb = openpyxl.Workbook() ws = wb.active ws.title = "PRODOTTI ACQUISTATI" # Riga 1: titolo ws.merge_cells(f"A1:{get_column_letter(len(HEADERS))}1") title_cell = ws["A1"] title_cell.value = f"GEMMA — Prodotti acquistati ({datetime.now().strftime('%d/%m/%Y')})" title_cell.font = Font(name=FONT_NAME, bold=True, size=14) title_cell.fill = GREEN 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 = GREEN 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 == "Data" and isinstance(value, pd.Timestamp): cell.value = value.to_pydatetime() 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["prodotti_acquistati_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 prodotto con diritti non marcato come Acquistato.") else: body = email_cfg.get("body", f"Report prodotti acquistati GEMMA ({n_rows} prodotti).") 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 / "prodotti_acquistati_{time:YYYY-MM-DD}.log", rotation="14 days", retention="60 days", level="DEBUG", ) mode = "[DRY-RUN] " if dry_run else "" logger.info(f"=== ProdottiAcquistatiReport avviato {mode}===") cfg = load_config() df = get_prodotti_con_diritti(cfg) eccezioni = get_eccezioni(cfg) logger.info(f"Whitelist eccezioni: {len(eccezioni)} IMDB esclusi") df_report = build_report(df, eccezioni) logger.info(f"Prodotti nel report: {len(df_report)}") if df_report.empty: logger.info("Nessun prodotto da segnalare — invio email senza allegato") if not dry_run: send_email(None, cfg, 0) else: logger.info("[DRY-RUN] Email NON inviata") logger.info(f"=== ProdottiAcquistatiReport 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['prodotti_acquistati_email']['to']}") else: send_email(excel_path, cfg, len(df_report)) logger.info(f"=== ProdottiAcquistatiReport 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)