"""
Pipeline weekly: produce tutti i Parquet Layer 1 in parquet/

Layer 1 = fedele alla sorgente CSV, solo parsing tipi e rinomina colonne.
Nessun filtro reti, nessuna deduplicazione, nessuna logica di business.

I transform di business (emesso_enrich, diritti_enrich, filtro 34 reti emesso)
vengono applicati in Layer 2 da Parquet_layer1ToParquet_layer2.
"""
import duckdb
from loguru import logger

from domains import (
    reti_cluster, prodotti, cast, imdb, veg,
    scelte_rete, boxoffice, boxoffice_dettaglio,
    diritti, emesso,
)


WEEKLY_DOMAINS = [
    ('reti_cluster',        reti_cluster),
    ('prodotti',            prodotti),
    ('cast',                cast),
    ('imdb',                imdb),
    ('veg',                 veg),
    ('scelte_rete',         scelte_rete),
    ('boxoffice',           boxoffice),
    ('boxoffice_dettaglio', boxoffice_dettaglio),
    ('diritti',             diritti),
    ('emesso',              emesso),
]


def run(dry_run: bool = False) -> dict:
    """
    Esegue la pipeline weekly Layer 1 completa.
    Restituisce un dict {nome: count}.
    """
    results = {}

    with duckdb.connect() as con:
        for name, mod in WEEKLY_DOMAINS:
            logger.info(f"[weekly] avvio domain: {name}")
            try:
                result = mod.run(con)
                results[name] = result
                logger.success(f"[weekly] {name}: {result:,} righe")
            except Exception as exc:
                logger.error(f"[weekly] {name} FALLITO: {exc}")
                results[name] = -1

    return results