"""
emesso.py — produce emesso.parquet (Layer 1)

Layer 1: fedele alla sorgente CSV, solo parsing tipi e rinomina colonne.
Nessun filtro reti, nessuna deduplicazione.

Il filtro 34 reti e la dedup vengono applicati in Layer 2
da Parquet_layer1ToParquet_layer2.
"""
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('emesso').replace('\\', '/')
    out = out_parquet('emesso').replace('\\', '/')

    con.execute(f"""
        COPY (
            SELECT
                TRIM("EMESSO_CCRETE")                                     AS rete,
                STRFTIME(
                    TRY_STRPTIME(TRIM("Emesso - Data"), '%Y%m%d'),
                    '%Y-%m-%d'
                )                                                         AS data_emissione,
                TRY_CAST(REPLACE(TRIM("Inizio Emissione"), ':', '') AS INTEGER) AS ora_inizio,
                TRY_CAST(REPLACE(TRIM("Fine Emissione"),   ':', '') AS INTEGER) AS ora_fine,
                TRY_CAST(TRIM("EPS_PROGRAM_CODE")           AS INTEGER)   AS prodotto,
                TRY_CAST(TRIM("Edizione")                   AS INTEGER)   AS edizione,
                TRY_CAST(TRIM("Emesso - Dur. netta")        AS INTEGER)   AS durata_netta,
                TRIM("Emesso - Prima Visione")                            AS prima_visione,
                TRY_CAST(TRIM("Emesso - Episodio")          AS INTEGER)   AS episodio,
                TRY_CAST(TRIM("Emesso - Audience")          AS INTEGER)   AS audience,
                TRY_CAST(
                    REPLACE(TRIM("Emesso - Share"), ',', '.') AS DOUBLE
                )                                                         AS share,
                TRY_CAST(TRIM("Emesso - Dur. lorda")        AS INTEGER)   AS durata_lorda,
                TRY_CAST(TRIM("EMESSO_NCPARPGM")            AS INTEGER)   AS ncparpgm,
                TRIM(UPPER("tipologia"))                                  AS tipologia,
                TRIM(UPPER("EMESSO_CCFASCIA"))                            AS fascia
            FROM {icr_csv(con, src)}
            WHERE "EPS_PROGRAM_CODE" IS NOT NULL
              AND "Emesso - Data"    IS NOT NULL
              AND "Inizio Emissione" IS NOT NULL
        ) TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)
    """)

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