"""
CUSTOM_EMESSO — emesso filtrato per reti CON/MED e tipologie "scripted".

Sorgenti:
- emesso.parquet      (Layer 1 — tutte le reti, no dedup)
- emesso_tyfx0160.csv (lookup ICR_MONDO per codice rete)
- db_warehouse.accdb  (tabPdfEsiste — date in cui esiste il PDF)

Filtri applicati:
- JOIN emesso_tyfx0160.csv su rete = CCRETETEL
- ICR_MONDO LIKE 'CON%' OR 'MED%'  (34 reti CON/MED)
- tipologia IN (content types con diritti: FILM, TELEFILM, ...)

Colonne output:
    CCPROD, CCRETETEL, DCDATGIO, TCORAINIZ, TCORAFINE,
    QCAUDMED, PCSHAREA, FCPRIMAVIS, TCDURLORDA, TCDURNET,
    ICR_MONDO, ICR_DECOD_RETE_BREVE, CCFASCIA, PDF_ESISTE
"""
from pathlib import Path

import duckdb
import pandas as pd
import pyodbc


# Tipologie "scripted/acquistato" — contenuti che hanno diritti
_TIPOLOGIE_EMESSO = (
    'FILM', 'MINISERIE', 'SIT COM', 'TELEFILM', 'TV MOVIE',
    'SOAP', 'TELENOVELAS', 'DOCUMENTARI', 'CORTOMETRAGGIO', 'TELEROMANZO',
)


def build(parquet_dir: Path, tyfx_csv: str, db_warehouse: str) -> pd.DataFrame:
    em_pq   = str(parquet_dir / 'emesso.parquet').replace('\\', '/')
    tyfx    = tyfx_csv.replace('\\', '/')
    tip_sql = ', '.join(f"'{t}'" for t in _TIPOLOGIE_EMESSO)

    with duckdb.connect() as con:
        df = con.execute(f"""
            SELECT
                e.prodotto       AS CCPROD,
                e.rete           AS CCRETETEL,
                e.data_emissione AS DCDATGIO,
                e.ora_inizio     AS TCORAINIZ,
                e.ora_fine       AS TCORAFINE,
                e.audience       AS QCAUDMED,
                e.share          AS PCSHAREA,
                e.prima_visione  AS FCPRIMAVIS,
                (e.durata_lorda + 10) // 60 AS TCDURLORDA,
                (e.durata_netta + 10) // 60 AS TCDURNET,
                t.ICR_MONDO,
                t.ICR_DECOD_RETE_BREVE,
                e.fascia         AS CCFASCIA
            FROM '{em_pq}' e
            JOIN read_csv('{tyfx}', delim=';', header=true) t
                ON e.rete = t.CCRETETEL
            WHERE (t.ICR_MONDO LIKE 'CON%' OR t.ICR_MONDO LIKE 'MED%')
              AND e.tipologia IN ({tip_sql})
        """).df()

    # PDF_ESISTE: True per C5/I1/R4 in fascia PR nelle date in cui esiste il PDF
    pdf_dates: set = set()
    try:
        conn_str = (
            "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};"
            f"DBQ={db_warehouse};"
        )
        with pyodbc.connect(conn_str, autocommit=True) as wh:
            cur = wh.cursor()
            cur.execute("SELECT data FROM tabPdfEsiste")
            for row in cur.fetchall():
                pdf_dates.add(str(row[0])[:10])
            cur.close()
    except Exception as e:
        print(f"  WARNING tabPdfEsiste non disponibile: {e}")

    df['PDF_ESISTE'] = (
        df['CCRETETEL'].isin(['C5', 'I1', 'R4']) &
        (df['CCFASCIA'] == 'PR') &
        df['DCDATGIO'].astype(str).isin(pdf_dates)
    ).astype(int)

    return df