# CLAUDE.md — DataHub v2
**Versione:** v1
**Ultimo aggiornamento:** 2026-04-09

## Scopo

Pipeline ETL che legge CSV sorgente e produce Parquet ("bidone") come dato unico trasversale
per tutto l'ecosistema (MyICR_Suite, DBeaver, tool futuri).

## Stack

- **DuckDB** in-memory — lettura CSV + scrittura Parquet
- **Python 3.11.9** embedded standalone (`python\` inclusa nel deploy)
- **loguru** per il logging

## Struttura directory

```
DataHub_v2\
├── config\
│   └── settings.py          ← path, reti_filtro, fornitore_cluster_file
├── domains\                 ← un modulo per CSV sorgente + sidecar _enrich
│   ├── emesso.py            ← raw
│   ├── emesso_enrich.py     ← sidecar (legge da bidone Parquet, non CSV)
│   ├── diritti.py
│   ├── prodotti.py
│   ├── cast.py
│   ├── boxoffice.py
│   ├── boxoffice_dettaglio.py
│   ├── scelte_rete.py
│   ├── veg.py
│   ├── imdb.py
│   ├── reti_cluster.py
│   ├── gemma.py
│   └── osservatorio.py
├── pipelines\
│   ├── weekly.py            ← produce tutti i Parquet
│   └── daily.py             ← produce gemma.parquet + osservatorio.parquet
├── parquet\                 ← output (gitignored)
├── logs\                    ← gitignored
├── main.py                  ← entrypoint: `python main.py [weekly|daily]`
├── create_views.py          ← crea DataHub_v2.duckdb con views sui Parquet
├── compare_v1_v2.py         ← tool di confronto v1 vs v2
├── deploy_DEV_LOCAL.bat     ← deploy A → server B (gitignored, solo su A)
├── run_weekly_SYS_SRV.bat   ← run su server B: sync codice + weekly + publish
└── run_daily_SYS_SRV.bat    ← run su server B: sync codice + daily + publish
```

## Filosofia bidone (Parquet come dato unico)

- **Tabelle raw**: fedeli al CSV sorgente, solo dedup e type parsing
- **Tabelle _enrich**: colonne derivate in sidecar separati, leggono da Parquet (non da CSV)
- **View calcolate**: aggregazioni on-the-fly in DuckDB (es. `boxoffice_aggregato`)
- Niente logica di business nelle tabelle raw

## Tabelle prodotte

| Parquet | Tipo | Note |
|---------|------|------|
| `emesso` | raw | 15 col, ~8.7M righe, dedup su (rete, data_emissione, ora_inizio) |
| `emesso_enrich` | sidecar | join key = (rete, data_emissione, ora_inizio), 1:1 con emesso |
| `diritti` | raw | tutte le colonne CSV incluse WIN_1..WIN_9 |
| `diritti_enrich` | sidecar | join key = id_diritto, aggiunge fornitore_cluster |
| `prodotti` | raw | |
| `cast` | raw | |
| `boxoffice` | raw | 11 col, ~239K righe |
| `boxoffice_dettaglio` | raw | |
| `scelte_rete` | raw | |
| `veg` | raw | |
| `imdb` | raw | |
| `reti_cluster` | raw | |
| `gemma` | raw | daily |
| `osservatorio` | raw | daily |

`boxoffice_aggregato` = view calcolata on-the-fly (non Parquet).

## Gotcha DuckDB

- **Cache issue**: leggere un Parquet nella stessa connessione che l'ha scritto restituisce
  metadati stale. `emesso_enrich` usa una connessione separata da `emesso`.
- **Fan-out JOIN**: se `_prod_anno` ha più righe per `prodotto`, la LEFT JOIN moltiplica le
  righe di emesso. Fix: `GROUP BY prodotto + MAX(anno_produzione)`.
- **Date format**: date in Parquet sono `'%Y-%m-%d'` (non `'%d/%m/%Y'` come nel CSV).
- **Path**: sempre forward slash `/` nei path passati a DuckDB (anche su Windows).

## Input CSV

Letti da `DATAHUB_INPUT_DIR` (env var) o fallback su path locale in `settings.py`.
Su server B il bat imposta: `\\mediaset.it\share\Indirizzo_controllo_risorse\SOFTWARE\SCHEDULATORE\INPUT`

## Deploy e run

- **Deploy A → B**: `deploy_DEV_LOCAL.bat` (manuale, solo su A)
- **Run su B**: `run_weekly_SYS_SRV.bat` / `run_daily_SYS_SRV.bat`
  - Step 1: installa `python\` se prima volta (PowerShell + robocopy)
  - Step 2: sync codice da server B verso `C:\PYTHON_LOCAL\DataHub_v2`
  - Step 3: `python main.py weekly|daily`
  - Step 4: pubblica Parquet su `MyICR_Suite\local_db\parquet\`
- **Sync su A**: `C:\PYTHON\MyICR_Suite\sync_DEV_LOCAL.bat` include sync Parquet

## Roadmap

- **Prossimo**: integrare `run_weekly/daily_SYS_SRV.bat` in Frank
- **Futuro**: migrare MyICR_Suite da SQLite a DuckDB/Parquet (connection.py)