"""
CUSTOM_ANAGR_ORIZ + EDIZIONI + SUPERSERIE da parquet Layer 1.

Sorgenti:
- prodotti.parquet, cast.parquet, boxoffice.parquet
- emesso.parquet, diritti.parquet
- imdb.parquet, scelte_rete.parquet
- emesso_tyfx0160.csv

Restituisce tre DataFrame: (df_anagr, df_ediz, df_super)
"""
import datetime
from pathlib import Path

import duckdb

# Tipologie "scripted/acquistato" — stesse 10 della pipeline VBA (DECOD_TIPOLOGIA.SCARICO_DATI='X')
_TIPOLOGIE_ANAGR = (
    'FILM', 'TV MOVIE', 'CORTOMETRAGGIO', 'DOCUMENTARI', 'TELEFILM',
    'MINISERIE', 'SIT COM', 'TELENOVELAS', 'SOAP', 'TELEROMANZO',
)

# Mapping tipologia estesa → codice TIPOL VBA (campo VARCHAR(1) in Access)
_TIPOL_MAP = {
    'FILM':           'F',
    'TV MOVIE':       'M',
    'CORTOMETRAGGIO': '7',
    'DOCUMENTARI':    'D',
    'TELEFILM':       'T',
    'MINISERIE':      'E',
    'SIT COM':        'G',
    'TELENOVELAS':    'A',
    'SOAP':           'O',
    'TELEROMANZO':    'Z',
}


def build(parquet_dir: Path, tyfx_csv: str) -> tuple:
    pq = {k: str(parquet_dir / f'{k}.parquet').replace('\\', '/') for k in
          ('prodotti', 'cast', 'boxoffice', 'emesso', 'diritti', 'imdb', 'scelte_rete')}
    tyfx = tyfx_csv.replace('\\', '/')
    today = datetime.date.today().isoformat()

    with duckdb.connect() as con:

        # ------------------------------------------------------------------ #
        # BASE: prodotti edizione=1, tipologie valide                         #
        # ------------------------------------------------------------------ #
        tip_sql = ', '.join(f"'{t}'" for t in _TIPOLOGIE_ANAGR)
        tipol_case = ' '.join(
            f"WHEN '{k}' THEN '{v}'" for k, v in _TIPOL_MAP.items()
        )

        con.execute(f"""
            CREATE TABLE _base AS
            SELECT
                prodotto                                          AS RIFER,
                CASE FIRST(tipologia) {tipol_case} END            AS TIPOL,
                FIRST(num_episodi)                                AS EPIS,
                FIRST(anno_produzione)                            AS ANNO,
                FIRST(durata)                                     AS DUR,
                CASE FIRST(veg)
                    WHEN 'DA ATTRIBUIRE'          THEN 'N.C'
                    WHEN 'UNIVOCA'                THEN 'UNIV.'
                    WHEN 'EVER GREEN'             THEN 'EV.GR'
                    WHEN 'FICTION AUTOPRODOTTA'   THEN 'F-AUT'
                    WHEN 'FICTION AUTOPRODOTTA RAI' THEN NULL
                    WHEN 'NON GESTITA'            THEN 'N.G'
                    WHEN 'SUPER TOP'              THEN 'SUPER'
                    WHEN 'VISTO-DA ATTRIBUIRE'    THEN 'VISTO'
                    ELSE FIRST(veg)
                END                                               AS VEG,
                FIRST(genere1)                                    AS GENERE1,
                FIRST(genere2)                                    AS GENERE2,
                FIRST(paesi_produzione1)                          AS PAESE,
                FIRST(titolo_italiano)                            AS TI,
                FIRST(titolo_originale)                           AS "TO",
                FIRST(superserie_descr)                           AS SUPERSERIE,
                FIRST(tipologia)                                  AS TIPOL_RAW
            FROM '{pq['prodotti']}'
            WHERE edizione = 1
              AND tipologia IN ({tip_sql})
            GROUP BY prodotto
        """)

        # Aggiungi colonne che verranno popolate successivamente
        for col in (
            'REGISTA_NOME', 'REGISTA_COGN', 'ATTORE1', 'ATTORE2', 'ATTORE3',
            'BOX_DATA', 'BOX_STAGIONE', 'BOX_INCASSO', 'BOX_SPETTATORI', 'BOX_DISTRIBUTORE',
            'IMDB_CODICE', 'SC_RETE_SINTESI',
            'EM_PG_RETE', 'EM_PG_DATA', 'EM_PG_HI', 'EM_PG_FASCIA', 'EM_PG_AUD', 'EM_PG_SHARE',
            'EM_UG_RETE', 'EM_UG_DATA', 'EM_UG_HI', 'EM_UG_FASCIA', 'EM_UG_AUD', 'EM_UG_SHARE',
            'EM_PT_RETE', 'EM_PT_DATA', 'EM_PT_HI', 'EM_PT_FASCIA', 'EM_PT_AUD', 'EM_PT_SHARE',
            'EM_UT_RETE', 'EM_UT_DATA', 'EM_UT_HI', 'EM_UT_FASCIA', 'EM_UT_AUD', 'EM_UT_SHARE',
            'EM_PP_RETE', 'EM_PP_DATA', 'EM_PP_HI', 'EM_PP_FASCIA',
            'EM_UP_RETE', 'EM_UP_DATA', 'EM_UP_HI', 'EM_UP_FASCIA',
            'EM_PSKY_RETE', 'EM_PSKY_DATA', 'EM_PSKY_HI', 'EM_PSKY_FASCIA',
            'EM_USKY_RETE', 'EM_USKY_DATA', 'EM_USKY_HI', 'EM_USKY_FASCIA',
            'DF_ICR_FLAG', 'DF_DECR', 'DF_SCAD', 'DF_FR_RR', 'DF_RAGSOC_DISTR',
            'DF_PERC', 'DF_PASS_CONS_TOT', 'DF_PASS_EFF_TOT', 'DF_CAUSALE',
            'DF_DECR_WIN_1', 'DF_SCAD_WIN_1', 'DF_DECR_WIN_2', 'DF_SCAD_WIN_2', 'DF_WIN_ALTRE',
            'DF_FUT_FR_RR', 'DF_FUT_DECR', 'DF_FUT_SCAD',
            'DTVOD_DECR', 'DTVOD_SCAD', 'DTVOD_FR_RR', 'DTVOD_RAGSOC_DISTR',
        ):
            con.execute(f"ALTER TABLE _base ADD COLUMN \"{col}\" VARCHAR")

        # ------------------------------------------------------------------ #
        # REGISTA (ruolo FRE, min progr_cast)                                 #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _regista AS
            SELECT c.prodotto, FIRST(c.nome) AS nome, FIRST(c.cognome) AS cognome
            FROM '{pq['cast']}' c
            JOIN (
                SELECT prodotto, MIN(progr_cast) AS min_pc
                FROM '{pq['cast']}'
                WHERE ruolo = 'FRE'
                GROUP BY prodotto
            ) m ON c.prodotto = m.prodotto AND c.progr_cast = m.min_pc
            WHERE c.ruolo = 'FRE'
            GROUP BY c.prodotto
        """)
        con.execute("""
            UPDATE _base SET REGISTA_NOME = r.nome, REGISTA_COGN = r.cognome
            FROM _regista r WHERE _base.RIFER = r.prodotto
        """)

        # ------------------------------------------------------------------ #
        # ATTORI (3 principali, ruolo C001)                                   #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _attori AS
            SELECT prodotto,
                MAX(CASE WHEN rn=1 THEN COALESCE(nome,'') || ' ' || cognome END) AS a1,
                MAX(CASE WHEN rn=2 THEN COALESCE(nome,'') || ' ' || cognome END) AS a2,
                MAX(CASE WHEN rn=3 THEN COALESCE(nome,'') || ' ' || cognome END) AS a3
            FROM (
                SELECT prodotto, nome, cognome,
                    ROW_NUMBER() OVER (PARTITION BY prodotto ORDER BY progr_cast) AS rn
                FROM '{pq['cast']}'
                WHERE ruolo = 'C001'
            ) t
            WHERE rn <= 3
            GROUP BY prodotto
        """)
        con.execute("""
            UPDATE _base SET ATTORE1=a.a1, ATTORE2=a.a2, ATTORE3=a.a3
            FROM _attori a WHERE _base.RIFER = a.prodotto
        """)

        # ------------------------------------------------------------------ #
        # BOXOFFICE (tipo_programmazione='D')                                 #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _box AS
            -- VBA logic: prima uscita (MIN data_debutto) per data/stagione/distributore;
            -- SUM(incasso/spettatori) solo per l'anno solare della prima uscita.
            WITH first_year AS (
                -- first_date e first_year dai soli row 'D' (debut)
                SELECT prodotto, MIN(data_debutto) AS first_date,
                    CAST(LEFT(MIN(data_debutto), 4) AS INTEGER) AS first_year
                FROM '{pq['boxoffice']}'
                WHERE tipo_programmazione = 'D'
                GROUP BY prodotto
            )
            SELECT
                f.prodotto,
                f.first_date                                                                      AS bdata,
                FIRST(b.stagione     ORDER BY b.data_debutto ASC)                                 AS bstagione,
                NULLIF(SUM(CASE WHEN CAST(LEFT(b.data_debutto, 4) AS INTEGER) = f.first_year THEN b.incasso     ELSE 0 END), 0) AS bincasso,
                NULLIF(SUM(CASE WHEN CAST(LEFT(b.data_debutto, 4) AS INTEGER) = f.first_year THEN b.spettatori  ELSE 0 END), 0) AS bspett,
                FIRST(b.distributore ORDER BY b.data_debutto ASC)                                 AS bdistr
            FROM '{pq['boxoffice']}' b
            JOIN first_year f ON b.prodotto = f.prodotto
            -- SUM su TUTTI i tipo_programmazione nell'anno di prima uscita
            GROUP BY f.prodotto, f.first_date
        """)
        con.execute("""
            UPDATE _base
            SET BOX_DATA=b.bdata, BOX_STAGIONE=b.bstagione,
                BOX_INCASSO=b.bincasso,
                BOX_SPETTATORI=b.bspett, BOX_DISTRIBUTORE=b.bdistr
            FROM _box b WHERE _base.RIFER = b.prodotto
        """)

        # ------------------------------------------------------------------ #
        # IMDB                                                                 #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _imdb AS
            SELECT codice AS rifer, FIRST(riferimento_imdb) AS imdb_cod
            FROM '{pq['imdb']}'
            WHERE riferimento_imdb LIKE 'tt%'
            GROUP BY codice
        """)
        con.execute("""
            UPDATE _base SET IMDB_CODICE = i.imdb_cod
            FROM _imdb i WHERE _base.RIFER = i.rifer
        """)

        # ------------------------------------------------------------------ #
        # SCELTE RETE                                                          #
        # Logica: per ogni (prodotto, tipologia) prende la stagione più       #
        # recente, mappa rete→codice e slot→abbreviazione VBA, concatena     #
        # con ", " ordinato per priorità rete (flagship prima).               #
        # ------------------------------------------------------------------ #
        con.execute("""
            CREATE TABLE _rete_prio (rete_code VARCHAR, prio INTEGER);
            INSERT INTO _rete_prio VALUES
                ('C5',1),('I1',2),('R4',3),('L5',4),('IR',5),
                ('C34',6),('C27',7),('I2',8),('20',9),('FO',10),('TC',11);
        """)
        # Stagione corrente: stessa logica VBA (cutoff 10 settembre)
        _today = datetime.date.today()
        _cutoff = datetime.date(_today.year, 9, 10)
        if _today > _cutoff:
            _current_season = f"{_today.year}/{_today.year + 1}"
        else:
            _current_season = f"{_today.year - 1}/{_today.year}"

        con.execute(f"""
            CREATE TABLE _scelte AS
            WITH max_stag AS (
                SELECT prodotto, tipologia, '{_current_season}' AS max_s
                FROM '{pq['scelte_rete']}'
                WHERE stagione = '{_current_season}'
                GROUP BY prodotto, tipologia
            ),
            mapped AS (
                SELECT DISTINCT
                    s.prodotto, s.tipologia,
                    CASE s.rete
                        WHEN 'CANALE 5'      THEN 'C5'
                        WHEN 'ITALIA 1'      THEN 'I1'
                        WHEN 'RETEQUATTRO'   THEN 'R4'
                        WHEN 'LA 5'          THEN 'L5'
                        WHEN 'IRIS'          THEN 'IR'
                        WHEN 'CINE34'        THEN 'C34'
                        WHEN 'TWENTYSEVEN'   THEN 'C27'
                        WHEN 'ITALIA 2'      THEN 'I2'
                        WHEN '20'            THEN '20'
                        WHEN 'FOCUS'         THEN 'FO'
                        WHEN 'TOP CRIME'     THEN 'TC'
                        ELSE NULL
                    END AS rc,
                    CASE s.slot
                        WHEN 'NOTTE'                       THEN 'NO'
                        WHEN 'PRIMETIME GARANZIA'          THEN 'PT G'
                        WHEN 'MATTINA'                     THEN 'MA'
                        WHEN 'POMERIGGIO'                  THEN 'PO'
                        WHEN 'SECONDA SERATA GARANZIA'     THEN 'SS G'
                        WHEN 'PRIMETIME ESTATE'            THEN 'PT E'
                        WHEN 'SECONDA SERATA ESTATE'       THEN 'SS E'
                        WHEN 'ACCESS PRIMETIME GARANZIA'   THEN 'AP G'
                        WHEN 'PRIMETIME STRENNE'           THEN 'PT S'
                        WHEN 'SECONDA SERATA STRENNE'      THEN 'SS S'
                        WHEN 'ACCESS PRIMETIME STRENNE'    THEN 'AP S'
                        WHEN 'ACCESS PRIMETIME ESTATE'     THEN 'AP E'
                        ELSE NULL
                    END AS sc
                FROM '{pq['scelte_rete']}' s
                JOIN max_stag m
                    ON s.prodotto = m.prodotto
                   AND s.tipologia = m.tipologia
                   AND s.stagione = m.max_s
                WHERE s.rete IS NOT NULL AND s.rete != 'None'
                  AND s.slot IS NOT NULL AND s.slot != 'None'
            )
            SELECT
                m.prodotto, m.tipologia,
                STRING_AGG(m.rc || ' ' || m.sc, ', '
                    ORDER BY COALESCE(p.prio, 99)) AS sc_rete
            FROM mapped m
            LEFT JOIN _rete_prio p ON m.rc = p.rete_code
            WHERE m.rc IS NOT NULL AND m.sc IS NOT NULL
            GROUP BY m.prodotto, m.tipologia
        """)
        con.execute("""
            UPDATE _base SET SC_RETE_SINTESI = s.sc_rete
            FROM _scelte s
            WHERE _base.RIFER = s.prodotto AND _base.TIPOL_RAW = s.tipologia
        """)

        # ------------------------------------------------------------------ #
        # EMESSO TIPIZZATO (join con EMESSO_TYFX0160)                         #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _em AS
            SELECT e.prodotto, e.rete, e.data_emissione, e.ora_inizio,
                   e.fascia, e.audience, e.share,
                   t.ICR_MONDO, t.ICR_DECOD_RETE_BREVE
            FROM '{pq['emesso']}' e
            JOIN read_csv('{tyfx}', delim=';', header=true) t ON e.rete = t.CCRETETEL
            WHERE t.ICR_MONDO IS NOT NULL AND t.ICR_MONDO != ''
        """)

        # Helper: genera prima/ultima emissione per un insieme di ICR_MONDO
        # VBA logic per PRIMA (MIN): esclude fascia MA (rerun mattutini non presenti nell'EMESSO VBA);
        #   fallback a tutti i broadcast se non esistono emissioni non-MA.
        # Per ULTIMA (MAX): usa MAX(data||time) originale.
        def _set_em(tag: str, mondi: list, agg: str) -> None:
            mondi_sql = ', '.join(f"'{m}'" for m in mondi)
            if agg == 'MIN':
                # Step 1: TRUE MIN date (include MA)
                # Step 2: on that date, prefer non-MA; fallback to MA (tiebreaker)
                con.execute(f"""
                    CREATE OR REPLACE TABLE _em_{tag} AS
                    SELECT e.prodotto, e.rete, e.ICR_DECOD_RETE_BREVE,
                           e.data_emissione, e.ora_inizio, e.fascia,
                           e.audience, e.share
                    FROM _em e
                    JOIN (
                        SELECT prodotto, MIN(data_emissione) AS target_date
                        FROM _em
                        WHERE ICR_MONDO IN ({mondi_sql})
                        GROUP BY prodotto
                    ) g ON e.prodotto = g.prodotto AND e.data_emissione = g.target_date
                    WHERE e.ICR_MONDO IN ({mondi_sql})
                    QUALIFY ROW_NUMBER() OVER (
                        PARTITION BY e.prodotto
                        ORDER BY CASE WHEN e.fascia = 'MA' THEN 1 ELSE 0 END ASC,
                                 e.ora_inizio ASC
                    ) = 1
                """)
            else:
                con.execute(f"""
                    CREATE OR REPLACE TABLE _em_{tag} AS
                    SELECT e.prodotto, e.rete, e.ICR_DECOD_RETE_BREVE,
                           e.data_emissione, e.ora_inizio, e.fascia,
                           e.audience, e.share
                    FROM _em e
                    JOIN (
                        SELECT prodotto,
                               MAX(data_emissione || LPAD(CAST(ora_inizio AS VARCHAR), 6, '0')) AS key_val
                        FROM _em
                        WHERE ICR_MONDO IN ({mondi_sql})
                        GROUP BY prodotto
                    ) g ON e.prodotto = g.prodotto
                      AND (e.data_emissione || LPAD(CAST(e.ora_inizio AS VARCHAR), 6, '0')) = g.key_val
                    WHERE e.ICR_MONDO IN ({mondi_sql})
                    QUALIFY ROW_NUMBER() OVER (PARTITION BY e.prodotto ORDER BY e.ora_inizio DESC) = 1
                """)
            # Colonne in _base: EM_{TAG}_RETE, _DATA, _HI, _FASCIA [, _AUD, _SHARE]
            has_aud = tag in ('PG', 'UG', 'PT', 'UT')
            aud_set = ', "EM_{tag}_AUD" = e.audience, "EM_{tag}_SHARE" = e.share' if has_aud else ""
            con.execute(f"""
                UPDATE _base b
                SET "EM_{tag}_RETE"  = e.ICR_DECOD_RETE_BREVE,
                    "EM_{tag}_DATA"  = e.data_emissione,
                    "EM_{tag}_HI"    = e.ora_inizio,
                    "EM_{tag}_FASCIA"= e.fascia
                    {aud_set.replace('{tag}', tag)}
                FROM _em_{tag} e
                WHERE b.RIFER = e.prodotto
            """)

        # Prima/Ultima generaliste (CONCOR + MED_GEN)
        _set_em('PG', ['CONCOR', 'MED_GEN'], 'MIN')
        _set_em('UG', ['CONCOR', 'MED_GEN'], 'MAX')
        # Prima/Ultima tematiche (MED_TEM + CON_TEM)
        _set_em('PT', ['MED_TEM', 'CON_TEM'], 'MIN')
        _set_em('UT', ['MED_TEM', 'CON_TEM'], 'MAX')
        # Prima/Ultima premium — VBA eliminava i record Premium come obsoleti → sempre NULL
        # _set_em('PP', ['PREMIUM'], 'MIN')
        # _set_em('UP', ['PREMIUM'], 'MAX')
        # Prima/Ultima SKY+FOX
        _set_em('PSKY', ['SKY', 'FOX'], 'MIN')
        _set_em('USKY', ['SKY', 'FOX'], 'MAX')

        # ------------------------------------------------------------------ #
        # DIRITTI — campi DF_* e DTVOD_*                                      #
        # ------------------------------------------------------------------ #
        con.execute(f"""
            CREATE TABLE _dir AS
            SELECT * REPLACE(
                IF(decr      = '0001-01-01', '1001-01-01', decr)      AS decr,
                IF(scad      = '0001-01-01', '1001-01-01', scad)      AS scad,
                IF(decr_inib = '0001-01-01', '1001-01-01', decr_inib) AS decr_inib,
                IF(scad_inib = '0001-01-01', '1001-01-01', scad_inib) AS scad_inib
            )
            FROM '{pq['diritti']}'
        """)

        # Diritti free analogico in essere
        con.execute(f"""
            CREATE TABLE _df_inbeing AS
            SELECT prod,
                FIRST(fr_rr ORDER BY contratto, riga, situazione)                                         AS fr_rr,
                FIRST(ragsoc_distr ORDER BY contratto, riga, situazione)                                   AS ragsoc,
                FIRST(perc ORDER BY contratto, riga, situazione)                                           AS perc,
                FIRST(pass_cons_tot ORDER BY contratto, riga, situazione)                                  AS pct,
                FIRST(pass_eff_tot ORDER BY contratto, riga, situazione)                                   AS pet,
                COALESCE(CAST(TRY_CAST(FIRST(causale ORDER BY contratto, riga, situazione) AS INTEGER) AS VARCHAR),
                         FIRST(causale ORDER BY contratto, riga, situazione))                              AS causale,
                FIRST(decr ORDER BY contratto, riga, situazione)                                           AS decr,
                FIRST(scad ORDER BY contratto, riga, situazione)                                           AS scad,
                FIRST(flag_inib ORDER BY contratto, riga, situazione)                                      AS flag_inib,
                FIRST("DECR_WIN_1" ORDER BY contratto, riga, situazione)                                   AS dw1,
                FIRST("SCAD_WIN_1" ORDER BY contratto, riga, situazione)                                   AS sw1,
                FIRST("DECR_WIN_2" ORDER BY contratto, riga, situazione)                                   AS dw2,
                FIRST("SCAD_WIN_2" ORDER BY contratto, riga, situazione)                                   AS sw2,
                FIRST("DECR_WIN_3" ORDER BY contratto, riga, situazione)                                   AS dw3,
                COUNT(*) AS n_rows
            FROM _dir
            WHERE UPPER(tipo_diritto) = 'FREE'
              AND UPPER(piattaforma) = 'ANALOGICO'
              AND decr <= '{today}'
              AND scad >= '{today}'
            GROUP BY prod
        """)
        con.execute("""
            UPDATE _base b
            SET DF_FR_RR        = d.fr_rr,
                DF_RAGSOC_DISTR = d.ragsoc,
                DF_PERC         = CAST(d.perc AS VARCHAR),
                DF_PASS_CONS_TOT= CAST(d.pct AS VARCHAR),
                DF_PASS_EFF_TOT = CAST(d.pet AS VARCHAR),
                DF_CAUSALE      = d.causale,
                DF_DECR         = d.decr,
                DF_SCAD         = d.scad,
                DF_ICR_FLAG     = CASE WHEN d.flag_inib = 'S' THEN 'I' ELSE NULL END,
                DF_DECR_WIN_1   = d.dw1,
                DF_SCAD_WIN_1   = d.sw1,
                DF_DECR_WIN_2   = d.dw2,
                DF_SCAD_WIN_2   = d.sw2,
                DF_WIN_ALTRE    = CASE WHEN d.dw3 IS NOT NULL THEN 'W' ELSE NULL END
            FROM _df_inbeing d WHERE b.RIFER = d.prod
        """)
        # Flag '+' se multipli contratti in essere
        con.execute("""
            UPDATE _base SET DF_ICR_FLAG = COALESCE(DF_ICR_FLAG, '') || '+'
            WHERE RIFER IN (SELECT prod FROM _df_inbeing WHERE n_rows > 1)
        """)

        # Diritti free analogico futuri (DECR > oggi)
        con.execute(f"""
            CREATE TABLE _df_future AS
            SELECT prod,
                FIRST(fr_rr ORDER BY decr, contratto, riga)                                          AS fr_rr,
                FIRST(ragsoc_distr ORDER BY decr, contratto, riga)                                   AS ragsoc,
                FIRST(decr ORDER BY decr, contratto, riga)                                           AS decr,
                FIRST(scad ORDER BY decr, contratto, riga)                                           AS scad
            FROM _dir
            WHERE UPPER(tipo_diritto) = 'FREE'
              AND UPPER(piattaforma) = 'ANALOGICO'
              AND decr > '{today}'
            GROUP BY prod
        """)
        con.execute("""
            UPDATE _base b
            SET DF_FUT_FR_RR    = d.fr_rr,
                DF_FUT_DECR     = d.decr,
                DF_FUT_SCAD     = d.scad,
                DF_RAGSOC_DISTR = d.ragsoc
            FROM _df_future d WHERE b.RIFER = d.prod
        """)

        # Diritti free analogico scaduti (SCAD < oggi, solo se nessun in essere)
        con.execute(f"""
            CREATE TABLE _df_expired AS
            SELECT prod,
                FIRST(fr_rr ORDER BY scad DESC, contratto, riga)        AS fr_rr,
                FIRST(ragsoc_distr ORDER BY scad DESC, contratto, riga)  AS ragsoc,
                FIRST(decr ORDER BY scad DESC, contratto, riga)          AS decr,
                FIRST(scad ORDER BY scad DESC, contratto, riga)          AS scad,
                FIRST(perc ORDER BY scad DESC, contratto, riga)          AS perc,
                FIRST(pass_cons_tot ORDER BY scad DESC, contratto, riga) AS pct,
                FIRST(pass_eff_tot ORDER BY scad DESC, contratto, riga)  AS pet,
                COALESCE(CAST(TRY_CAST(FIRST(causale ORDER BY scad DESC, contratto, riga) AS INTEGER) AS VARCHAR),
                         FIRST(causale ORDER BY scad DESC, contratto, riga))       AS causale
            FROM _dir
            WHERE UPPER(tipo_diritto) = 'FREE'
              AND UPPER(piattaforma) = 'ANALOGICO'
              AND scad < '{today}'
            GROUP BY prod
        """)
        con.execute("""
            UPDATE _base b
            SET DF_FR_RR        = d.fr_rr,
                DF_ICR_FLAG     = 'S',
                DF_DECR         = d.decr,
                DF_SCAD         = d.scad,
                DF_RAGSOC_DISTR = d.ragsoc,
                DF_PERC         = CAST(d.perc AS VARCHAR),
                DF_PASS_CONS_TOT= CAST(d.pct AS VARCHAR),
                DF_PASS_EFF_TOT = CAST(d.pet AS VARCHAR),
                DF_CAUSALE      = COALESCE(CAST(TRY_CAST(d.causale AS INTEGER) AS VARCHAR), d.causale)
            FROM _df_expired d WHERE b.RIFER = d.prod AND b.DF_DECR IS NULL
        """)

        # TVOD — in essere, futuro, scaduto
        for tvod_filter, tvod_label in [
            (f"decr <= '{today}' AND scad >= '{today}'", 'inbeing'),
            (f"decr > '{today}'",                        'future'),
            (f"scad < '{today}'",                        'expired'),
        ]:
            con.execute(f"""
                CREATE OR REPLACE TABLE _tvod_{tvod_label} AS
                SELECT prod,
                    FIRST(decr ORDER BY decr DESC, contratto DESC, riga DESC)                            AS decr,
                    FIRST(scad ORDER BY decr DESC, contratto DESC, riga DESC)                             AS scad,
                    FIRST(fr_rr ORDER BY decr DESC, contratto DESC, riga DESC)                           AS fr_rr,
                    FIRST(ragsoc_distr ORDER BY decr DESC, contratto DESC, riga DESC)                    AS ragsoc
                FROM _dir
                WHERE UPPER(tipo_diritto) = 'TVOD'
                  AND {tvod_filter}
                GROUP BY prod
            """)
            con.execute(f"""
                UPDATE _base b
                SET DTVOD_DECR         = d.decr,
                    DTVOD_SCAD         = d.scad,
                    DTVOD_FR_RR        = d.fr_rr,
                    DTVOD_RAGSOC_DISTR = d.ragsoc
                FROM _tvod_{tvod_label} d
                WHERE b.RIFER = d.prod AND b.DTVOD_DECR IS NULL
            """)

        # Flag 'T': prodotti con solo DVB-T (nessun Free Analogico)
        con.execute(f"""
            CREATE TABLE _dvbt_only AS
            SELECT DISTINCT d.prod
            FROM _dir d
            WHERE UPPER(d.tipo_diritto) = 'FREE'
              AND UPPER(d.piattaforma) = 'DVB-T'
              AND NOT EXISTS (
                  SELECT 1 FROM _dir a
                  WHERE UPPER(a.tipo_diritto) = 'FREE'
                    AND UPPER(a.piattaforma) = 'ANALOGICO'
                    AND a.prod = d.prod AND a.decr = d.decr AND a.scad = d.scad
              )
        """)
        # Popola DF_* dai record DVB-T. VBA inseriva DVB-T come 'analogico' PRIMA della logica
        # DF_*, quindi i DVB-T in-being sovrascrivevano eventuali Analogico scaduti già settati.
        # Replica: fasi in-being e future fanno UPDATE incondizionale (override Analogico);
        # fase expired è condizionale (DF_DECR IS NULL) e setta solo chi non ha diritti attivi.
        for dvbt_filter, dvbt_order, dvbt_override, skip_analog_twin, is_future in [
            (f"decr <= '{today}' AND scad >= '{today}'", "contratto, riga, situazione", True,  True,  False),
            (f"decr > '{today}'",                        "decr, contratto, riga",       True,  True,  True),
            (f"scad < '{today}'",                        "scad DESC, contratto, riga",  False, False, False),
        ]:
            # In-being/future: skip DVB-T rows that already have a matching Analogico (same decr+scad).
            # Those are covered by the Analogico phases; including them here would wrongly set 'T'.
            no_twin = """
                  AND NOT EXISTS (
                      SELECT 1 FROM _dir analog
                      WHERE UPPER(analog.piattaforma) = 'ANALOGICO'
                        AND analog.prod = _dir.prod
                        AND analog.decr = _dir.decr AND analog.scad = _dir.scad
                  )""" if skip_analog_twin else ""
            con.execute(f"""
                CREATE OR REPLACE TABLE _dvbt_df AS
                SELECT prod,
                    FIRST(fr_rr       ORDER BY {dvbt_order}) AS fr_rr,
                    FIRST(ragsoc_distr ORDER BY {dvbt_order}) AS ragsoc,
                    FIRST(decr        ORDER BY {dvbt_order}) AS decr,
                    FIRST(scad        ORDER BY {dvbt_order}) AS scad,
                    FIRST(perc        ORDER BY {dvbt_order}) AS perc,
                    FIRST(pass_cons_tot ORDER BY {dvbt_order}) AS pct,
                    FIRST(pass_eff_tot  ORDER BY {dvbt_order}) AS pet,
                    COALESCE(CAST(TRY_CAST(FIRST(causale ORDER BY {dvbt_order}) AS INTEGER) AS VARCHAR),
                             FIRST(causale ORDER BY {dvbt_order})) AS causale
                FROM _dir
                WHERE UPPER(tipo_diritto) = 'FREE'
                  AND UPPER(piattaforma) = 'DVB-T'
                  AND prod IN (SELECT prod FROM _dvbt_only)
                  AND {dvbt_filter}{no_twin}
                GROUP BY prod
            """)
            where_cond = "b.RIFER = d.prod" if dvbt_override else "b.RIFER = d.prod AND b.DF_DECR IS NULL"
            if is_future:
                # DVB-T futuro: popola DF_FUT_* (non DF_DECR/DF_SCAD), come la fase Analogico futura
                con.execute(f"""
                    UPDATE _base b
                    SET DF_ICR_FLAG     = 'T',
                        DF_FUT_FR_RR    = d.fr_rr,
                        DF_FUT_DECR     = d.decr,
                        DF_FUT_SCAD     = d.scad,
                        DF_RAGSOC_DISTR = d.ragsoc
                    FROM _dvbt_df d
                    WHERE {where_cond}
                """)
                continue
            flag_set   = "DF_ICR_FLAG = 'T'," if dvbt_override else ""
            con.execute(f"""
                UPDATE _base b
                SET {flag_set}
                    DF_FR_RR        = d.fr_rr,
                    DF_RAGSOC_DISTR = d.ragsoc,
                    DF_DECR         = d.decr,
                    DF_SCAD         = d.scad,
                    DF_PERC         = CAST(d.perc AS VARCHAR),
                    DF_PASS_EFF_TOT = CAST(d.pet AS VARCHAR),
                    DF_CAUSALE      = d.causale
                FROM _dvbt_df d
                WHERE {where_cond}
            """)

        # Expired DVB-T non già gestiti → 'S'; resto ancora NULL → 'T'
        con.execute(f"""
            UPDATE _base SET DF_ICR_FLAG = 'S'
            FROM _dvbt_only x
            WHERE _base.RIFER = x.prod
              AND _base.DF_ICR_FLAG IS NULL
              AND _base.DF_SCAD < '{today}'
        """)
        con.execute(f"""
            UPDATE _base SET DF_ICR_FLAG = 'T'
            FROM _dvbt_only x
            WHERE _base.RIFER = x.prod
              AND _base.DF_ICR_FLAG IS NULL
              AND NOT EXISTS (
                  SELECT 1 FROM _dir a
                  WHERE UPPER(a.tipo_diritto) = 'FREE'
                    AND UPPER(a.piattaforma) = 'ANALOGICO'
                    AND a.prod = x.prod
                    AND a.decr <= '{today}' AND a.scad >= '{today}'
              )
        """)

        # ------------------------------------------------------------------ #
        # Risultato finale                                                     #
        # ------------------------------------------------------------------ #
        df_anagr = con.execute("SELECT * EXCLUDE (TIPOL_RAW) FROM _base").df()

        # EM_PP_RETE / EM_UP_RETE: Access field is VARCHAR(2).
        # ICR_DECOD_RETE_BREVE can be 'JOI' (3 chars) for PREMIUM networks →
        # truncate explicitly so Access receives exactly 2 chars.
        for _col in ('EM_PP_RETE', 'EM_UP_RETE'):
            if _col in df_anagr.columns:
                df_anagr[_col] = df_anagr[_col].str[:2]

        # EDIZIONI: tutte le edizioni, solo tipologie valide, TIPOL mappato a codice VBA
        ediz_tip_sql = tip_sql  # stesse 10 tipologie di _base
        ediz_tipol_case = tipol_case
        df_ediz = con.execute(f"""
            SELECT
                prodotto    AS RIFER,
                edizione    AS ED,
                CASE tipologia {ediz_tipol_case} END AS TIPOL,
                FIRST(durata) AS DUR,
                FIRST(vm)   AS TARGET
            FROM '{pq['prodotti']}'
            WHERE tipologia IN ({ediz_tip_sql})
            GROUP BY prodotto, edizione, tipologia
        """).df()

        # SUPERSERIE
        df_super = con.execute(f"""
            SELECT DISTINCT
                superserie_descr AS SUPERSERIE,
                prodotto         AS RIFER,
                superserie_id    AS COD_SUPERSERIE
            FROM '{pq['prodotti']}'
            WHERE superserie_id IS NOT NULL
        """).df()

    return df_anagr, df_ediz, df_super