"""
diritti.py — produce diritti.parquet (raw)

Logica:
- Tutte le colonne originali del CSV incluse WIN_1..WIN_9
- Nessun filtro tipo_diritto/piattaforma (dati grezzi)
- Parsing date DD/MM/YYYY → YYYY-MM-DD con gestione sentinelle
- id_diritto deterministico (ROW_NUMBER ordinato per chiave naturale)

Note: il filtro Free/Analogico e fornitore_cluster sono logica di business
e vanno applicati on-the-fly o in tabelle enrich dedicate.
"""
import duckdb
from config.settings import csv_path, out_parquet
from utils.csv_reader import icr_csv

_WIN_STUBS = [
    'CONTRAENTE_WIN', 'DECR_WIN', 'SCAD_WIN',
    'CAUSALE_WIN', 'CAUSALE_INIBIZIONE_WIN', 'NOTE_WIN',
]
_WIN_RANGE = range(1, 10)  # WIN_1 .. WIN_9


def _date_expr(col: str) -> str:
    """Converte DD/MM/YYYY → YYYY-MM-DD, gestendo le date sentinella."""
    return f"""
        CASE
            WHEN TRIM({col}) = '31/12/9999' THEN '9999-12-31'
            WHEN TRIM({col}) = '01/01/0001' THEN '0001-01-01'
            ELSE STRFTIME(TRY_STRPTIME(TRIM({col}), '%d/%m/%Y'), '%Y-%m-%d')
        END
    """


def run(con: duckdb.DuckDBPyConnection) -> int:
    src = csv_path('diritti').replace('\\', '/')
    out = out_parquet('diritti').replace('\\', '/')

    win_cols = ', '.join(
        f'TRIM("{s}_{i}") AS "{s}_{i}"'
        for s in _WIN_STUBS
        for i in _WIN_RANGE
    )

    con.execute(f"""
        COPY (
            WITH raw AS (
                SELECT
                    TRY_CAST(TRIM(PROD)       AS INTEGER) AS prod,
                    TRY_CAST(TRIM(EDIZ)       AS INTEGER) AS ediz,
                    TRIM(TIPOLOGIA)                        AS tipologia,
                    TRIM(D_T)                              AS d_t,
                    TRIM(RAGSOC_DISTR)                     AS ragsoc_distr,
                    TRY_CAST(REPLACE(TRIM(PERC),          ',', '.') AS DOUBLE) AS perc,
                    TRY_CAST(REPLACE(TRIM(PASS_CONS),     ',', '.') AS DOUBLE) AS pass_cons,
                    TRY_CAST(REPLACE(TRIM(PASS_CONS_TOT), ',', '.') AS DOUBLE) AS pass_cons_tot,
                    TRY_CAST(REPLACE(TRIM(PASS_EFF),      ',', '.') AS DOUBLE) AS pass_eff,
                    TRY_CAST(REPLACE(TRIM(PASS_EFF_TOT),  ',', '.') AS DOUBLE) AS pass_eff_tot,
                    TRIM(TIPO_DIRITTO)                     AS tipo_diritto,
                    TRIM(PIATTAFORMA)                      AS piattaforma,
                    {_date_expr('DECR')}                   AS decr,
                    {_date_expr('SCAD')}                   AS scad,
                    TRIM(CAUSALE)                          AS causale,
                    TRIM(FLAG_INIB)                        AS flag_inib,
                    {_date_expr('DECR_INIB')}              AS decr_inib,
                    {_date_expr('SCAD_INIB')}              AS scad_inib,
                    TRY_CAST(TRIM(CONTRATTO)  AS INTEGER)  AS contratto,
                    TRY_CAST(TRIM(RIGA)       AS INTEGER)  AS riga,
                    TRY_CAST(TRIM(SITUAZIONE) AS INTEGER)  AS situazione,
                    TRIM(FR_RR)                            AS fr_rr,
                    {win_cols}
                FROM {icr_csv(con, src)}
                WHERE PROD IS NOT NULL
            )
            SELECT
                ROW_NUMBER() OVER (
                    ORDER BY prod, ediz, contratto, riga, situazione
                ) AS id_diritto,
                *
            FROM (SELECT DISTINCT * FROM raw)
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)

    return con.execute(f"SELECT COUNT(*) FROM '{out}'").fetchone()[0]