import duckdb
from config.settings import csv_path, out_parquet
from utils.csv_reader import icr_csv


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

    con.execute(f"""
        COPY (
            SELECT
                TRY_CAST(PRODOTTO        AS INTEGER) AS prodotto,
                TRY_CAST(EDIZIONE        AS INTEGER) AS edizione,
                TRIM(PROGRAM_ID)                     AS program_id,
                TRIM(TITOLO_ITALIANO)                AS titolo_italiano,
                TRIM(TITOLO_ORIGINALE)               AS titolo_originale,
                TRIM(DESCR_EDIZIONE)                 AS descr_edizione,
                TRIM(UNICO_SERIALE)                  AS unico_seriale,
                TRIM(SUPERSERIE_DESCR)               AS superserie_descr,
                TRY_CAST(SUPERSERIE_ID   AS INTEGER) AS superserie_id,
                TRIM(SUPERSERIE_EXTKEY)              AS superserie_extkey,
                TRY_CAST(STAGIONE        AS INTEGER) AS stagione,
                TRY_CAST(DURATA          AS INTEGER) AS durata,
                TRIM(CROMIA)                         AS cromia,
                TRIM(VEG)                            AS veg,
                TRY_CAST(NUM_EPISODI     AS INTEGER) AS num_episodi,
                TRIM(PAESI_PRODUZIONE1)              AS paesi_produzione1,
                TRIM(PAESI_PRODUZIONE2)              AS paesi_produzione2,
                TRIM(PAESI_PRODUZIONE3)              AS paesi_produzione3,
                TRY_CAST(ANNO_PRODUZIONE AS INTEGER) AS anno_produzione,
                TRIM(GENERE1)                        AS genere1,
                TRIM(GENERE2)                        AS genere2,
                TRIM(GENERE3)                        AS genere3,
                TRIM(PROVENIENZA)                    AS provenienza,
                TRIM(VM)                             AS vm,
                TRIM(UPPER(TIPOLOGIA))               AS tipologia
            FROM {icr_csv(con, src)}
            WHERE PRODOTTO IS NOT NULL
              AND EDIZIONE IS NOT NULL
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)
    return con.execute(f"SELECT COUNT(*) FROM '{out}'").fetchone()[0]