"""
CUSTOM_DIRITTI — diritti Free/Analogico + workaround DVB-T + s_motivo_inibiz.

Sorgenti:
- diritti.parquet
- df_emesso: DataFrame CUSTOM_EMESSO già costruito (per fun_disponibilita_netta)

Logica:
1. Filtra Free/Analogico + DVB-T senza corrispondente Analogico
2. Calcola calMotivoInibiz per anni 2015..oggi+5
3. Calcola CalFrRr (F/R da FR_RR + storia)
"""
import datetime
from pathlib import Path

import duckdb
import pandas as pd


_TIPOLOGIE_INIBIZ = {'FILM', 'MINISERIE', 'SIT COM', 'TELEFILM', 'TV MOVIE', 'SOAP', 'TELENOVELAS'}


def _fun_disponibilita_netta_fast(
    prod_rows: pd.DataFrame,
    emesso_dates: set,
    dal: datetime.date,
    al: datetime.date,
) -> str:
    """
    Calcola stato disponibilità per un singolo PROD in intervallo annuale.
    prod_rows: righe già filtrate per PROD e che overlappano [dal, al].
    emesso_dates: set di date string 'YYYY-MM-DD' del prodotto in MED_GEN/MED_TEM.
    Restituisce "" se disponibile, altrimenti il motivo di inibizione.
    """
    if prod_rows.empty:
        return ""

    nr_record = len(prod_rows)
    var_perc_totale = 0.0
    var_inibito = ""
    var_gg_disp = 0
    dal_str = dal.isoformat()
    al_str  = al.isoformat()

    for _, row in prod_rows.iterrows():
        row_inibito = ""
        pct = row.get('PASS_CONS_TOT')
        pef = row.get('PASS_EFF_TOT')
        if pct is not None and pef is not None:
            try:
                if not (pd.isna(pct) or pd.isna(pef)) and float(pct) - float(pef) <= 0:
                    row_inibito = "passaggi esauriti"
            except Exception:
                pass
        if row.get('FLAG_INIB') == 'S':
            row_inibito = "inibito"

        var_inibito = row_inibito
        var_perc_totale += float(row.get('PERC') or 0)

        # Overlap O(1) invece del loop giornaliero
        decr = row['DECR_D']
        scad = row['SCAD_D']
        if decr and scad:
            overlap_start = max(decr, dal)
            overlap_end   = min(scad, al)
            var_gg_disp = max(0, (overlap_end - overlap_start).days + 1)
        else:
            var_gg_disp = 0

        if nr_record > 1 and row_inibito == "" and var_perc_totale >= 100 and var_gg_disp > 30:
            return ""

    if var_perc_totale < 100:
        var_inibito = "<100%"
    if var_gg_disp < 30:
        var_inibito = "<30gg"

    if var_inibito and emesso_dates:
        for d in emesso_dates:
            if dal_str <= d <= al_str:
                return ""

    return var_inibito


def _s_motivo_inibiz(df: pd.DataFrame, df_emesso: pd.DataFrame) -> pd.DataFrame:
    """
    Calcola calMotivoInibiz e CalFrRr su df CUSTOM_DIRITTI.
    Versione ottimizzata: pre-index by PROD, overlap O(1), emesso pre-indexed.
    """
    today = datetime.date.today()
    df = df.copy()

    # Converti date per uso interno
    df['DECR_D'] = pd.to_datetime(df['DECR'], errors='coerce').dt.date
    df['SCAD_D'] = pd.to_datetime(df['SCAD'], errors='coerce').dt.date

    df['calMotivoInibiz'] = ''
    df['CalFrRr'] = ''

    # ------------------------------------------------------------------ #
    # Pre-index: dizionario PROD → gruppo di righe                        #
    # ------------------------------------------------------------------ #
    prod_index: dict = {}
    for prod, grp in df.groupby('PROD'):
        prod_index[prod] = grp

    # ------------------------------------------------------------------ #
    # Pre-index emesso: CCPROD → set di date 'YYYY-MM-DD' (MED_GEN/TEM)  #
    # ------------------------------------------------------------------ #
    emesso_index: dict = {}
    if df_emesso is not None and not df_emesso.empty:
        try:
            em_fil = df_emesso[df_emesso['ICR_MONDO'].isin(['MED_GEN', 'MED_TEM'])]
            for prod, grp in em_fil.groupby('CCPROD'):
                emesso_index[int(prod)] = set(grp['DCDATGIO'].astype(str).str[:10])
        except Exception:
            pass

    mask_tipol = df['TIPOLOGIA'].str.upper().isin(_TIPOLOGIE_INIBIZ)

    for yn in range(2015, today.year + 6):
        dal = datetime.date(yn, 1, 1)
        al  = datetime.date(yn, 12, 31)

        # PRODs attivi in questo anno con tipologia valida
        year_mask = (
            mask_tipol &
            df['DECR_D'].notna() & df['SCAD_D'].notna() &
            (df['DECR_D'] <= al) & (df['SCAD_D'] >= dal)
        )
        active_prods = df.loc[year_mask, 'PROD'].unique()
        if len(active_prods) == 0:
            continue

        for prod in active_prods:
            # Righe per questo prod che overlappano l'anno (da pre-index)
            all_rows = prod_index.get(prod)
            if all_rows is None:
                continue
            prod_rows = all_rows[
                all_rows['DECR_D'].notna() & all_rows['SCAD_D'].notna() &
                (all_rows['DECR_D'] <= al) & (all_rows['SCAD_D'] >= dal)
            ]
            if prod_rows.empty:
                continue

            inibizione = _fun_disponibilita_netta_fast(
                prod_rows, emesso_index.get(prod, set()), dal, al
            )

            if inibizione:
                prod_year_idxs = df.index[year_mask & (df['PROD'] == prod)]
                for idx in prod_year_idxs:
                    cal = df.at[idx, 'calMotivoInibiz'] or ''
                    if str(yn) not in cal:
                        df.at[idx, 'calMotivoInibiz'] = cal + f"|{inibizione[:4]}_{yn}|"

    # ------------------------------------------------------------------ #
    # CalFrRr: prima passata                                               #
    # ------------------------------------------------------------------ #
    df['CalFrRr'] = df['FR_RR'].apply(lambda x: 'R' if x == 'R' else 'F')

    # Seconda passata: se esiste riga con DECR antecedente → R
    for idx, row in df[df['CalFrRr'] == 'F'].iterrows():
        prod = row['PROD']
        decr_y = row['DECR_D'].year if row['DECR_D'] else None
        if decr_y and decr_y != 1001:
            all_rows = prod_index.get(prod)
            if all_rows is not None:
                earlier = all_rows[
                    all_rows['DECR_D'].notna() &
                    (all_rows['DECR_D'].apply(lambda d: d.year if d else None) < decr_y) &
                    (all_rows['DECR_D'].apply(lambda d: d.year if d else None) != 1001)
                ]
                if not earlier.empty:
                    df.at[idx, 'CalFrRr'] = 'R'

    df.drop(columns=['DECR_D', 'SCAD_D'], inplace=True)
    return df


def build(parquet_dir: Path, df_emesso: pd.DataFrame) -> pd.DataFrame:
    dir_pq = str(parquet_dir / 'diritti.parquet').replace('\\', '/')

    win_decr = ', '.join(f'"DECR_WIN_{i}"' for i in range(1, 10))
    win_scad = ', '.join(f'"SCAD_WIN_{i}"' for i in range(1, 10))

    with duckdb.connect() as con:
        df = con.execute(f"""
            WITH _raw 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 '{dir_pq}'
            ),
            free_ana AS (
                SELECT
                    tipo_diritto        AS TIPO_DIRITTO,
                    piattaforma         AS PIATTAFORMA,
                    tipologia           AS TIPOLOGIA,
                    prod                AS PROD,
                    ediz                AS EDIZ,
                    d_t                 AS D_T,
                    ragsoc_distr        AS RAGSOC_DISTR,
                    perc                AS PERC,
                    pass_cons_tot       AS PASS_CONS_TOT,
                    pass_eff_tot        AS PASS_EFF_TOT,
                    decr                                                           AS DECR,
                    scad                AS SCAD,
                    causale             AS CAUSALE,
                    flag_inib           AS FLAG_INIB,
                    decr_inib           AS DECR_INIB,
                    scad_inib           AS SCAD_INIB,
                    fr_rr               AS FR_RR,
                    contratto           AS CONTRATTO,
                    riga                AS RIGA,
                    situazione          AS SITUAZIONE,
                    {win_decr},
                    {win_scad}
                FROM _raw
                WHERE UPPER(tipo_diritto) = 'FREE'
                  AND UPPER(piattaforma) = 'ANALOGICO'
            ),
            dvbt_only AS (
                SELECT
                    d.tipo_diritto        AS TIPO_DIRITTO,
                    'Analogico'           AS PIATTAFORMA,
                    d.tipologia           AS TIPOLOGIA,
                    d.prod                AS PROD,
                    d.ediz                AS EDIZ,
                    d.d_t                 AS D_T,
                    d.ragsoc_distr        AS RAGSOC_DISTR,
                    d.perc                AS PERC,
                    d.pass_cons_tot       AS PASS_CONS_TOT,
                    d.pass_eff_tot        AS PASS_EFF_TOT,
                    d.decr                                                               AS DECR,
                    d.scad                                                               AS SCAD,
                    d.causale             AS CAUSALE,
                    d.flag_inib           AS FLAG_INIB,
                    d.decr_inib           AS DECR_INIB,
                    d.scad_inib           AS SCAD_INIB,
                    d.fr_rr               AS FR_RR,
                    d.contratto           AS CONTRATTO,
                    d.riga                AS RIGA,
                    d.situazione          AS SITUAZIONE,
                    {win_decr},
                    {win_scad}
                FROM _raw d
                WHERE UPPER(d.tipo_diritto) = 'FREE'
                  AND UPPER(d.piattaforma) = 'DVB-T'
                  AND NOT EXISTS (
                      SELECT 1 FROM _raw 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
                  )
            )
            SELECT * FROM free_ana
            UNION ALL
            SELECT * FROM dvbt_only
        """).df()

    df = _s_motivo_inibiz(df, df_emesso)
    return df