"""
BOXOFFICE — film con uscita cinema dal 2010.

Sorgenti:
- df_anagr: DataFrame CUSTOM_ANAGR_ORIZ già costruito
  (contiene BOX_DATA, EM_PG_*/EM_PT_* per _fun_prima_tv)
- df_diritti: non usato (mantenuto per compatibilità firma)

Colonne output:
    RIFER, TI, TO, TIPOL, ANNO, PAESE, VEG,
    BOX_DATA, BOX_INCASSO, BOX_SPETTATORI, BOX_DISTRIBUTORE,
    STAGIONE, PRIMA_TV_LIBERA, RETE_FREE
"""
import pandas as pd


# Mapping ICR_DECOD_RETE_BREVE → broadcaster (come VBA)
_RETE_TO_NETWORK = {
    # Mediaset generaliste (MED_GEN)
    'C5': 'MEDIASET', 'I1': 'MEDIASET', 'R4': 'MEDIASET',
    # Mediaset tematiche (MED_TEM)
    'B5': 'MEDIASET', 'CI34': 'MEDIASET', 'BP': 'MEDIASET', 'I2': 'MEDIASET',
    'IRIS': 'MEDIASET', '20': 'MEDIASET', '27': 'MEDIASET', 'FOC': 'MEDIASET',
    'EXTR': 'MEDIASET', 'LA': 'MEDIASET', 'TOPC': 'MEDIASET', 'LA5': 'MEDIASET',
    # Mediaset Premium (PREMIUM)
    'KD': 'MEDIASET', 'KE': 'MEDIASET', 'KG': 'MEDIASET', 'JOI': 'MEDIASET',
    'KO': 'MEDIASET', 'KP': 'MEDIASET', 'KR': 'MEDIASET', 'KS': 'MEDIASET',
    'LC': 'MEDIASET', 'LG': 'MEDIASET', 'LH': 'MEDIASET', 'LM': 'MEDIASET',
    'LR': 'MEDIASET', 'LT': 'MEDIASET', 'LV': 'MEDIASET',
    # RAI (CONCOR + CON_TEM)
    'R1': 'RAI', 'R2': 'RAI', 'R3': 'RAI',
    'RAI4': 'RAI', 'RAIM': 'RAI', 'RAIP': 'RAI',
    # LA7 group (CONCOR + CON_TEM)
    'LA7': 'LA7', 'LA7C': 'LA7', 'LA7D': 'LA7',
    # SKY (ICR_MONDO=SKY) e canali tematici SKY (CON_TEM)
    'CIEL': 'SKY', 'TV8': 'SKY',
    'AF': 'SKY', 'AX': 'SKY', 'CC': 'SKY', 'EY': 'SKY', 'FX': 'SKY',
    'JY': 'SKY', 'KN': 'SKY', 'RC': 'SKY',
    'S1': 'SKY', 'S3': 'SKY', 'S6': 'SKY', 'SA': 'SKY', 'SB': 'SKY',
    'SC': 'SKY', 'SF': 'SKY', 'SG': 'SKY', 'SH': 'SKY', 'SI': 'SKY',
    'SL': 'SKY', 'SM': 'SKY', 'SN': 'SKY', 'SP': 'SKY', 'SQ': 'SKY',
    'SR': 'SKY', 'ST': 'SKY', 'SV': 'SKY', 'SW': 'SKY', 'SY': 'SKY',
    'V1': 'SKY', 'V2': 'SKY', 'V3': 'SKY', 'V4': 'SKY', 'V5': 'SKY',
    'V6': 'SKY', 'V7': 'SKY', 'VQ': 'SKY', 'VR': 'SKY', 'VS': 'SKY',
    'VT': 'SKY', 'VU': 'SKY',
    # FOX → SKY (ora parte di Sky)
    'FB': 'SKY', 'FC': 'SKY', 'FG': 'SKY', 'FL': 'SKY', 'FO': 'SKY', 'FR': 'SKY',
    # BOING
    'BG': 'BOING',
    # DISCOVERY e canali correlati (CON_TEM)
    'DISCOV': 'DISCOVERY', 'NOVE': 'DISCOVERY', 'RT': 'DISCOVERY', 'DM': 'DISCOVERY',
    'WB': 'DISCOVERY', 'GI': 'DISCOVERY', 'POP': 'DISCOVERY',
    # PARAMOUNT
    'PA': 'PARAMOUNT',
    # SONY
    'SONY': 'SONY',
}


def _fun_prima_tv(row: pd.Series) -> str:
    """
    Costruisce stringa "RETE dd/mm/yy HH.MM" dalla prima emissione libera:
    MIN(EM_PG_DATA, EM_PT_DATA) — generalista o tematica, la più antica vince.
    """
    candidates = []
    for prefix in ('EM_PG', 'EM_PT'):
        rete = row.get(f'{prefix}_RETE')
        data = row.get(f'{prefix}_DATA')
        hi   = row.get(f'{prefix}_HI')
        if rete and data and not (pd.isna(rete) or pd.isna(data)):
            try:
                d = pd.to_datetime(data)
            except Exception:
                continue
            candidates.append((d, rete, hi))

    if not candidates:
        return ''

    d, rete, hi = min(candidates, key=lambda x: x[0])
    data_str = d.strftime('%d/%m/%y')
    ora_str = ''
    if hi and not pd.isna(hi):
        try:
            hi_int = int(float(hi))
            hh = hi_int // 10000
            mm = (hi_int % 10000) // 100
            ora_str = f'{hh:02d}.{mm:02d}'
        except Exception:
            pass
    rete_s = str(rete).strip()
    return f'{rete_s} {data_str} {ora_str}'.strip()


def _network_from_prima_tv(prima_tv: str) -> str:
    """Estrae il broadcaster da '1TV FREE' (rete = primo token prima dello spazio-data)."""
    if not prima_tv or (isinstance(prima_tv, float)):
        return ''
    rete = str(prima_tv).split(' ')[0].strip()
    return _RETE_TO_NETWORK.get(rete, '')


def _stagione(stagione_raw) -> str:
    """Restituisce la stagione cinema (formato 'YYYY/YY') dal parquet boxoffice."""
    if not stagione_raw or pd.isna(stagione_raw):
        return ''
    return str(stagione_raw).strip()


def build(df_anagr: pd.DataFrame) -> pd.DataFrame:
    """
    Costruisce la tabella BOXOFFICE filtrando df_anagr per prodotti
    con BOX_DATA >= 2010-01-01 e aggiungendo STAGIONE, PRIMA_TV_LIBERA, RETE_FREE.
    """
    if df_anagr is None or df_anagr.empty:
        return pd.DataFrame()

    # Filtra solo prodotti con uscita cinema dal 2010 (come VBA)
    df = df_anagr[df_anagr['BOX_DATA'].notna()].copy()
    df = df[df['BOX_DATA'].astype(str) >= '2010-01-01']

    if df.empty:
        return pd.DataFrame()

    # Colonne sorgente necessarie + EM_* temporanee per _fun_prima_tv
    em_cols = [c for c in df.columns if c.startswith('EM_PG_') or c.startswith('EM_PT_')]
    src_cols = ['RIFER', 'TI', 'BOX_DATA', 'BOX_STAGIONE', 'BOX_DISTRIBUTORE', 'IMDB_CODICE'] + em_cols
    df = df[[c for c in src_cols if c in df.columns]].copy()

    # Campi calcolati
    df['STAGIONE']        = df['BOX_STAGIONE'].apply(_stagione)
    df['PRIMA_TV_LIBERA'] = df.apply(_fun_prima_tv, axis=1)
    df['RETE_FREE']       = df['PRIMA_TV_LIBERA'].apply(_network_from_prima_tv)

    # Rimuovi colonne temporanee
    df = df.drop(columns=em_cols + ['BOX_STAGIONE'], errors='ignore')

    # Rinomina per corrispondere alle colonne Access
    df = df.rename(columns={
        'RIFER':           'rifer',
        'BOX_DATA':        'uscita',
        'BOX_DISTRIBUTORE':'DISTRIB BOX-OFFICE',
        'RETE_FREE':       'network free',
        'PRIMA_TV_LIBERA': '1TV FREE',
        'IMDB_CODICE':     'IMDB',
    })

    # Colonne mancanti nel master (lasciate vuote)
    df['ANNUNCIATO'] = None
    df['1TV PAY']    = None

    # Ordine colonne = ordine tabella Access
    access_cols = ['STAGIONE', 'ANNUNCIATO', 'uscita', 'TI',
                   'DISTRIB BOX-OFFICE', 'network free', '1TV FREE',
                   '1TV PAY', 'rifer', 'IMDB']
    df = df[[c for c in access_cols if c in df.columns]]

    return df.reset_index(drop=True)