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

Tutte le colonne originali di estr_box_totali.csv.
L'aggregato per prodotto (incasso/spettatori totali, data debutto)
viene calcolato on-the-fly come view boxoffice_aggregato in create_views.py.
"""
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('boxoffice').replace('\\', '/')
    out = out_parquet('boxoffice').replace('\\', '/')

    con.execute(f"""
        COPY (
            SELECT
                TRY_CAST(TRIM(Prodotto)            AS INTEGER) AS prodotto,
                TRIM(Stagione)                                  AS stagione,
                STRFTIME(
                    TRY_STRPTIME(TRIM(Data_Debutto), '%d/%m/%Y'),
                    '%Y-%m-%d'
                )                                              AS data_debutto,
                TRIM(Distributore)                             AS distributore,
                TRIM(Tipo_programmazione)                      AS tipo_programmazione,
                TRY_CAST(TRIM(Sale)                AS INTEGER) AS sale,
                TRY_CAST(TRIM(Cinema)              AS INTEGER) AS cinema,
                TRY_CAST(TRIM("Citta'")            AS INTEGER) AS citta,
                TRY_CAST(TRIM(Ggprog)              AS INTEGER) AS ggprog,
                TRY_CAST(REPLACE(TRIM(Spettatori), ',', '.') AS BIGINT)  AS spettatori,
                TRY_CAST(REPLACE(TRIM(Incasso),    ',', '.') AS DOUBLE)  AS incasso
            FROM {icr_csv(con, src)}
            WHERE Prodotto IS NOT NULL
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)

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