X2j%dZddlZddlmZdZedZedZdedefd Zdded edzfd Z dd ed efdZ d efdZ dS)u Parquet L3 aggregation — generazione emesso_agg.parquet + custom_agg.parquet. Chiamato da bootstrap.py (al sync) e da linker/launcher.py (avvio da admin panel). Prende app_root e server_root espliciti — nessuna dipendenza da costanti di bootstrap. N)Path)N1N2N3C5I1MCR4z$I:\SOFTWARE\PYTHON_SRV\sqlite_layer2z/I:\SOFTWARE\PYTHON_SRV\PYTHON_LOCAL\MyICR_Suiteapp_rootreturnc|dz dz dz }|sdS|j}|dz dz }dD];}||z }|r |j|krdS= emesso.parquet locale) → pull. 3. Altrimenti → build locale → push su server → scrivi mirror_l2.sqlite. NrrrrrSz*Aggiornamento dati aggregati dal server...Tparentsexist_ok)rcustom_agg.parquetuElaborazione dati in corso…Wz)Pubblicazione dati aggregati su server...zAggiornamento SQLite layer2...z5[parquet_l3] mirror_l2.sqlite fallito (non critico): )_SERVER_ROOT_FALLBACKr rrrmkdirshutilcopy2build_parquet_l3parent_SQLITE_LAYER2_DIRwrite_mirror_l2_sqlite Exceptionprint) r r" progress_cbragg_dir server_agg server_marker emesso_l2rsrces rsync_parquet_l3r;s+ z)I5Mz)M9G*,} K< = = = O "7 + + + + + O O O M!MM N N N N N N N N N O OOs1G G) G$$G)rr5cddl}|ddt|dz dd}t|dz dd}t|d z dd}t|d z dd}t|d z dd}|} d d t D} |r |ddt|dz dd} | d| d|d| d|r |ddt|dz dd} | d|d|d|d|d|d|d|d| d| dS) zKLegge i parquet L2 locali e scrive emesso_agg.parquet + custom_agg.parquet.rNTr%r\/rrrrz, c3"K|] }d|dV dS)'N).0ns r z#build_parquet_l3..cs*44!XXXX444444r!Tu&Elaborazione dati in corso… (emesso)rz COPY ( WITH base AS ( SELECT CAST(prodotto AS VARCHAR) AS RIFER, rete, data_emissione, ora_inizio, fascia, audience, share, CASE WHEN rete IN (zH) THEN 'G' ELSE 'T' END AS tipo_rete FROM read_parquet('a') ) SELECT RIFER, tipo_rete, FIRST(rete ORDER BY data_emissione ASC, ora_inizio ASC) AS rete_first, FIRST(data_emissione ORDER BY data_emissione ASC, ora_inizio ASC) AS data_first, FIRST(ora_inizio ORDER BY data_emissione ASC, ora_inizio ASC) AS ora_first, FIRST(fascia ORDER BY data_emissione ASC, ora_inizio ASC) AS fascia_first, FIRST(audience ORDER BY data_emissione ASC, ora_inizio ASC) AS aud_first, FIRST(share ORDER BY data_emissione ASC, ora_inizio ASC) AS sha_first, LAST(rete ORDER BY data_emissione ASC, ora_inizio ASC) AS rete_last, LAST(data_emissione ORDER BY data_emissione ASC, ora_inizio ASC) AS data_last, LAST(ora_inizio ORDER BY data_emissione ASC, ora_inizio ASC) AS ora_last, LAST(fascia ORDER BY data_emissione ASC, ora_inizio ASC) AS fascia_last, LAST(audience ORDER BY data_emissione ASC, ora_inizio ASC) AS aud_last, LAST(share ORDER BY data_emissione ASC, ora_inizio ASC) AS sha_last FROM base GROUP BY RIFER, tipo_rete ) TO 'z' (FORMAT PARQUET) Vu*Elaborazione dati in corso… (anagrafica)r(zc COPY ( WITH prod AS ( SELECT * FROM read_parquet('z') QUALIFY ROW_NUMBER() OVER (PARTITION BY prodotto ORDER BY edizione) = 1 ), imdb AS ( SELECT codice, MIN(riferimento_imdb) AS riferimento_imdb FROM read_parquet('z') GROUP BY codice ), reg AS ( SELECT prodotto, cognome AS REGISTA_COGN, nome AS REGISTA_NOME FROM read_parquet('aA') WHERE ruolo = 'FRE' QUALIFY ROW_NUMBER() OVER (PARTITION BY prodotto ORDER BY progr_cast) = 1 ), a1 AS ( SELECT prodotto, TRIM(COALESCE(nome,'') || ' ' || COALESCE(cognome,'')) AS ATTORE1 FROM read_parquet('z') WHERE ruolo='C001' AND progr_cast=1 ), a2 AS ( SELECT prodotto, TRIM(COALESCE(nome,'') || ' ' || COALESCE(cognome,'')) AS ATTORE2 FROM read_parquet('z') WHERE ruolo='C001' AND progr_cast=2 ), a3 AS ( SELECT prodotto, TRIM(COALESCE(nome,'') || ' ' || COALESCE(cognome,'')) AS ATTORE3 FROM read_parquet('z') WHERE ruolo='C001' AND progr_cast=3 ), distr AS ( SELECT prod AS prodotto, MIN(ragsoc_distr) AS DF_RAGSOC_DISTR FROM read_parquet('a') GROUP BY prod ) SELECT CAST(p.prodotto AS VARCHAR) AS RIFER, p.tipologia AS TIPOL, p.titolo_originale AS "TO", p.titolo_italiano AS TI, p.anno_produzione AS ANNO, p.paesi_produzione1 AS PAESE, p.genere1 AS GENERE1, p.num_episodi AS EPIS, p.durata AS DUR, p.superserie_descr AS SUPERSERIE, CASE p.veg WHEN 'EVER GREEN' THEN 'EV.GR' ELSE p.veg END AS VEG_FREE, i.riferimento_imdb AS IMDB_CODICE, r.REGISTA_COGN, r.REGISTA_NOME, a1.ATTORE1, a2.ATTORE2, a3.ATTORE3, d.DF_RAGSOC_DISTR FROM prod p LEFT JOIN imdb i ON i.codice = p.prodotto LEFT JOIN reg r ON r.prodotto = p.prodotto LEFT JOIN a1 ON a1.prodotto = p.prodotto LEFT JOIN a2 ON a2.prodotto = p.prodotto LEFT JOIN a3 ON a3.prodotto = p.prodotto LEFT JOIN distr d ON d.prodotto = p.prodotto ) TO ') duckdbr+strreplaceconnectjoin _GENERALISTEexecuteclose) rr5r4rGem_fprod_fcast_fimdb_fdir_fconnets emesso_out custom_outs rr.r.WszMMM MM$M... !11 2 2 : :4 E ED !33 4 4 < r(zSELECT * FROM read_parquet('z') emesso_aggrIF) if_existsindex custom_aggzPRAGMA journal_mode=WAL) rGsqlite3r0rJrHrIrMdfrNto_sql) r5rG_sl mirror_pathrTemesso_fcustom_f df_emesso df_customsl_cons rr1r1sMMMM$'99K ..  C71122::4EEH71122::4EEH G8GGGHHKKMMIHHHHHIILLNNIIIKKK [[[)) * *F \6YeLLL \FiuMMM NN,--- LLNNNNNr!)NN)N) __doc__r,pathlibrrLr0r*boolr r;r.r1rAr!rrjs  9 TABBOPP t     6O6Od6O6O6O6O6OrooDo4oooodDr!