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('cast').replace('\\', '/')
    out = out_parquet('cast').replace('\\', '/')

    con.execute(f"""
        COPY (
            SELECT
                TRY_CAST(CCPROD      AS INTEGER) AS prodotto,
                TRIM(RUOLO)                      AS ruolo,
                TRIM(DESCR_RUOLO)                AS descr_ruolo,
                TRY_CAST(PROGR_RUOLO AS INTEGER) AS progr_ruolo,
                TRY_CAST(PROGR_CAST  AS INTEGER) AS progr_cast,
                TRIM(NOME)                       AS nome,
                TRIM(COGNOME)                    AS cognome,
                TRIM(CCPRODTIP)                  AS ccprodtip
            FROM {icr_csv(con, src)}
            WHERE CCPROD IS NOT NULL
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)
    return con.execute(f"SELECT COUNT(*) FROM '{out}'").fetchone()[0]