"""
diritti_enrich.py — produce diritti_enrich.parquet (sidecar)

Join key con diritti.parquet: id_diritto

Colonne derivate:
- fornitore_cluster : UPPER(TRIM(ragsoc_distr)) → fornitori_cluster_mapping.csv

Dipende da: diritti.parquet (già prodotto)
"""
import duckdb
from config.settings import out_parquet, FORNITORE_CLUSTER_FILE


def run(con: duckdb.DuckDBPyConnection) -> int:
    forn_file = FORNITORE_CLUSTER_FILE.replace('\\', '/')
    dir_pq    = out_parquet('diritti').replace('\\', '/')
    out       = out_parquet('diritti_enrich').replace('\\', '/')

    con.execute(f"""
        CREATE OR REPLACE TABLE _forn_map AS
        SELECT
            UPPER(TRIM(fornitore_diritto)) AS fornitore_key,
            fornitore_cluster
        FROM read_csv('{forn_file}', delim=';', encoding='cp1252', header=true)
    """)

    con.execute(f"""
        COPY (
            SELECT
                d.id_diritto,
                CASE
                    WHEN d.ragsoc_distr IS NOT NULL
                        THEN COALESCE(fm.fornitore_cluster, 'altro')
                    ELSE NULL
                END AS fornitore_cluster
            FROM '{dir_pq}' d
            LEFT JOIN _forn_map fm
                ON UPPER(TRIM(d.ragsoc_distr)) = fm.fornitore_key
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)

    con.execute("DROP TABLE IF EXISTS _forn_map")

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