From 4a467f5fd65c55ab13d805577465dfdcbd4c4cb4 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Fri, 28 Aug 2026 10:01:03 +0100 Subject: [PATCH] =?UTF-8?q?refactor:=20dashboard=20ACB=20v5=20=E2=80=94=20?= =?UTF-8?q?2=20fonti,=20-42%=20righe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 4 +- pages/00_Vista_Insieme.py | 261 ++++++-------- pages/01_Dataset_Explorer.py | 197 ++++++++--- pages/02_Pipeline_Health.py | 328 +++++------------ pages/09_Query_SQL.py | 366 +++---------------- pyproject.toml | 31 ++ requirements.txt | 5 +- sources.py | 415 ++++++++-------------- tests/test_sources.py | 667 +++++++++++++++-------------------- 9 files changed, 861 insertions(+), 1413 deletions(-) diff --git a/app.py b/app.py index 8e5341b..a52417e 100644 --- a/app.py +++ b/app.py @@ -40,10 +40,10 @@ st.Page("pages/07_Fonte.py", title="Scheda fonte", icon="πŸ”"), ], "Dataset Incubator": [ - st.Page("pages/02_Pipeline_Health.py", title="Pipeline candidate", icon="βš™οΈ"), + st.Page("pages/02_Pipeline_Health.py", title="Registry / Repo", icon="πŸ“¦"), ], "Catalogo": [ - st.Page("pages/01_Dataset_Explorer.py", title="Esplora dataset", icon="πŸ“š"), + st.Page("pages/01_Dataset_Explorer.py", title="Catalogo", icon="πŸ“š"), st.Page("pages/09_Query_SQL.py", title="Query SQL", icon="πŸ§ͺ"), ], } diff --git a/pages/00_Vista_Insieme.py b/pages/00_Vista_Insieme.py index b8f8ddf..8e58fd0 100644 --- a/pages/00_Vista_Insieme.py +++ b/pages/00_Vista_Insieme.py @@ -1,7 +1,4 @@ -""" -Vista d'insieme β€” polso del DataCivicLab. -Metriche e stato da Source Observatory, Dataset Incubator e Community. -""" +"""Vista d'insieme β€” polso del DataCivicLab.""" import altair as alt import pandas as pd @@ -10,200 +7,160 @@ from sources import ( data_freshness_note, load_catalog, - load_check_coverage, - load_inventory_report, load_radar, load_signals, - load_sources_dashboard, - load_sources_registry, + load_workspace_triage, ) st.title("πŸ“Š Vista d'insieme") -st.markdown("Salute del Lab: dalle fonti monitorate ai dataset pubblicati.") - -# ── Carica tutti i dati ────────────────────────────────────────── +# ── Carica dati ───────────────────────────────────────────────── +triage = load_workspace_triage() radar = load_radar() -registry = load_sources_registry() -coverage_df = load_check_coverage() -inventory_report = load_inventory_report() catalog = load_catalog() -pipeline_signals = load_signals() +signals_data = load_signals() sources = radar.get("sources", []) status_counts = radar.get("status_counts", {}) persistent_red = radar.get("persistent_red", 0) -inventory_sources = inventory_report.get("sources", {}) datasets = catalog.get("datasets", []) -sigs = pipeline_signals.get("signals", []) +sigs = signals_data.get("signals", []) + +prs = triage.get("prs", []) +issues = triage.get("issues", []) +discussions = triage.get("discussions", []) # Conteggi -n_registry = len(registry) -n_radar = len(sources) +tot = len(datasets) +published = sum(1 for d in datasets if d.get("stage") == "published") +incubating = tot - published n_green = status_counts.get("GREEN", 0) n_yellow = status_counts.get("YELLOW", 0) n_red = status_counts.get("RED", 0) +ok_count = sum(1 for s in sigs if s.get("status") == "ok") +warn_count = sum(1 for s in sigs if s.get("status") == "warn") +error_count = sum(1 for s in sigs if s.get("status") == "error") -n_inv_ok = sum(1 for v in inventory_sources.values() if v.get("status") == "ok") - -tot_inv = int(coverage_df["inv_items"].sum()) if not coverage_df.empty else 0 -tot_chk = int(coverage_df["chk_items"].sum()) if not coverage_df.empty else 0 -coverage_pct = round(tot_chk / tot_inv * 100, 1) if tot_inv else 0 - -n_published = sum(1 for ds in datasets if ds.get("stage") == "published") -n_incubating = sum(1 for ds in datasets if ds.get("stage") == "incubating") -n_pipeline_err = sum(1 for sig in sigs if sig.get("status") == "error") - -# ── KPI ────────────────────────────────────────────────────────── +# ══════════════════════════════════════════════════════════════════ +# KPI COMPATTA +# ══════════════════════════════════════════════════════════════════ col1, col2, col3, col4 = st.columns(4) -col1.metric("πŸ“‘ Radar fonti", f"{n_radar}/{n_registry}", f"{n_green}🟒 {n_yellow}🟑 {n_red}πŸ”΄") -col2.metric("πŸ“¦ Items inventario", f"{tot_inv:,}", f"{coverage_pct}% checked ({tot_chk:,})") -col3.metric( - "πŸ“š Dataset", f"{len(datasets)}", f"{n_published} pubblicati Β· {n_incubating} in incubazione" -) -col4.metric("βœ… Pubblicati", n_published, f"{n_incubating} in incubazione") +col1.metric("πŸ“‘ Radar", f"{n_green + n_yellow + n_red}", f"{n_green}🟒 {n_yellow}🟑 {n_red}πŸ”΄") +col2.metric("πŸ“š Dataset", f"{tot}", f"{published} pubblicati") +col3.metric("⚑ Pipeline", f"{ok_count}", f"{warn_count}⚠️ {error_count}❌") +col4.metric("πŸ”€ PR", len(prs), f"{len(issues)} issues Β· {len(discussions)} disc") if persistent_red: - st.warning( - f"πŸ”΄ **{persistent_red} fonte/i persistentemente RED** " - "(streak > 7 giorni) β€” vedi Radar per dettaglio" - ) - -if n_pipeline_err: - st.error(f"❌ **{n_pipeline_err} pipeline in errore** β€” vedi Pipeline CI") + st.warning(f"πŸ”΄ **{persistent_red} fonte/i RED persistente** (streak > 7gg)") +if error_count: + st.error(f"❌ **{error_count} pipeline in errore**") st.markdown("---") # ══════════════════════════════════════════════════════════════════ -# SOURCE OBSERVATORY +# RADAR β€” barra segmentata # ══════════════════════════════════════════════════════════════════ -st.subheader("Source Observatory") - -# -- Stato radar -- -col_s1, col_s2, col_s3 = st.columns([1, 1, 1]) - -with col_s1: - st.metric("🟒 GREEN", n_green) -with col_s2: - st.metric("🟑 YELLOW", n_yellow) -with col_s3: - st.metric("πŸ”΄ RED", n_red) - -# -- KPI aggregati SO (report v2) -- -so_dashboard = load_sources_dashboard() -so_summary = so_dashboard.get("summary", {}) -by_verdict = so_summary.get("by_verdict", {}) -n_datasets_use = so_summary.get("tot_datasets_in_use", 0) -n_inv_changed = by_verdict.get("INVENTORY_CHANGED", 0) -n_partial = by_verdict.get("PARTIALLY_SCOPED", 0) - -if so_summary: - col_s4, col_s5 = st.columns(2) - with col_s4: - st.metric("🧩 Dataset in uso", f"{n_datasets_use:,}") - with col_s5: - st.metric( - "πŸ”„ Inventario cambiato", - f"{n_inv_changed}", - f"{n_partial} scoping parziale", - ) +st.subheader("Radar fonti") -# Bar chart radar: barra per stato -if n_radar: - radar_df = pd.DataFrame( - [ - {"stato": "GREEN", "conteggio": n_green, "colore": "#16a34a"}, - {"stato": "YELLOW", "conteggio": n_yellow, "colore": "#fbbf24"}, - {"stato": "RED", "conteggio": n_red, "colore": "#dc2626"}, - ] - ) - radar_bars = ( +radar_df = pd.DataFrame( + [ + {"stato": "GREEN", "n": n_green}, + {"stato": "YELLOW", "n": n_yellow}, + {"stato": "RED", "n": n_red}, + ] +) +radar_df = radar_df[radar_df["n"] > 0] # nascondi zero + +if not radar_df.empty: + chart = ( alt.Chart(radar_df) - .mark_bar() + .mark_bar(height=30) .encode( - x=alt.X("stato:N", title=None, sort=["GREEN", "YELLOW", "RED"]), - y=alt.Y("conteggio:Q", title="Fonti"), + x=alt.X("n:Q", stack="normalize", title=None, axis=None), color=alt.Color( "stato:N", scale={ "domain": ["GREEN", "YELLOW", "RED"], "range": ["#16a34a", "#fbbf24", "#dc2626"], }, - title=None, legend=None, ), - tooltip=["stato:N", "conteggio:Q"], + tooltip=["stato", "n"], ) - .properties(height=150) + .properties(height=30) ) - st.altair_chart(radar_bars, width="stretch") - + st.altair_chart(chart, width="stretch") + # Legenda manuale + parts = [] + if n_green: + parts.append(f"🟒 {n_green}") + if n_yellow: + parts.append(f"🟑 {n_yellow}") + if n_red: + parts.append(f"πŸ”΄ {n_red}") + st.caption(f"{' Β· '.join(parts)} β€” {len(sources)} fonti totali") + +# ── Fonti RED non healthy ─────────────────────────────────────── +unhealthy = [s for s in sources if s.get("status") in ("YELLOW", "RED")] +if unhealthy: + with st.expander(f"⚠️ {len(unhealthy)} fonti non healthy", expanded=False): + for s in unhealthy: + icon = "πŸ”΄" if s.get("status") == "RED" else "🟑" + streak = s.get("red_streak", 0) + st.write( + f"{icon} **{s['id']}** β€” {s.get('note', '')}{' (streak ' + str(streak) + ')' if streak else ''}" + ) st.markdown("---") # ══════════════════════════════════════════════════════════════════ -# DATASET +# DATASET PER FONTE # ══════════════════════════════════════════════════════════════════ -st.subheader("Dataset") - -stages = sorted(set(d.get("stage", "unknown") for d in datasets)) -stage_filter = st.selectbox("Filtra per stage", ["Tutti"] + stages) - -search = st.text_input("Cerca dataset", placeholder="slug, nome o descrizione...") - -filtered = datasets -if stage_filter != "Tutti": - filtered = [d for d in filtered if d.get("stage") == stage_filter] -if search: - q = search.lower() - filtered = [ - d - for d in filtered - if q in d.get("slug", "").lower() - or q in d.get("name", "").lower() - or q in d.get("description", "").lower() - ] +st.subheader("Dataset per fonte") -st.write(f"**{len(filtered)} dataset** trovati") - -for ds in filtered: - period = ds.get("period", {}) - yrs = f"{period.get('start', '?')}–{period.get('end', '?')}" if period else "?" - with st.expander(f"**{ds.get('slug', '?')}** β€” {ds.get('stage', '?')}"): - st.write(f"**Nome:** {ds.get('name', 'β€”')}") - st.write(f"**Descrizione:** {ds.get('description', 'β€”')}") - st.write(f"**Fonte:** {ds.get('source', '?')}") - st.write(f"**Anni:** {yrs}") - loc = ds.get("location", {}) - if loc.get("path"): - st.write(f"**Path GCS:** `{loc['path']}`") - - # Schema colonne - cols = ds.get("columns", []) - if cols: - st.markdown("**Schema colonne**") - col_df = pd.DataFrame( - [ - { - "colonna": c.get("name", "?"), - "tipo": c.get("type", "?"), - "ruolo": c.get("role", "?"), - "descrizione": c.get("description", ""), - } - for c in cols - ] - ) - st.dataframe(col_df, hide_index=True, width="stretch") - -# -- Alert run falliti -- -run_failed = [s for s in sigs if s.get("run", {}).get("status") == "failed"] -if run_failed: - st.warning( - f"⚠️ **{len(run_failed)} candidate con run CI fallito** " - f"β€” vai a βš™οΈ Pipeline candidate per dettagli" - ) +by_source: dict[str, list[dict]] = {} +for ds in datasets: + sid = ds.get("source_id") or ds.get("source", "unknown") + by_source.setdefault(sid, []).append(ds) + +# Top 15 fonti +top = sorted(by_source.items(), key=lambda x: -len(x[1]))[:15] +chart_df = pd.DataFrame([{"fonte": s[:30], "n": len(ds)} for s, ds in top]) -if n_pipeline_err: - st.warning(f"⚠️ **{n_pipeline_err} pipeline in errore**") +if not chart_df.empty: + chart = ( + alt.Chart(chart_df) + .mark_bar(color="#3b82f6") + .encode( + y=alt.Y("fonte:N", title=None, sort="-x"), + x=alt.X("n:Q", title="Dataset"), + tooltip=["fonte", "n"], + ) + .properties(height=max(22 * len(chart_df), 80)) + ) + st.altair_chart(chart, width="stretch") + +# Tabella compatta fonti +st.write(f"**{len(by_source)} fonti** Β· {tot} dataset totali") +table_rows = [] +for s, ds in top: + pub = sum(1 for d in ds if d.get("stage") == "published") + inc = sum(1 for d in ds if d.get("stage") == "incubating") + stage_str = "+".join(filter(None, [f"{pub}pub" if pub else "", f"{inc}inc" if inc else ""])) + table_rows.append({"fonte": s, "n": len(ds), "stage": stage_str}) + +if table_rows: + df_table = pd.DataFrame(table_rows) + st.dataframe( + df_table, + column_config={ + "fonte": "Fonte", + "n": st.column_config.NumberColumn("Dataset", format="%d"), + "stage": "Stage", + }, + hide_index=True, + width="stretch", + height=min(35 * len(table_rows) + 35, 300), + ) data_freshness_note() diff --git a/pages/01_Dataset_Explorer.py b/pages/01_Dataset_Explorer.py index 209e425..2160b47 100644 --- a/pages/01_Dataset_Explorer.py +++ b/pages/01_Dataset_Explorer.py @@ -1,4 +1,4 @@ -"""Dataset Explorer β€” copertura anni e verifica parquet su GCS.""" +"""Catalogo β€” esplora dataset con filtri, tabella e copertura anni.""" import altair as alt import pandas as pd @@ -7,18 +7,154 @@ from sources import data_freshness_note, load_catalog, verify_parquet -st.title("πŸ“š Esplora dataset") - -st.markdown( - "Copertura anni dei dataset pubblicati e verifica parquet su GCS. " - "Per query SQL avanzate, usa la pagina **Query SQL**." -) +st.title("πŸ“š Catalogo") catalog = load_catalog() datasets = catalog.get("datasets", []) -# ── Copertura anni (ex Copertura Dati) ──────────────────────────────────── +if not datasets: + st.error("Catalogo non disponibile.") + st.stop() + +# ── KPI ───────────────────────────────────────────────────────── +n_total = len(datasets) +n_published = sum(1 for d in datasets if d.get("stage") == "published") +n_incubating = n_total - n_published +by_source = {} +for d in datasets: + sid = d.get("source_id") or d.get("source", "unknown") + by_source[sid] = by_source.get(sid, 0) + 1 +n_sources = len(by_source) +categories = {d.get("category", "") for d in datasets if d.get("category")} +n_categories = len(categories) + +col1, col2, col3, col4 = st.columns(4) +col1.metric("πŸ“š Dataset", n_total) +col2.metric("βœ… Pubblicati", n_published) +col3.metric("πŸ”¬ Incubazione", n_incubating) +col4.metric("🏷️ Fonti", n_sources) + +st.markdown("---") + +# ── Filtri ─────────────────────────────────────────────────────── +st.subheader("Filtri") + +source_options = sorted(by_source.keys(), key=lambda s: -by_source[s]) +source_counts = {s: by_source[s] for s in source_options} +stage_options = ["Tutti", "published", "incubating"] +cat_options = ["Tutti"] + sorted(categories) + +col_f1, col_f2, col_f3 = st.columns(3) +with col_f1: + src_filter = st.selectbox( + "Fonte (source_id)", + ["Tutti"] + [f"{s} ({source_counts[s]})" for s in source_options], + key="cat_src", + ) +with col_f2: + stage_filter = st.selectbox("Stage", stage_options, key="cat_stage") +with col_f3: + cat_filter = st.selectbox("Categoria", cat_options, key="cat_cat") + +search = st.text_input("Cerca", placeholder="slug, nome o descrizione...", key="cat_search") + +# Applica filtri +filtered = datasets +if src_filter != "Tutti": + src_id = src_filter.split(" (")[0] + filtered = [d for d in filtered if (d.get("source_id") or d.get("source")) == src_id] +if stage_filter != "Tutti": + filtered = [d for d in filtered if d.get("stage") == stage_filter] +if cat_filter != "Tutti": + filtered = [d for d in filtered if d.get("category") == cat_filter] +if search: + q = search.lower() + filtered = [ + d + for d in filtered + if q in d.get("slug", "").lower() + or q in d.get("name", "").lower() + or q in d.get("description", "").lower() + ] + +st.write(f"**{len(filtered)} dataset** trovati") + +# ── Tabella ────────────────────────────────────────────────────── rows = [] +for ds in filtered: + period = ds.get("period", {}) + start = period.get("start", "?") + end = period.get("end", "?") + yrs = f"{start}–{end}" if start != "?" else "?" + tags = ds.get("tags", []) + n_cols = len(ds.get("columns", [])) + stage_icon = "βœ…" if ds.get("stage") == "published" else "πŸ”¬" + rows.append( + { + "slug": ds.get("slug", ""), + "nome": ds.get("name", "")[:50], + "fonte": ds.get("source_id") or ds.get("source", "?"), + "stage": f"{stage_icon} {ds.get('stage', '?')}", + "anni": yrs, + "tags": ", ".join(tags[:3]), + "schema": f"{n_cols} colonne" if n_cols else "β€”", + } + ) + +if rows: + df = pd.DataFrame(rows) + st.dataframe( + df, + column_config={ + "slug": st.column_config.TextColumn("Slug", width="medium"), + "nome": st.column_config.TextColumn("Nome", width="medium"), + "fonte": st.column_config.TextColumn("Fonte", width="small"), + "stage": st.column_config.TextColumn("Stage", width="small"), + "anni": st.column_config.TextColumn("Anni", width="small"), + "tags": st.column_config.TextColumn("Tags", width="small"), + "schema": st.column_config.TextColumn("Schema", width="small"), + }, + hide_index=True, + width="stretch", + height=min(40 * len(rows) + 35, 500), + ) + + # Expander dettaglio per ogni dataset + for ds in filtered: + slug = ds.get("slug", "") + loc = ds.get("location", {}) + cols = ds.get("columns", []) + with st.expander(f"**{slug}** β€” {ds.get('name', '')}"): + st.write(f"**Descrizione:** {ds.get('description', 'β€”')}") + st.write(f"**Fonte:** {ds.get('source_id') or ds.get('source', '?')}") + period = ds.get("period", {}) + st.write(f"**Anni:** {period.get('start', '?')}–{period.get('end', '?')}") + st.write(f"**Stage:** {ds.get('stage', '?')}") + if loc.get("path"): + st.write(f"**GCS:** `{loc['path']}`") + if cols: + st.markdown("**Schema colonne**") + col_df = pd.DataFrame( + [ + { + "colonna": c.get("name", "?"), + "tipo": c.get("type", "?"), + "ruolo": c.get("role", "?"), + "desc": c.get("description", ""), + } + for c in cols + ] + ) + st.dataframe(col_df, hide_index=True, width="stretch") +else: + st.info("Nessun dataset trovato con i filtri selezionati.") + +st.markdown("---") + +# ── Copertura anni ─────────────────────────────────────────────── +st.subheader("Copertura anni") + +year_rows = [] for ds in datasets: slug = ds.get("slug", "") period = ds.get("period", {}) @@ -27,24 +163,17 @@ stage = ds.get("stage", "?") if start and end: for y in range(start, end + 1): - rows.append({"dataset": slug, "stage": stage, "anno": str(y)}) - else: - rows.append({"dataset": slug, "stage": stage, "anno": "?"}) - -cov_df = pd.DataFrame(rows) + year_rows.append({"dataset": slug, "stage": stage, "anno": str(y)}) -if not cov_df.empty: +if year_rows: + cov_df = pd.DataFrame(year_rows) col_mat, col_chart = st.columns([1.5, 1]) - with col_mat: pivot = cov_df.pivot_table( index="dataset", columns="anno", values="stage", aggfunc="first" ).fillna("") - # Anni in ordine decrescente (piΓΉ recente primo) pivot = pivot[sorted(pivot.columns, reverse=True)] - st.subheader("Copertura anni") st.dataframe(pivot, width="stretch", height=320) - with col_chart: real = cov_df[cov_df["anno"] != "?"] if not real.empty: @@ -60,31 +189,24 @@ .properties(height=280) ) st.altair_chart(chart, width="stretch") - n_max = real["anno"].max() n_avg = real.groupby("dataset").size().mean() - st.info( - f"πŸ“Š **{len(datasets)}** dataset Β· copertura fino a **{n_max}** Β· " - f"media **{n_avg:.1f}** anni/dataset" - ) + st.info(f"πŸ“Š Copertura fino a **{n_max}** Β· media **{n_avg:.1f}** anni/dataset") st.markdown("---") -# ── Verifica parquet su GCS ──────────────────────────────────────────────── -st.subheader("Verifica e scarica parquet da GCS") -st.markdown( - "Seleziona un dataset e un anno per controllare se il parquet esiste " - "sul bucket e quanti record contiene." -) - +# ── Verifica parquet ───────────────────────────────────────────── +st.subheader("Verifica parquet da GCS") slug_options = [ds.get("slug", "") for ds in datasets if ds.get("period", {}).get("start")] col_vs, col_vy, _ = st.columns([2, 1, 4]) with col_vs: - verify_slug = st.selectbox("Dataset", slug_options) + verify_slug = st.selectbox("Dataset", slug_options, key="cat_verify_slug") with col_vy: - verify_year = st.number_input("Anno", min_value=2010, max_value=2026, value=2023, step=1) + verify_year = st.number_input( + "Anno", min_value=2010, max_value=2026, value=2023, step=1, key="cat_verify_year" + ) -if st.button("πŸ” Verifica su GCS"): +if st.button("πŸ” Verifica su GCS", key="cat_verify_btn"): with st.spinner(f"Verifica {verify_slug}/{verify_year}..."): try: result = verify_parquet(verify_slug, verify_year) @@ -94,18 +216,11 @@ ) st.success(f"βœ… **{verify_slug}**/{verify_year} β€” **{result['records']:,}** record") st.markdown( - f"πŸ“₯ **[Scarica parquet]({parquet_url})** " - f"β€” {result['records']:,} righe, formato colonnare" + f"πŸ“₯ **[Scarica parquet]({parquet_url})** β€” {result['records']:,} righe" ) else: st.warning("⚠️ Parquet trovato ma 0 record") except Exception as e: st.error(f"❌ Parquet non raggiungibile: {e}") -st.markdown("---") -st.caption( - "La verifica usa DuckDB per leggere direttamente il parquet da GCS. " - "Per query SQL avanzate, usa la pagina **Query SQL**." -) - data_freshness_note() diff --git a/pages/02_Pipeline_Health.py b/pages/02_Pipeline_Health.py index 4c313d8..6977970 100644 --- a/pages/02_Pipeline_Health.py +++ b/pages/02_Pipeline_Health.py @@ -1,255 +1,111 @@ -"""Pipeline candidate β€” funnel intake β†’ analisi pubblica, salute CI, dettaglio.""" +"""Registry / Repo β€” stato dei registry di ogni repo del Lab.""" +import pandas as pd import streamlit as st -from sources import ( - data_freshness_note, - de_slug, - load_analysis_registry, - load_catalog, - load_explorer_datasets, - load_signals, -) - -st.title("βš™οΈ Pipeline candidate") +from sources import data_freshness_note, load_workspace_triage +st.title("πŸ“¦ Registry / Repo") st.markdown( - "Pipeline dei dataset del Lab: " - "dal candidate (dataset.yml + CI) alla pubblicazione in Explorer " - "fino all'analisi pubblica su dataciviclab.org. " - "Funnel end-to-end e dettaglio operativo in una vista." + "Stato dei registry di ogni repo del Lab: dataset, mart, segnali pipeline e salute complessiva." ) -# ── Carica dati ─────────────────────────────────────────────────────────────── -signals = load_signals() -catalog = load_catalog() - -sigs = signals.get("signals", []) -datasets = catalog.get("datasets", []) - -# Carica Explorer e Analisi per il funnel end-to-end -explorer_slugs = load_explorer_datasets() -analysis_map = load_analysis_registry() # {analysis_slug: dataset_slug} -analysis_dataset_slugs = set(analysis_map.values()) # dataset con analisi - -catalog_slugs = set(ds["slug"] for ds in datasets) - -# ── Indice segnali per slug (per lookup run) ────────────────────────────────── -signals_by_slug: dict[str, dict] = {} -for sig in sigs: - sig_slug = sig["id"].replace("-", "_") - signals_by_slug[sig_slug] = sig - -# ── Classifica candidate ────────────────────────────────────────────────────── -candidati = [] -for sig in sigs: - slug = sig["id"].replace("-", "_") - if slug not in catalog_slugs: - sr = sig.get("run", {}) or {} - candidati.append( - { - "slug": slug, - "id": sig["id"], - "label": sig.get("label", slug), - "source_id": sig.get("source_id", "?"), - "status": sig.get("status", "?"), - "detail": sig.get("detail", ""), - "checked_at": sr.get("checked_at", "?"), - "run_url": sr.get("run_url", ""), - "run_status": sr.get("status", ""), - "tipo": "compose" if sig["id"].startswith("compose:") else "candidate", - } - ) +# ── Carica dati ───────────────────────────────────────────────── +triage = load_workspace_triage() +registry_summary = triage.get("registry_summary", []) -incubazione = [] -published_datasets = [] -for ds in datasets: - stage = ds.get("stage", "") - slug = ds["slug"] - sig_data = signals_by_slug.get(slug, {}) - sr = sig_data.get("run", {}) or {} - item = { - "slug": slug, - "name": ds.get("name", ""), - "source_id": ds.get("source_id", "?"), - "description": ds.get("description", "")[:120], - "on_explorer": de_slug(slug) in explorer_slugs, - "has_analysis": slug in analysis_dataset_slugs, - "run_status": sr.get("status", ""), - "run_url": sr.get("run_url", ""), - "checked_at": sr.get("checked_at", ""), - } - if stage == "published": - published_datasets.append(item) - elif stage == "incubating": - incubazione.append(item) - -# Elenco completo di tutti i segnali con run falliti (candidate + catalogo) -all_failed = [] -for sig in sigs: - sr = sig.get("run", {}) or {} - if sr.get("status") == "failed": - slug = sig["id"].replace("-", "_") - all_failed.append( - { - "slug": slug, - "id": sig["id"], - "label": sig.get("label", slug), - "source_id": sig.get("source_id", "?"), - "detail": sig.get("detail", ""), - "checked_at": sr.get("checked_at", "?"), - "run_url": sr.get("run_url", ""), - "in_catalogo": slug in catalog_slugs, - } - ) +# Separa repo attivi da non-attivi +active = [r for r in registry_summary if r.get("available")] +inactive = [r for r in registry_summary if not r.get("available")] + +# ── KPI aggregati ─────────────────────────────────────────────── +tot_ds = sum(r.get("datasets", 0) for r in active) +tot_marts = sum(r.get("marts", 0) for r in active) +tot_signals = sum(r.get("signals", 0) for r in active) +tot_gcs = sum(r.get("gcs", 0) for r in active) -n_intake = len(candidati) -n_validation = len(incubazione) -n_published = len(published_datasets) -n_explorer = sum(1 for d in published_datasets if d["on_explorer"]) -n_analisi = sum(1 for d in published_datasets if d["has_analysis"]) -n_failed = len(all_failed) -n_compose = sum(1 for c in candidati if c["tipo"] == "compose") - -# ── Funnel ───────────────────────────────────────────────────────────────── -st.subheader("Funnel pipeline") - -ok_count = sum(1 for s in sigs if s.get("status") == "ok") -warn_count = sum(1 for s in sigs if s.get("status") == "warn") -err_count = sum(1 for s in sigs if s.get("status") == "error") -run_passed = sum(1 for s in sigs if s.get("run", {}).get("status") == "passed") -run_failed = sum(1 for s in sigs if s.get("run", {}).get("status") == "failed") -run_none = len(sigs) - run_passed - run_failed - -max_n = max(n_intake, n_validation, n_published, 1) -stages = [ - ("πŸ“₯ Intake", n_intake, "#94a3b8", f"{n_compose} compose"), - ("πŸ”¬ Incubazione", n_validation, "#3b82f6", ""), - ("βœ… Pubblicati", n_published, "#16a34a", ""), - ("🌐 Su Explorer", n_explorer, "#8b5cf6", ""), - ("πŸ“„ Con analisi", n_analisi, "#ec4899", ""), -] -for label, count, color, note in stages: - pct = count / max_n if max_n else 0 - r0, r1 = st.columns([2.5, 12]) - with r0: - st.write(f"**{label}**") - with r1: - note_html = f" Β· {note}" if note else "" - bar_html = f""" -
-
- {count}{note_html} -
-
- """ - st.markdown(bar_html, unsafe_allow_html=True) - -# Metriche in full width sotto +st.subheader("Panoramica") col1, col2, col3, col4 = st.columns(4) -col1.metric("βœ… Segnali OK", ok_count, "configurazione valida") -col2.metric("πŸƒ Run passati", run_passed, f"{run_passed}/{ok_count} segnali") -col3.metric( - "❌ Run falliti", - run_failed, - f"{round(run_failed / (run_passed + run_failed) * 100)}% dei run" if run_failed else "nessuno", -) -col4.metric("⏳ Mai eseguiti", run_none, "senza run CI") +col1.metric("πŸ“¦ Repo con registry", f"{len(active)}", f"{len(inactive)} senza registry") +col2.metric("πŸ“š Dataset totali", f"{tot_ds}") +col3.metric("πŸ“Š Mart totali", f"{tot_marts}") +col4.metric("πŸ“‘ Segnali totali", f"{tot_signals}") st.markdown("---") -# ── Dettaglio per tab ───────────────────────────────────────────────────────── -tab1, tab2, tab3 = st.tabs( - [ - f"πŸ“‹ Da rivedere ({n_failed})", - f"πŸ”¬ In corso ({n_intake + n_validation - n_failed})", - f"βœ… Completati ({n_published})", - ] -) - -with tab1: - if all_failed: - for f in all_failed: - badge = "🧩 compose" if f["id"].startswith("compose:") else "" - cat = "πŸ“¦ in catalogo" if f["in_catalogo"] else "πŸ“₯ candidate" - parts = [f"❌ **{f['label']}** β€” {f['source_id']}", cat] - if badge: - parts.append(badge) - with st.expander(" Β· ".join(parts)): - st.write(f"**Dettaglio:** {f['detail']}") - st.write(f"**Ultimo check:** {f['checked_at']}") - st.write("**Ultimo run:** ❌ fallito") - if f["run_url"]: - st.write(f"**Run CI:** [{f['run_url']}]({f['run_url']})") - else: - st.success("Nessun run fallito.") - -with tab2: - # Candidate intake (non falliti) - st.markdown(f"**πŸ“₯ Candidate intake ({n_intake - n_failed})**") - st.caption("dataset.yml + pipeline CI, in attesa di clean parquet su GCS") - for c in candidati: - if c["run_status"] == "failed": - continue - run_badge = {"passed": "βœ… passato", "": "βšͺ in attesa"}.get( - c["run_status"], "βšͺ sconosciuto" +# ── Tabella repo ──────────────────────────────────────────────── +st.subheader("Dettaglio repo") + +rows = [] +for r in active: + sigs = r.get("signals_detail", []) + ok = sum(1 for s in sigs if s.get("status") == "ok") + warn = sum(1 for s in sigs if s.get("status") == "warn") + error = sum(1 for s in sigs if s.get("status") == "error") + rows.append( + { + "repo": r["repo"], + "source_repo": r.get("source_repo", ""), + "datasets": r.get("datasets", 0), + "marts": r.get("marts", 0), + "signals": r.get("signals", 0), + "gcs": r.get("gcs", 0), + "ok": ok, + "warn": warn, + "error": error, + "updated_at": r.get("updated_at", ""), + } + ) + +df = pd.DataFrame(rows) + +if not df.empty: + st.dataframe( + df, + column_config={ + "repo": st.column_config.TextColumn("Repo", width="medium"), + "source_repo": st.column_config.TextColumn("Source repo", width="medium"), + "datasets": st.column_config.NumberColumn("Dataset", format="%d"), + "marts": st.column_config.NumberColumn("Mart", format="%d"), + "signals": st.column_config.NumberColumn("Segnali", format="%d"), + "gcs": st.column_config.NumberColumn("GCS", format="%d"), + "ok": st.column_config.NumberColumn("βœ…", format="%d"), + "warn": st.column_config.NumberColumn("⚠️", format="%d"), + "error": st.column_config.NumberColumn("❌", format="%d"), + "updated_at": st.column_config.TextColumn("Aggiornato", width="small"), + }, + hide_index=True, + width="stretch", + ) + + # Grafico a barre: dataset per repo + chart_df = df[["repo", "datasets"]].sort_values("datasets", ascending=False) + if not chart_df.empty: + st.subheader("Dataset per repo") + import altair as alt + + chart = ( + alt.Chart(chart_df) + .mark_bar(color="#3b82f6") + .encode( + y=alt.Y("repo:N", title=None, sort="-x"), + x=alt.X("datasets:Q", title="Dataset"), + tooltip=["repo", "datasets"], + ) + .properties(height=max(25 * len(chart_df), 100)) ) - e = {"ok": "βœ…", "warn": "⚠️", "error": "❌"}.get(c["status"], "❓") - tag = "🧩 compose" if c["tipo"] == "compose" else "" - title = f"{e} **{c['label']}** β€” run: {run_badge}" - if tag: - title += f" Β· {tag}" - with st.expander(title): - st.write(f"**Dettaglio:** {c['detail']}") - st.write(f"**Fonte:** {c['source_id']}") - st.write(f"**Ultimo check:** {c['checked_at']}") - if c["run_url"]: - st.write(f"**Run CI:** [{c['run_url']}]({c['run_url']})") + st.altair_chart(chart, width="stretch") +else: + st.info("Nessun registry disponibile.") +# ── Repo senza registry ───────────────────────────────────────── +if inactive: st.markdown("---") - st.markdown(f"**πŸ”¬ In incubazione ({n_validation})**") - st.caption("Clean parquet su GCS, in attesa di pubblicazione in Explorer") - for ds in incubazione: - with st.expander(f"πŸ”¬ **{ds['slug']}** β€” fonte: {ds['source_id']}"): - st.write(f"**Nome:** {ds['name']}") - st.write(f"**Descrizione:** {ds['description']}") - -with tab3: - for ds in published_datasets: - badges = [] - if ds["on_explorer"]: - badges.append("🌐 Explorer") - if ds["has_analysis"]: - badges.append("πŸ“„ Analisi") - badge_str = " Β· ".join(badges) if badges else "β€”" - with st.expander(f"βœ… **{ds['slug']}** β€” {badge_str}"): - st.write(f"**Nome:** {ds.get('name', '?')}") - st.write(f"**Descrizione:** {ds.get('description', '?')}") - - run_badge = {"passed": "βœ… passato", "failed": "❌ fallito", "": "βšͺ sconosciuto"}.get( - ds["run_status"], "βšͺ sconosciuto" - ) - st.write(f"**Ultimo run:** {run_badge}") - if ds["checked_at"]: - st.write(f"**Check:** {ds['checked_at']}") - if ds["run_url"]: - st.write(f"**Run CI:** [{ds['run_url']}]({ds['run_url']})") - - if ds["on_explorer"]: - de = de_slug(ds["slug"]) - st.write( - f"🌐 **Explorer:** " - f"[{de}](https://dataciviclab.github.io/data-explorer/dataset/{de})" - ) - if ds["has_analysis"]: - for a_slug, d_slug in analysis_map.items(): - if d_slug == ds["slug"]: - st.write( - f"πŸ“„ **Analisi:** [{a_slug}](https://dataciviclab.org/analisi/{a_slug})" - ) - break + st.subheader("Repo senza registry") + st.caption("Questi repo non hanno ancora un registry.json migrato.") + for r in inactive: + st.write(f"- **{r['repo']}** β€” {r.get('reason', 'registry_not_found')}") st.markdown("---") -st.caption("Dati: dataset-incubator (pipeline_signals + clean_catalog)") +st.caption("Dati: ACB (workspace_triage.json β†’ registry_summary)") data_freshness_note() diff --git a/pages/09_Query_SQL.py b/pages/09_Query_SQL.py index b5d2739..0cb4f10 100644 --- a/pages/09_Query_SQL.py +++ b/pages/09_Query_SQL.py @@ -1,334 +1,58 @@ -"""Query SQL interattiva sui dataset pubblici del DataCivicLab. +"""Query SQL interattiva sui dataset pubblici del DataCivicLab.""" -L'utente seleziona un dataset, scrive SQL su ``clean_input``, e ottiene risultati -in tempo reale via DuckDB che legge i Parquet su GCS β€” niente download, niente -pre-processing. Stesso pattern di clean-query-mcp: la CTE viene risolta -automaticamente sugli URL GCS per tutti gli anni del dataset. -""" - -from __future__ import annotations - -import time -from typing import Any - -import duckdb -import pandas as pd -import streamlit as st -from lab_connectors.gcs.paths import https_url +from lab_connectors.duckdb.sql_page import render_sql_query +from lab_connectors.registry import Column, Dataset, Location, Registry from sources import data_freshness_note, load_catalog -st.title("πŸ§ͺ Query SQL") -st.markdown( - "Scrivi query **SQL** sui dataset pubblici. " - "Usa ``clean_input`` come nome della tabella virtuale β€” " - "viene risolta automaticamente sui **Parquet GCS** per tutti gli anni " - "del dataset selezionato." -) - -# ── Helper (cached) ───────────────────────────────────────────────────────── - - -@st.cache_data(ttl=60, show_spinner=False) -def _resolve_slug(slug: str) -> tuple[list[str], str, dict[str, Any]]: - """Risolve slug β†’ (urls, cte_expr, dataset_info). - Gestisce due pattern dal catalogo: - - ``multi_file=True``: un Parquet per anno β†’ lista URL per tutti gli anni - - ``multi_file=False``: un unico file multi-anno β†’ URL dal path del catalogo +def _build_registry_from_acb() -> Registry: + """Build a Registry object from ACB topic_index data. - La CTE expression Γ¨ pronta per essere usata in: - WITH clean_input AS ({cte_expr}) SELECT ... + Only includes datasets with a valid GCS location (path non-empty). """ catalog = load_catalog() + datasets = [] for ds in catalog.get("datasets", []): - if ds["slug"] == slug: - loc = ds.get("location", {}) - multi = loc.get("multi_file", True) - - if multi: - # Un Parquet per anno: costruisce URL per ogni anno - period = ds.get("period", {}) - start = period.get("start") - end = period.get("end") - if not start or not end: - raise ValueError(f"Periodo non definito per '{slug}' nel catalogo") - years = list(range(start, end + 1)) - urls = [https_url("clean", "clean_parquet", slug=slug, year=y) for y in years] - else: - # Singolo file multi-anno: prende il path dal catalogo - gcs_path = loc.get("path", "") - if not gcs_path: - raise ValueError(f"Path non definito per '{slug}' nel catalogo") - # Converte gs://BUCKET/PATH β†’ https://storage.googleapis.com/BUCKET/PATH - https_path = gcs_path.replace("gs://", "https://storage.googleapis.com/", 1) - urls = [https_path] - - if len(urls) == 1: - cte_expr = f"SELECT * FROM read_parquet('{urls[0]}')" - else: - paths = "', '".join(urls) - cte_expr = f"SELECT * FROM read_parquet(['{paths}'])" - - return urls, cte_expr, ds - - raise ValueError(f"Dataset '{slug}' non trovato nel catalogo") - - -@st.cache_data(ttl=300, show_spinner=False) -def _get_schema_df(slug: str) -> pd.DataFrame: - """Schema colonne: da catalogo (se presente) o fallback via DESCRIBE.""" - catalog = load_catalog() - for ds in catalog.get("datasets", []): - if ds["slug"] == slug: - cols = ds.get("columns", []) - if cols: - return pd.DataFrame( - [ - { - "colonna": c.get("name", "?"), - "tipo": c.get("type", "?"), - "ruolo": c.get("role", "?"), - "descrizione": c.get("description", ""), - } - for c in cols - ] - ) - # Fallback: DESCRIBE dal primo parquet disponibile - try: - urls, _, _ = _resolve_slug(slug) - if urls: - with duckdb.connect() as con: - return con.sql(f"DESCRIBE SELECT * FROM read_parquet('{urls[0]}')").df() - except Exception: - pass - return pd.DataFrame() - return pd.DataFrame() - - -def _build_query(user_sql: str, cte_expr: str, max_rows: int) -> str: - """Avvolge la SQL utente nella CTE e applica il LIMIT.""" - return f"WITH clean_input AS ({cte_expr}) SELECT * FROM ({user_sql}) AS _q LIMIT {max_rows}" - - -def _default_sql(ds: dict[str, Any]) -> str: - """Query di esempio per il dataset selezionato.""" - period = ds.get("period", {}) - start = period.get("start", "?") - end = period.get("end", "?") - name = ds.get("name", ds.get("slug", "")) - cols = ds.get("columns", []) - col_hint = "" - if cols: - names = [c["name"] for c in cols[:5]] - col_hint = f"-- Colonne: {', '.join(names)}..." - return ( - f"-- Dataset: {name}\n" - f"-- Periodo: {start}–{end}\n" - f"{col_hint}\n" - f"-- Usa clean_input come tabella virtuale\n" - f"SELECT * FROM clean_input LIMIT 10" - ) - - -# ── Carica catalogo ───────────────────────────────────────────────────────── - -catalog = load_catalog() -datasets: list[dict[str, Any]] = catalog.get("datasets", []) - -if not datasets: - st.error("Catalogo non disponibile. Verifica connessione a GitHub.") - st.stop() - -slug_options = sorted(d["slug"] for d in datasets) -default_idx = slug_options.index("irpef_comunale") if "irpef_comunale" in slug_options else 0 - -# ── Toolbar: dataset e info ───────────────────────────────────────────────── - -col_sel, col_actions = st.columns([2, 3]) - -with col_sel: - selected_slug = st.selectbox( - "Dataset", - slug_options, - index=default_idx, - key="sql_query_slug", - ) - -ds_info = next((d for d in datasets if d["slug"] == selected_slug), None) -if not ds_info: - st.stop() - -with col_actions: - st.markdown("") # spacing - st.markdown("") # spacing - info_cols = st.columns([1, 1, 1]) - with info_cols[0]: - period = ds_info.get("period", {}) - st.markdown(f"**Anni:** {period.get('start', '?')}–{period.get('end', '?')}") - with info_cols[1]: - st.markdown(f"**Stage:** {ds_info.get('stage', 'β€”')}") - with info_cols[2]: - st.markdown(f"**Slug:** ``{selected_slug}``") - -# Schema + Info in expander compatti -exp_cols = st.columns([1, 1]) -with exp_cols[0]: - schema_df = _get_schema_df(selected_slug) - if not schema_df.empty: - with st.expander("Schema colonne", expanded=False): - st.dataframe( - schema_df, - hide_index=True, - width="stretch", - column_config={ - "colonna": "Colonna", - "tipo": "Tipo", - "ruolo": "Ruolo", - "descrizione": st.column_config.TextColumn("Descrizione", width="large"), - }, + loc_data = ds.get("location", {}) + path = loc_data.get("path", "") + if not path: + continue # Skip datasets without GCS location + is_multi = loc_data.get("multi_file", True) + location = Location(type=loc_data.get("type", "gcs"), path=path, multi_file=is_multi) + columns = [ + Column( + name=c.get("name", ""), + type=c.get("type", ""), + role=c.get("role", ""), + description=c.get("description", ""), ) -with exp_cols[1]: - desc = ds_info.get("description", "") - source = ds_info.get("source", "") - if desc or source: - with st.expander("Info dataset", expanded=False): - if desc: - st.markdown(f"**Descrizione:** {desc}") - if source: - st.markdown(f"**Fonte:** {source}") - -# ── Editor SQL ────────────────────────────────────────────────────────────── - -default_sql = _default_sql(ds_info) -sql = st.text_area( - "Scrivi la query SQL", - value=st.session_state.get("sql_query_sql", default_sql), - height=180, - key="sql_query_input", - placeholder="SELECT * FROM clean_input LIMIT 10", - help=( - "Usa clean_input come tabella virtuale.\n" - "WHERE, GROUP BY, ORDER BY, JOIN funzionano.\n" - "Per JOIN tra dataset usa read_parquet('url') diretto." + for c in ds.get("columns", []) + ] + datasets.append( + Dataset( + slug=ds.get("slug", ""), + name=ds.get("name", ""), + description=ds.get("description", ""), + source=ds.get("source", ""), + source_id=ds.get("source_id", ""), + period=ds.get("period", {}), + stage=ds.get("stage", ""), + columns=columns, + location=location, + ) + ) + return Registry(schema_version=1, repo="acb", datasets=datasets) + + +registry = _build_registry_from_acb() +render_sql_query( + registry=registry, + title="πŸ§ͺ Query SQL", + description=( + "Scrivi query **SQL** sui dataset pubblici. " + "Usa ``clean_input`` come nome della tabella virtuale β€” " + "viene risolta automaticamente sui **Parquet GCS**." ), ) - -# Opzioni esecuzione -col_max, col_btn, _ = st.columns([1, 1, 5]) -with col_max: - max_rows = st.number_input( - "Max righe", - min_value=1, - max_value=50_000, - value=1_000, - step=100, - ) -with col_btn: - execute = st.button(":material/play_arrow: Esegui", type="primary") - reset = st.button(":material/refresh: Reset", type="secondary") - -if reset: - st.session_state.sql_query_sql = _default_sql(ds_info) - st.rerun() - -# ── Storico query (prima dei risultati) ───────────────────────────────────── - -if "sql_history" not in st.session_state: - st.session_state.sql_history = [] - -if st.session_state.sql_history: - with st.expander("Storico query", expanded=False): - for i, entry in enumerate(st.session_state.sql_history[-8:]): - label = entry["sql"][:60].replace("\n", " ") - if len(entry["sql"]) > 60: - label += "…" - col_a, col_b = st.columns([6, 1]) - with col_a: - if st.button( - f"`{entry['slug']}` {label}", - key=f"hist_{i}", - help=f"{entry['rows']} righe Β· {entry['time']}", - ): - st.session_state.sql_query_sql = entry["sql"] - st.rerun() - with col_b: - st.caption(f"{entry['rows']} rows") - if st.button("Svuota storico", key="clear_hist"): - st.session_state.sql_history = [] - st.rerun() - -# ── Esecuzione query ──────────────────────────────────────────────────────── - -if execute: - st.session_state.sql_query_sql = sql - - with st.spinner(f"Esecuzione su `{selected_slug}` via DuckDB…"): - try: - # Risolvi slug β†’ URL GCS - urls, cte_expr, _ = _resolve_slug(selected_slug) - wrapped_sql = _build_query(sql, cte_expr, max_rows) - - # Esegui - t0 = time.perf_counter() - with duckdb.connect() as con: - df = con.sql(wrapped_sql).df() - elapsed = time.perf_counter() - t0 - - n_rows = len(df) - is_truncated = n_rows >= max_rows - - # Metriche - m1, m2, m3 = st.columns(3) - m1.metric("Righe restituite", f"{n_rows:,}") - m2.metric("Tempo esecuzione", f"{elapsed:.2f}s") - file_label = "1 file" if len(urls) == 1 else f"{len(urls)} file" - m3.metric("Parquet letti", file_label) - - if is_truncated: - st.info( - f"Risultato troncato a {max_rows} righe. " - "Aumenta il limite o aggiungi ``LIMIT`` nella query." - ) - - if n_rows == 0 and not is_truncated: - st.success("Query eseguita correttamente β€” **0 righe** restituite.") - elif n_rows > 0: - st.dataframe( - df, - width="stretch", - column_config={ - col: st.column_config.Column(col, width="medium") for col in df.columns[:8] - }, - ) - - # Download CSV - csv_data = df.to_csv(index=False).encode("utf-8") - st.download_button( - ":material/download: Scarica CSV", - data=csv_data, - file_name=f"{selected_slug}_query_{int(time.time())}.csv", - mime="text/csv", - ) - - # Storico - st.session_state.sql_history.append( - { - "slug": selected_slug, - "sql": sql, - "rows": n_rows, - "time": f"{elapsed:.2f}s", - } - ) - - # SQL eseguita in expander (debug) - with st.expander("SQL effettivamente eseguita", expanded=False): - st.code(wrapped_sql, language="sql") - - except Exception as e: - st.error(f"Errore durante l'esecuzione: {e}") - if "wrapped_sql" in locals(): - with st.expander("SQL che ha causato l'errore", expanded=True): - st.code(wrapped_sql, language="sql") - data_freshness_note() diff --git a/pyproject.toml b/pyproject.toml index 35b93b1..c1d57a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,34 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "lab-dashboard" +version = "0.1.0" +description = "Dashboard operativi interni DataCivicLab β€” pipeline, radar, metriche" +readme = "README.md" +license = "MIT" +requires-python = ">=3.12" +authors = [{name = "DataCivicLab"}] + +dependencies = [ + "streamlit>=1.61.1", + "altair>=6.2.2", + "pandas>=3.0.5", + "duckdb>=1.5.5", + "requests>=2.34.2", + "pyyaml>=6.0.3", + "lab-connectors[duckdb] @ git+https://github.com/dataciviclab/lab-connectors.git", + "agent-context-builder @ git+https://github.com/dataciviclab/agent-context-builder.git", +] + +[project.optional-dependencies] +dev = [ + "pytest>=9.1.1", + "pytest-cov>=7.1.0", + "ruff>=0.8.0", +] + [tool.ruff] line-length = 100 target-version = "py312" diff --git a/requirements.txt b/requirements.txt index 80bf4b4..16a3eb3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,4 +8,7 @@ pytest>=9.1.1 pytest-cov>=7.1.0 # Path contract GCS β€” vedere lab-connectors/gcs/paths.py -lab-connectors @ git+https://github.com/dataciviclab/lab-connectors.git +lab-connectors[duckdb] @ git+https://github.com/dataciviclab/lab-connectors.git + +# ACB artifacts (topic_index + workspace_triage) β€” fonte unica metadati +agent-context-builder @ git+https://github.com/dataciviclab/agent-context-builder.git diff --git a/sources.py b/sources.py index 208ec39..20766b0 100644 --- a/sources.py +++ b/sources.py @@ -1,17 +1,20 @@ """ Fonti dati condivise per il dashboard. -Legge da GitHub raw (metadati) e, opzionalmente, GCS parquet via DuckDB. + +Architettura: + ACB (2 JSON) β€” catalogo, radar, segnali, discussions, PR, issues, analyses. + SO direct (5 file) β€” radar history, source dashboard, source reports, + catalog signals, inventory report, check coverage. + GCS DuckDB β€” verify parquet. I path GCS seguono il path contract canonico definito in: lab-connectors/lab_connectors/gcs/paths.py (paths.json) - -I loader usano st.cache_data e mostrano errori con st.error() per robustezza -in produzione Streamlit. I fallback su dict/list vuoti evitano crash di pagina. """ -import os +from __future__ import annotations + from datetime import datetime, timezone -from typing import Any, Optional +from typing import Any import duckdb import pandas as pd @@ -24,32 +27,32 @@ LOGO_URL = "https://raw.githubusercontent.com/dataciviclab/lab-dashboard/main/static/logo.jpg" -REGISTRY_BASE = "https://raw.githubusercontent.com/dataciviclab/dataset-incubator/main/registry" +# ── URLs ────────────────────────────────────────────────────────────────────── +ACB_BASE = "https://raw.githubusercontent.com/dataciviclab/agent-context-builder/context" +TOPIC_INDEX_URL = f"{ACB_BASE}/topic_index.json" +WORKSPACE_TRIAGE_URL = f"{ACB_BASE}/workspace_triage.json" SO_BASE = "https://raw.githubusercontent.com/dataciviclab/source-observatory/main" GCS_BASE = f"https://storage.googleapis.com/{CLEAN_BUCKET}" - -# ── Data fetching ───────────────────────────────────────────────────────────────── +# ── HTTP session ────────────────────────────────────────────────────────────── _LAST_FETCH: dict[str, datetime] = {} - _HTTP = requests.Session() _HTTP.mount( "https://", HTTPAdapter( - max_retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504]), + max_retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504]) ), ) _HTTP.mount( "http://", HTTPAdapter( - max_retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504]), + max_retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504]) ), ) def _fetch_json(url: str) -> Any: - """Fetch JSON. Solleva eccezioni β€” la UI gestisce l'errore.""" r = _HTTP.get(url, timeout=15) r.raise_for_status() _LAST_FETCH[url] = datetime.now(timezone.utc) @@ -57,254 +60,199 @@ def _fetch_json(url: str) -> Any: def _fetch_yaml(url: str) -> dict: - """Fetch YAML. Solleva eccezioni β€” la UI gestisce l'errore.""" r = _HTTP.get(url, timeout=15) r.raise_for_status() _LAST_FETCH[url] = datetime.now(timezone.utc) return yaml.safe_load(r.text) or {} -# ── Caricatori con cache β€” errori mostrati nella UI ────────────────────────────── +# ══════════════════════════════════════════════════════════════════════════════ +# ACB loaders +# ══════════════════════════════════════════════════════════════════════════════ + + @st.cache_data(ttl=300, show_spinner=False) -def load_catalog(): - """Catalogo dataset dal registry fusion (registry.json).""" +def load_topic_index() -> dict[str, Any]: try: - reg = _fetch_json(f"{REGISTRY_BASE}/registry.json") - return reg + return _fetch_json(TOPIC_INDEX_URL) except Exception as e: - st.error(f"❌ Catalogo non disponibile: {e}") + st.error(f"❌ Topic index non disponibile: {e}") return {} @st.cache_data(ttl=300, show_spinner=False) -def load_signals(): - """Segnali pipeline dal registry fusion (registry.json). - - Il blocco ``run`` del registry viene esposto direttamente (chiave ``run``): - i campi ``checked_at``/``run_url`` non esistono nel registry, si derivano - da ``started_at`` e ``run_id``. - """ +def load_workspace_triage() -> dict[str, Any]: try: - reg = _fetch_json(f"{REGISTRY_BASE}/registry.json") - signals = [] - for s in reg.get("signals", []): - sig = dict(s) - run = s.get("run") or {} - if run: - sig["run"] = { - "status": "passed" - if run.get("status") == "SUCCESS" - else run.get("status", "").lower(), - "run_id": run.get("run_id", ""), - "checked_at": (run.get("started_at") or "")[:10], - "run_url": f"https://github.com/dataciviclab/dataset-incubator/actions/runs/{run.get('run_id', '')}" - if run.get("run_id") - else "", - "year": run.get("year"), - } - signals.append(sig) - return {"schema_version": reg.get("schema_version", "1"), "signals": signals} + return _fetch_json(WORKSPACE_TRIAGE_URL) except Exception as e: - st.error(f"❌ Segnali pipeline non disponibili: {e}") - return {"signals": []} + st.error(f"❌ Workspace triage non disponibile: {e}") + return {} @st.cache_data(ttl=300, show_spinner=False) -def load_radar(): - try: - return _fetch_json(f"{SO_BASE}/data/radar/radar_summary.json") - except Exception as e: - st.error(f"❌ Radar fonti non disponibile: {e}") - return {} +def load_catalog() -> dict[str, Any]: + """Catalogo dataset β€” tutti i dataset da tutti i repo, con details.""" + ti = load_topic_index() + all_datasets = [] + for source, ds_list in ti.get("datasets", {}).items(): + for ds in ds_list: + entry = dict(ds) + entry["source"] = source + all_datasets.append(entry) + return {"datasets": all_datasets} @st.cache_data(ttl=300, show_spinner=False) -def load_sources_registry(): - try: - return _fetch_yaml(f"{SO_BASE}/data/radar/sources_registry.yaml") - except Exception as e: - st.error(f"❌ Registro fonti non disponibile: {e}") - return {} +def load_signals() -> dict[str, Any]: + """Segnali pipeline β€” da registry_summary.signals_detail.""" + triage = load_workspace_triage() + signals = [] + for repo_info in triage.get("registry_summary", []): + for sig in repo_info.get("signals_detail", []): + entry = dict(sig) + if "run" not in entry or entry["run"] is None: + entry["run"] = {} + signals.append(entry) + return { + "schema_version": "2", + "signals": signals, + "pipeline_state": triage.get("pipeline_state", {}), + } @st.cache_data(ttl=300, show_spinner=False) -def load_radar_history(): - """ - Storico probe radar: transizioni stato per fonte. - Usato in 05_Radar.py per timeline chart. - """ - try: - return _fetch_json(f"{SO_BASE}/data/radar/radar_history.json") - except Exception as e: - st.error(f"❌ Storico radar non disponibile: {e}") - return {"probes": []} +def load_radar() -> dict[str, Any]: + """Radar fonti β€” 36 fonti da workspace_triage.radar.""" + triage = load_workspace_triage() + radar = triage.get("radar", {}) + return { + "generated_at": radar.get("generated_at", ""), + "probe_date": radar.get("probe_date", ""), + "sources_total": radar.get("sources_total", 0), + "status_counts": { + "GREEN": radar.get("green", 0), + "YELLOW": radar.get("yellow", 0), + "RED": radar.get("red", 0), + }, + "persistent_red": radar.get("persistent_red", 0), + "sources": radar.get("sources", []), + } @st.cache_data(ttl=300, show_spinner=False) -def load_catalog_signals(): - """ - Segnali inventariali SO (report v1): signal_type, result, metric_value per fonte. - Usato in 06_Inventario.py per il badge segnale. - """ - try: - return _fetch_json(f"{SO_BASE}/data/catalog/catalog_signals.json") - except Exception as e: - st.error(f"❌ Segnali catalogo non disponibili: {e}") - return {"signals": []} +def load_explorer_datasets() -> set[str]: + """Dataset slug Explorer β€” da topic_index.explorer_themes.""" + ti = load_topic_index() + slugs: set[str] = set() + for t in ti.get("explorer_themes", []): + slugs.update(t.get("datasets", [])) + return slugs @st.cache_data(ttl=300, show_spinner=False) -def load_sources_dashboard(): - """ - Report consolidato fonti SO (report v2): per ogni fonte verdict, readiness, - datasets_in_use, items inventory/scored/reachable. - Artifact canonico: source-observatory/data/reports/sources_dashboard.json - """ - try: - return _fetch_json(f"{SO_BASE}/data/reports/sources_dashboard.json") - except Exception as e: - st.error(f"❌ Dashboard fonti non disponibile: {e}") - return {"sources": []} +def load_discussion_counts() -> dict[str, int]: + """Discussioni per categoria β€” da workspace_triage.discussions.""" + triage = load_workspace_triage() + counts: dict[str, int] = {} + for d in triage.get("discussions", []): + cat = d.get("category", "Senza categoria") + counts[cat] = counts.get(cat, 0) + 1 + return counts @st.cache_data(ttl=300, show_spinner=False) -def load_source_report(source_id: str): - """ - Report per-fonte SO (report v1): health, inventory (con drift vs baseline), - source_check, datasets_in_use, signals, operational_verdict. - Artifact canonico: source-observatory/data/reports/source_reports/{source_id}.json - """ +def load_discussions() -> list[dict[str, Any]]: + """Discussioni recenti β€” da workspace_triage.discussions.""" + triage = load_workspace_triage() + return triage.get("discussions", [])[:15] + + +# ══════════════════════════════════════════════════════════════════════════════ +# SO direct loaders +# ══════════════════════════════════════════════════════════════════════════════ + + +@st.cache_data(ttl=300, show_spinner=False) +def load_radar_history() -> dict[str, Any]: try: - return _fetch_json(f"{SO_BASE}/data/reports/source_reports/{source_id}.json") - except Exception as e: - st.error(f"❌ Report fonte '{source_id}' non disponibile: {e}") + return _fetch_json(f"{SO_BASE}/data/radar/radar_history.json") + except Exception: return {} @st.cache_data(ttl=300, show_spinner=False) -def load_inventory_report(): - """ - Report inventario SO da GCS: stato build, righe, errore per fonte. - Usato in 05_Radar.py e 07_Fonti.py per badge βœ…/❌ e tabella fonti. - """ +def load_sources_registry() -> dict[str, Any]: + """Registro fonti β€” da SO sources_registry.yaml (per-source: protocol, observation_mode).""" try: - return _fetch_json(https_url("clean", "catalog_inventory_report")) - except Exception as e: - st.error(f"❌ Report inventario non disponibile: {e}") + return _fetch_yaml(f"{SO_BASE}/data/radar/sources_registry.yaml") + except Exception: return {} @st.cache_data(ttl=300, show_spinner=False) -def load_check_coverage(): - """ - Items in inventario vs items checked per fonte, via DuckDB su GCS parquet. - Incrocia catalog_inventory_latest.parquet con source_check_results.parquet. - Ritorna DataFrame con: source_id, inv_items, chk_items, reachable, candidates. - """ +def load_sources_dashboard() -> dict[str, Any]: try: - inv_url = https_url("clean", "catalog_inventory_latest") - chk_url = https_url("clean", "catalog_inventory_source_check") - with duckdb.connect() as con: - return con.sql(f""" - SELECT COALESCE(i.source_id, c.source_id) AS source_id, - COALESCE(i.inv_items, 0)::BIGINT AS inv_items, - COALESCE(c.chk_items, 0)::BIGINT AS chk_items, - COALESCE(c.reachable, 0)::BIGINT AS reachable, - COALESCE(c.candidates, 0)::BIGINT AS candidates - FROM (SELECT source_id, COUNT(*) AS inv_items - FROM read_parquet('{inv_url}') GROUP BY source_id) i - FULL JOIN (SELECT source_id, - COUNT(*) AS chk_items, - SUM(CASE WHEN reachable THEN 1 ELSE 0 END) AS reachable, - SUM(CASE WHEN intake_candidate THEN 1 ELSE 0 END) AS candidates - FROM read_parquet('{chk_url}') GROUP BY source_id) c - ON i.source_id = c.source_id - ORDER BY inv_items DESC - """).df() - except Exception as e: - st.error(f"❌ Check coverage non disponibile: {e}") - return pd.DataFrame() + return _fetch_json(f"{SO_BASE}/data/reports/sources_dashboard.json") + except Exception: + return {} -def last_fetch_time() -> Optional[datetime]: - if not _LAST_FETCH: - return None - return max(_LAST_FETCH.values()) +@st.cache_data(ttl=300, show_spinner=False) +def load_source_report(source_id: str) -> dict[str, Any]: + try: + return _fetch_json(f"{SO_BASE}/data/reports/source_reports/{source_id}.json") + except Exception: + return {} -def data_freshness_note(): - """Mostra nota 'dati caricati al ...' nella pagina chiamante.""" - t = last_fetch_time() - if t: - st.caption(f"πŸ“‘ Dati caricati: {t.strftime('%d/%m/%Y %H:%M')} UTC") +@st.cache_data(ttl=300, show_spinner=False) +def load_catalog_signals() -> dict[str, Any]: + """Segnali catalogo β€” da SO catalog_signals.json (per-source: result, metric_value).""" + try: + return _fetch_json(f"{SO_BASE}/data/catalog/catalog_signals.json") + except Exception: + return {} -# ── GitHub Discussions ──────────────────────────────────────────────────────────── -def _github_token(): - """Ritorna GITHUB_TOKEN da st.secrets o env. None se assente.""" +@st.cache_data(ttl=600, show_spinner=False) +def load_inventory_report() -> dict[str, Any]: try: - return st.secrets.get("github_token") or os.environ.get("GITHUB_TOKEN") + return _fetch_json(f"{GCS_BASE}/catalog_inventory/catalog_inventory_report.json") except Exception: - return os.environ.get("GITHUB_TOKEN") - - -def load_discussion_counts(): - """ - Ritorna conteggi per categoria: {'totale': N, 'domande': N, 'analisi': N, ...} - """ - token = _github_token() - if not token: - return {"totale": 0, "domande": 0, "analisi": 0} - - query = { - "query": """{ - repository(owner: "dataciviclab", name: "dataciviclab") { - totale: discussions(first: 0) { totalCount } - } - }""" - } + return {} + +@st.cache_data(ttl=600, show_spinner=False) +def load_check_coverage() -> pd.DataFrame: try: - r = requests.post( - "https://api.github.com/graphql", - json=query, - headers={"Authorization": f"bearer {token}"}, - timeout=10, - ) - data = r.json() - total = data["data"]["repository"]["totale"]["totalCount"] - return {"totale": total, "domande": "?", "analisi": "?"} + url = f"{GCS_BASE}/catalog_inventory/catalog_inventory_latest.parquet" + with duckdb.connect() as con: + return con.sql( + "SELECT source_id, inv_items, chk_items FROM read_parquet(?)", params=[url] + ).df() except Exception: - return {"totale": 0, "domande": 0, "analisi": 0} + return pd.DataFrame(columns=["source_id", "inv_items", "chk_items"]) + + +# ══════════════════════════════════════════════════════════════════════════════ +# GCS helpers +# ══════════════════════════════════════════════════════════════════════════════ -# ── DuckDB (opzionale β€” attualmente usato solo per verifica spot) ──────────────── def duckdb_query(sql: str) -> pd.DataFrame: - """Esegue SQL su DuckDB (in-memory). Chiude la connessione al termine.""" with duckdb.connect() as con: return con.sql(sql).df() -def verify_parquet(slug: str, year: int) -> dict: - """ - Verifica se un parquet GCS esiste e ha dati. - Usa parametri DuckDB, non f-string, per evitare SQL injection. - Ritorna {'slug': ..., 'year': ..., 'records': N} o solleva eccezione. - """ +def verify_parquet(slug: str, year: int) -> dict[str, Any]: path = https_url("clean", "clean_parquet", slug=slug, year=year) with duckdb.connect() as con: df = con.sql("SELECT COUNT(*) AS records FROM read_parquet(?)", params=[path]).df() - records = int(df["records"].iloc[0]) - return {"slug": slug, "year": year, "records": records} - + return {"slug": slug, "year": year, "records": int(df["records"].iloc[0])} -# ── Explorer e Analisi (per pipeline end-to-end) ────────────────────────────── -DE_BASE = "https://raw.githubusercontent.com/dataciviclab/data-explorer/main/src/data" -DCL_BASE = "https://raw.githubusercontent.com/dataciviclab/dataciviclab/main/analisi" +# ── Utilities ───────────────────────────────────────────────────────────────── - -# Mapping slug dataset-incubator β†’ slug data-explorer (pochi casi con nome diverso). DE_SLUG_MAP = { "aifa_spesa_consumo": "spesa-farmaceutica", "ispra_ru_base": "rifiuti-urbani", @@ -317,83 +265,10 @@ def verify_parquet(slug: str, year: int) -> dict: def de_slug(di_slug: str) -> str: - """Converti slug dataset-incubator β†’ slug data-explorer.""" return DE_SLUG_MAP.get(di_slug, di_slug.replace("_", "-")) -@st.cache_data(ttl=3600, show_spinner=False) -def load_explorer_datasets() -> set[str]: - """Dataset slug DE presenti su data-explorer. - - Scarica e fa il parse di ``themes.json.py`` usando ``ast.literal_eval`` - (sicuro: nessuna esecuzione di codice remoto). Restituisce l'insieme - di tutti gli slug DE presenti negli themes. - """ - import ast - - try: - r = _HTTP.get(f"{DE_BASE}/themes.json.py", timeout=15) - r.raise_for_status() - # themes.json.py contiene anche ``json.dump(themes, ...)`` dopo l'array. - # Usiamo AST per estrarre solo il nodo ``themes`` senza eseguire codice. - module = ast.parse(r.text) - themes = None - for node in module.body: - if isinstance(node, ast.Assign): - for target in node.targets: - if isinstance(target, ast.Name) and target.id == "themes": - themes = ast.literal_eval(node.value) - break - if themes is not None: - break - if themes is None: - return set() - slugs: set[str] = set() - for t in themes: - slugs.update(t.get("datasets", [])) - return slugs - except Exception: - # Fallback silenzioso: upstream irraggiungibile - return set() - - -@st.cache_data(ttl=3600, show_spinner=False) -def load_analysis_registry() -> dict[str, str]: - """Mappa slug analisi β†’ slug dataset (da README frontmatter). - - Usa GitHub API per listare le directory in ``analisi/``, poi legge - il ``dataset_slug`` dal frontmatter YAML di ogni README.md. - Restituisce {analysis_slug: dataset_slug}. - """ - try: - r = _HTTP.get( - "https://api.github.com/repos/dataciviclab/dataciviclab/contents/analisi", - timeout=15, - ) - r.raise_for_status() - items = r.json() - except Exception: - # Fallback silenzioso: upstream irraggiungibile - return {} - - registry: dict[str, str] = {} - for item in items: - if item["type"] != "dir": - continue - slug = item["name"] - if slug in ("registry", "_template"): - continue - - # Legge README.md e cerca dataset_slug nel frontmatter - try: - rr = _HTTP.get(f"{DCL_BASE}/{slug}/README.md", timeout=10) - rr.raise_for_status() - for line in rr.text.splitlines(): - if line.startswith("dataset_slug:"): - ds_slug = line.split(":", 1)[1].strip() - registry[slug] = ds_slug - break - except Exception: - pass - - return registry +def data_freshness_note() -> None: + if _LAST_FETCH: + t = max(_LAST_FETCH.values()) + st.caption(f"πŸ“‘ Dati caricati: {t.strftime('%d/%m/%Y %H:%M')} UTC") diff --git a/tests/test_sources.py b/tests/test_sources.py index 7cd5250..1a917e7 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -1,14 +1,9 @@ """ -Test per sources.py β€” loader, fetching e fallback. -Non testa pagine Streamlit (troppo dipendenti dal runtime). +Test per sources.py β€” loader ACB-based e SO-direct. -Contratto: i loader () producono dict/list strutturati da GitHub raw. - _fetch_json/_fetch_yaml gestiscono successo/errore HTTP. - I loader hanno fallback su dict/list vuoti quando HTTP fallisce. - La serializzazione e' controllata da st.cache_data. - -Prova del fuoco: se cancello questi test, un refactor di sources.py puo' -rompere tutti i 9 loader che alimentano il dashboard. +Contratto: i loader ACB producono dict/list strutturati da topic_index.json + e workspace_triage.json. I fallback su dict/list vuoti quando fetch fallisce. + I loader SO diretti mantengono il comportamento legacy. """ import json @@ -19,36 +14,172 @@ from sources import ( _fetch_json, _fetch_yaml, - _github_token, de_slug, - duckdb_query, - load_analysis_registry, load_catalog, load_catalog_signals, + load_discussion_counts, load_explorer_datasets, - load_inventory_report, load_radar, - load_radar_history, load_signals, - load_source_report, - load_sources_dashboard, load_sources_registry, - verify_parquet, ) -# ── Helpers ───────────────────────────────────────────────────────────────── +# ── Mock data ───────────────────────────────────────────────────────────────── + +_MOCK_TOPIC_INDEX = { + "schema_version": 4, + "generated_at": "2026-08-27T10:00:00", + "repos": { + "dataciviclab": { + "description": "Hub", + "url": "https://github.com/dataciviclab/dataciviclab", + }, + "dataset-incubator": { + "description": "Incubation", + "url": "https://github.com/dataciviclab/dataset-incubator", + }, + }, + "datasets": { + "Agenzia delle Entrate": [ + { + "slug": "ade_cinque_per_mille", + "name": "5x1000", + "period": {"start": 2023, "end": 2025}, + "stage": "published", + }, + ], + "ANAC": [ + { + "slug": "anac_bandi_gara", + "name": "Bandi gara", + "period": {"start": 2015, "end": 2024}, + "stage": "published", + }, + { + "slug": "anac_smartcig", + "name": "SmartCIG", + "period": {"start": 2020, "end": 2024}, + "stage": "incubating", + }, + ], + }, + "explorer_themes": [ + { + "slug": "finanza-pubblica", + "name": "Finanza pubblica", + "datasets": ["ade_cinque_per_mille"], + }, + ], + "analyses": [ + { + "slug": "cinque-per-mille", + "name": "5x1000", + "datasets": ["ade_cinque_per_mille"], + "status": "active", + }, + ], + "analyses_by_dataset": {"ade_cinque_per_mille": ["cinque-per-mille"]}, + "operational_topics": {}, +} +_MOCK_WORKSPACE_TRIAGE = { + "generated_at": "2026-08-27T10:00:00", + "repos": ["dataciviclab", "dataset-incubator"], + "radar": { + "available": True, + "probe_date": "2026-08-26", + "sources_total": 36, + "green": 34, + "yellow": 1, + "red": 1, + "persistent_red": 1, + "sources": [ + { + "id": "istat_sdmx", + "status": "GREEN", + "protocol": "sdmx", + "http_code": "200", + "note": "", + "red_streak": 0, + }, + { + "id": "ispra_linked_data", + "status": "RED", + "protocol": "sparql", + "http_code": "-", + "note": "Connection error", + "red_streak": 14, + }, + ], + "unhealthy": [ + { + "id": "ispra_linked_data", + "status": "RED", + "protocol": "sparql", + "note": "Connection error", + "red_streak": 14, + }, + ], + }, + "source_health": { + "available": True, + "captured_at": "2026-08-26", + "sources_checked": 36, + "regressions": [], + "alerts": [], + }, + "pipeline_state": { + "available": True, + "generated_at": "2026-08-16", + "summary": {"total": 100, "by_status": {"ok": 100}}, + "actionable": [], + }, + "registry_summary": [ + { + "repo": "dataset-incubator", + "datasets": 92, + "marts": 149, + "signals": 100, + "source_repo": "dataciviclab/dataset-incubator", + "updated_at": "2026-08-16", + "signals_detail": [ + { + "id": "aci_prime", + "source_id": "aci", + "status": "ok", + "label": "ACI", + "detail": "ok", + }, + { + "id": "ispra_ru", + "source_id": "ispra", + "status": "ok", + "label": "ISPRA", + "detail": "ok", + }, + ], + }, + { + "repo": "eurostat", + "datasets": 30, + "marts": 90, + "signals": 30, + "source_repo": "dataciviclab/eurostat", + "updated_at": "2026-08-16", + "signals_detail": [], + }, + ], + "discussions": [ + {"title": "Test discussion", "number": 1, "repo": "dataciviclab", "category": "Domande"}, + ], + "prs": [], + "issues": [], + "git_state": {}, + "warnings": [], +} -def _py_resp(source: str, status: int = 200) -> MagicMock: - """Mock response per file Python (es. themes.json.py).""" - m = MagicMock() - m.status_code = status - m.text = source - if status >= 400: - m.raise_for_status.side_effect = Exception(f"HTTP {status}") - else: - m.raise_for_status.return_value = None - return m + +# ── Helpers ───────────────────────────────────────────────────────────────── def _resp(data, status=200): @@ -111,400 +242,156 @@ def test_http_error(self): _fetch_yaml(self.URL) -# ── Loader fallback ───────────────────────────────────────────────────────── - - -LOADERS = [ - ("load_radar", load_radar, {}), - ("load_radar_history", load_radar_history, {"probes": []}), - ("load_catalog_signals", load_catalog_signals, {"signals": []}), - ("load_sources_dashboard", load_sources_dashboard, {"sources": []}), - ("load_inventory_report", load_inventory_report, {}), - ("load_catalog", load_catalog, {}), - ("load_signals", load_signals, {"signals": []}), -] +# ── ACB-based loaders ────────────────────────────────────────────────────── @pytest.mark.contract -def test_load_source_report_fallback_on_http_error(): - """load_source_report deve ritornare {} quando HTTP fallisce (502).""" - with patch("sources._HTTP.get", return_value=_resp({}, status=502)): - result = load_source_report("anac") - assert result == {} +class TestLoadCatalog: + """Contratto: load_catalog() produce {datasets: [...]} da topic_index.""" + + def test_flattens_datasets_from_topic_index(self): + with patch("sources._fetch_json", return_value=_MOCK_TOPIC_INDEX): + result = load_catalog() + datasets = result["datasets"] + assert len(datasets) == 3 # 1 ADE + 2 ANAC + slugs = {ds["slug"] for ds in datasets} + assert "ade_cinque_per_mille" in slugs + assert "anac_bandi_gara" in slugs + # 每δΈͺ dataset ha campo source + for ds in datasets: + assert "source" in ds + + def test_returns_empty_on_error(self): + with patch("sources._fetch_json", side_effect=Exception("fail")): + result = load_catalog() + assert result == {"datasets": []} @pytest.mark.contract -@pytest.mark.parametrize("name,loader,expected_fallback", LOADERS) -def test_loader_fallback_on_http_error(name, loader, expected_fallback): - """Ogni loader deve ritornare fallback quando HTTP fallisce (502).""" - with patch("sources._HTTP.get", return_value=_resp({}, status=502)): - result = loader() - assert result == expected_fallback - - -# ── Loader risposta positiva ──────────────────────────────────────────────── - - -RADAR_SAMPLE = { - "sources_total": 23, - "sources": [{"id": "istat_sdmx", "status": "GREEN", "protocol": "sdmx"}], - "status_counts": {"GREEN": 18, "YELLOW": 4, "RED": 1}, -} - -RADAR_HISTORY_SAMPLE = { - "probes": [{"probe_date": "2026-05-18", "sources": [{"id": "istat_sdmx", "status": "GREEN"}]}] -} - -SIGNALS_SAMPLE = { - "signals": [ - { - "source_id": "aifa", - "signal_type": "validated_metrics", - "result": "stable", - "metric_value": 62, - "detail": "reachable=96.8%", - "suggested_action": None, - } - ] -} +class TestLoadSignals: + """Contratto: load_signals() produce {signals: [...]} da workspace_triage.""" -SOURCES_DASHBOARD_SAMPLE = { - "generated_at": "2026-08-03T06:48:44+00:00", - "report_version": 2, - "total_sources": 36, - "summary": {"tot_inventory_items": 15139}, - "sources": [ - { - "source_id": "anac", - "protocol": "ckan", - "radar": "GREEN", - "inventory_items": 70, - "scored_items": 48, - "reachable": 47, - "avg_readiness": 7.6, - "datasets_in_use": 8, - "verdict": "STABLE", - "last_inventory": "2026-08-03T06:41:09+00:00", - } - ], -} + def test_builds_signals_from_registry_summary(self): + with patch("sources._fetch_json", return_value=_MOCK_WORKSPACE_TRIAGE): + result = load_signals() + signals = result["signals"] + assert len(signals) == 2 # 2 signals_detail from dataset-incubator + ids = {s["id"] for s in signals} + assert "aci_prime" in ids + assert "ispra_ru" in ids -SOURCE_REPORT_SAMPLE = { - "source_id": "istat_sdmx", - "report_version": 1, - "identity": {"protocol": "sdmx", "observation_mode": "catalog-watch", "verdict": "go"}, - "health": {"radar_status": "GREEN", "http_code": "200"}, - "inventory": { - "total_items": 4899, - "method": "dataflow_count", - "baseline_value": 4212, - "baseline_date": "2026-04-10", - "delta": 687, - "delta_pct": 16.3, - }, - "source_check": {"total_scored": 3574, "reachable": 3574, "avg_readiness": 5.0}, - "datasets_in_use": [{"slug": "istat_gini_regionale", "status": "published"}], - "operational_verdict": { - "label": "INVENTORY_CHANGED", - "next_action": "review inventory changes", - }, -} - -INVENTORY_SAMPLE = { - "sources": {"istat_sdmx": {"status": "ok", "rows": 4849, "method": "dataflow_count"}} -} - -# Registry fusion (registry.json) β€” fonte unica per load_catalog/load_signals -REGISTRY_SAMPLE = { - "schema_version": 1, - "datasets": [{"slug": "test", "name": "Test", "stage": "published", "period": {}}], - "signals": [ - { - "id": "test", - "status": "ok", - "run": { - "run_id": "20260101T000000Z_abc", - "year": 2025, - "status": "SUCCESS", - "started_at": "2026-01-01T00:00:00+00:00", - }, - } - ], -} - -REGISTRY_SAMPLE_YAML = """istat_sdmx: - protocol: sdmx - verdict: go - observation_mode: catalog-watch -""" - - -@pytest.mark.contract -class TestLoaderSuccess: - @patch("sources._HTTP.get", return_value=_resp(RADAR_SAMPLE)) - def test_load_radar(self, mock_get): - result = load_radar() - assert result["sources_total"] == 23 - - @patch("sources._HTTP.get", return_value=_resp(RADAR_HISTORY_SAMPLE)) - def test_load_radar_history(self, mock_get): - result = load_radar_history() - assert len(result["probes"]) == 1 - - @patch("sources._HTTP.get", return_value=_resp(SIGNALS_SAMPLE)) - def test_load_catalog_signals(self, mock_get): - result = load_catalog_signals() - assert len(result["signals"]) == 1 - assert result["signals"][0]["source_id"] == "aifa" - - @patch("sources._HTTP.get", return_value=_resp(SOURCES_DASHBOARD_SAMPLE)) - def test_load_sources_dashboard(self, mock_get): - result = load_sources_dashboard() - assert result["report_version"] == 2 - assert len(result["sources"]) == 1 - assert result["sources"][0]["verdict"] == "STABLE" - assert result["sources"][0]["avg_readiness"] == 7.6 - - @patch("sources._HTTP.get", return_value=_resp(SOURCE_REPORT_SAMPLE)) - def test_load_source_report(self, mock_get): - result = load_source_report("istat_sdmx") - assert result["source_id"] == "istat_sdmx" - assert result["inventory"]["delta"] == 687 - assert result["operational_verdict"]["next_action"] == "review inventory changes" - - @patch("sources._HTTP.get", return_value=_resp(INVENTORY_SAMPLE)) - def test_load_inventory_report(self, mock_get): - result = load_inventory_report() - assert result["sources"]["istat_sdmx"]["rows"] == 4849 - - @patch("sources._HTTP.get", return_value=_resp(REGISTRY_SAMPLE)) - def test_load_catalog(self, mock_get): - result = load_catalog() - assert len(result["datasets"]) == 1 - - @patch("sources._HTTP.get", return_value=_resp(REGISTRY_SAMPLE)) - def test_load_signals(self, mock_get): - result = load_signals() - assert len(result["signals"]) == 1 - # il blocco run del registry viene esposto come run (status normalizzato) - assert result["signals"][0]["run"]["status"] == "passed" - - @patch("sources._HTTP.get", return_value=_yaml_resp(REGISTRY_SAMPLE_YAML)) - def test_load_sources_registry(self, mock_get): - result = load_sources_registry() - assert result["istat_sdmx"]["verdict"] == "go" - - -# ── _github_token ─────────────────────────────────────────────────────────── + def test_includes_pipeline_state(self): + with patch("sources._fetch_json", return_value=_MOCK_WORKSPACE_TRIAGE): + result = load_signals() + assert "pipeline_state" in result + assert result["pipeline_state"]["summary"]["total"] == 100 @pytest.mark.contract -class TestGithubToken: - def test_from_secrets(self): - with patch("sources.st.secrets", {"github_token": "tok-secret"}): - with patch("sources.os.environ", {}): - assert _github_token() == "tok-secret" - - def test_from_env(self): - with patch("sources.st.secrets", {}): - with patch("sources.os.environ", {"GITHUB_TOKEN": "tok-env"}): - assert _github_token() == "tok-env" - - def test_secrets_overrides_env(self): - with patch("sources.st.secrets", {"github_token": "tok-secret"}): - with patch("sources.os.environ", {"GITHUB_TOKEN": "tok-env"}): - assert _github_token() == "tok-secret" - - @pytest.mark.policy - def test_returns_none_when_missing(self): - """Senza token ne' in secrets ne' in env β†’ None.""" - with patch("sources.st.secrets", {}): - with patch("sources.os.environ", {}): - assert _github_token() is None - - @pytest.mark.policy - def test_handles_secrets_exception(self): - """st.secrets puo' sollevare Exception (es. in ambiente senza secrets).""" - with patch("sources.st.secrets") as mock_secrets: - mock_secrets.get.side_effect = Exception("no secrets file") - with patch("sources.os.environ", {"GITHUB_TOKEN": "tok-env"}): - assert _github_token() == "tok-env" - - -# ── DuckDB functions ──────────────────────────────────────────────────────── +class TestLoadRadar: + """Contratto: load_radar() produce status_counts + sources da workspace_triage.""" - -class FakeDuckDB: - """Simula duckdb.connect() per test.""" - - class FakeResult: - def df(self): - import pandas as pd - - return pd.DataFrame({"records": [42]}) - - class FakeConnection: - def sql(self, query, params=None): - return FakeDuckDB.FakeResult() - - def close(self): - pass - - def __enter__(self): - return self - - def __exit__(self, *args): - pass - - @staticmethod - def connect(): - return FakeDuckDB.FakeConnection() + def test_transforms_radar_from_triage(self): + with patch("sources._fetch_json", return_value=_MOCK_WORKSPACE_TRIAGE): + result = load_radar() + assert result["status_counts"] == {"GREEN": 34, "YELLOW": 1, "RED": 1} + assert result["persistent_red"] == 1 + assert len(result["sources"]) == 2 # all sources (GREEN + RED) + green = [s for s in result["sources"] if s["status"] == "GREEN"] + red = [s for s in result["sources"] if s["status"] == "RED"] + assert len(green) == 1 + assert len(red) == 1 @pytest.mark.contract -class TestVerifyParquet: - """Contratto: verify_parquet() verifica parquet GCS via DuckDB.""" - - def test_returns_record_count(self): - with patch("sources.duckdb.connect", FakeDuckDB.connect): - result = verify_parquet("test-slug", 2023) - assert result["slug"] == "test-slug" - assert result["year"] == 2023 - assert result["records"] == 42 - - def test_raises_on_error(self): - with patch("sources.duckdb.connect") as mock_con: - mock_con.return_value.__enter__.return_value.sql.side_effect = Exception("DuckDB error") - with pytest.raises(Exception, match="DuckDB error"): - verify_parquet("test-slug", 2023) +class TestLoadSourcesRegistry: + """Contratto: load_sources_registry() legge da SO sources_registry.yaml.""" + + def test_returns_per_source_registry(self): + mock_data = { + "istat_sdmx": { + "protocol": "sdmx", + "observation_mode": "api", + "base_url": "https://...", + }, + "ispra_ru": { + "protocol": "sparql", + "observation_mode": "endpoint", + "base_url": "https://...", + }, + } + with patch("sources._fetch_yaml", return_value=mock_data): + result = load_sources_registry() + assert "istat_sdmx" in result + assert result["istat_sdmx"]["protocol"] == "sdmx" + + def test_returns_empty_on_error(self): + with patch("sources._fetch_yaml", side_effect=Exception("fail")): + result = load_sources_registry() + assert result == {} @pytest.mark.contract -class TestDuckdbQuery: - """Contratto: duckdb_query() esegue SQL e restituisce DataFrame.""" - - def test_executes_sql(self): - fake_df = "fake_df" - with patch("sources.duckdb.connect") as mock_con: - mock_conn = MagicMock() - mock_conn.__enter__.return_value.sql.return_value.df.return_value = fake_df - mock_con.return_value = mock_conn - result = duckdb_query("SELECT 1") - assert result == fake_df - - -# ── Explorer + Analisi ──────────────────────────────────────────────────────── +class TestLoadCatalogSignals: + """Contratto: load_catalog_signals() legge da SO catalog_signals.json.""" + + def test_returns_per_source_signals(self): + mock_data = { + "signals": [ + { + "source_id": "istat_sdmx", + "signal_type": "inventory", + "result": "stabile", + "metric_value": 100, + }, + ] + } + with patch("sources._fetch_json", return_value=mock_data): + result = load_catalog_signals() + assert "signals" in result + assert result["signals"][0]["source_id"] == "istat_sdmx" + + def test_returns_empty_on_error(self): + with patch("sources._fetch_json", side_effect=Exception("fail")): + result = load_catalog_signals() + assert result == {} @pytest.mark.contract -class TestDeSlug: - """Contratto: de_slug() mappa slug DI β†’ slug DE (fallback underscoreβ†’dash).""" - - def test_mapped_slug(self): - assert de_slug("aifa_spesa_consumo") == "spesa-farmaceutica" - assert de_slug("bdap_entrate_stato") == "entrate-stato" - - def test_fallback_replace_underscore(self): - assert de_slug("anac_bandi_gara") == "anac-bandi-gara" - - def test_unknown_keeps_dash_slug(self): - assert de_slug("senato_ddl") == "senato-ddl" - - -_THEMES_REALISTIC = """#!/usr/bin/env python3import json, sys - -themes = [ - {"slug": "territorio-ambiente", - "datasets": ["rifiuti-urbani", "capacita-rinnovabile"]}, - {"slug": "finanza-pubblica", - "datasets": ["irpef-comunale", "entrate-stato"]}, -] +class TestLoadDiscussionCounts: + """Contratto: load_discussion_counts() conta per categoria.""" -json.dump(themes, sys.stdout, ensure_ascii=False) -""" - -_THEMES_SIMPLE = """themes = [ - {"slug": "a", "datasets": ["x", "y"]}, -]""" + def test_counts_by_category(self): + with patch("sources._fetch_json", return_value=_MOCK_WORKSPACE_TRIAGE): + result = load_discussion_counts() + assert result == {"Domande": 1} @pytest.mark.contract class TestLoadExplorerDatasets: - """Contratto: load_explorer_datasets() estrae slug da themes.json.py.""" + """Contratto: load_explorer_datasets() estrae slug da explorer_themes.""" - def test_parses_realistic_file_with_extra_code(self): - """File realistico: ha ``json.dump(...)`` dopo l'array themes. - - Il vecchio parser (partition + literal_eval) falliva su questo caso - perche' literal_eval non accetta codice extra dopo il literal. - """ - with patch("sources._HTTP.get") as mock_get: - mock_get.return_value = _py_resp(_THEMES_REALISTIC) + def test_extracts_slugs(self): + with patch("sources._fetch_json", return_value=_MOCK_TOPIC_INDEX): result = load_explorer_datasets() - assert result == { - "rifiuti-urbani", - "capacita-rinnovabile", - "irpef-comunale", - "entrate-stato", - } + assert result == {"ade_cinque_per_mille"} - def test_parses_simple_file(self): - """File minimale: solo l'assegnamento themes.""" - with patch("sources._HTTP.get") as mock_get: - mock_get.return_value = _py_resp(_THEMES_SIMPLE) - result = load_explorer_datasets() - assert result == {"x", "y"} - - def test_returns_empty_on_http_error(self): - with patch("sources._HTTP.get") as mock_get: - mock_get.return_value = _py_resp("", status=500) + def test_returns_empty_on_error(self): + with patch("sources._fetch_json", side_effect=Exception("fail")): result = load_explorer_datasets() assert result == set() -_ANALISI_README = """--- -title: Test -dataset_slug: test_dataset ---- -# Test analysis""" +@pytest.mark.contract +class TestDeSlug: + def test_known_mapping(self): + assert de_slug("aifa_spesa_consumo") == "spesa-farmaceutica" + + def test_default_mapping(self): + assert de_slug("ispra_ru_base") == "rifiuti-urbani" + def test_generic_conversion(self): + assert de_slug("some_dataset") == "some-dataset" -@pytest.mark.contract -class TestLoadAnalysisRegistry: - """Contratto: load_analysis_registry() mappa analisi β†’ dataset_slug.""" - - def test_parses_readme_frontmatter(self): - gh_api_response = [ - {"type": "dir", "name": "test-analisi"}, - {"type": "dir", "name": "registry"}, - {"type": "file", "name": "README.md"}, - ] - with patch("sources._HTTP.get") as mock_get: - mock_get.side_effect = [ - _resp(gh_api_response), # API directory listing - _py_resp(_ANALISI_README), # README.md - ] - result = load_analysis_registry() - assert result == {"test-analisi": "test_dataset"} - - def test_skips_registry_and_template(self): - gh_api_response = [ - {"type": "dir", "name": "registry"}, - {"type": "dir", "name": "_template"}, - {"type": "dir", "name": "irpef-comunale"}, - ] - with patch("sources._HTTP.get") as mock_get: - mock_get.side_effect = [ - _resp(gh_api_response), # API listing - _py_resp("---\ndataset_slug: irpef_comunale\n---"), # README - ] - result = load_analysis_registry() - assert "registry" not in result - assert "_template" not in result - assert result.get("irpef-comunale") == "irpef_comunale" - - def test_returns_empty_on_http_error(self): - with patch("sources._HTTP.get") as mock_get: - mock_get.return_value = _resp([], status=500) - result = load_analysis_registry() - assert result == {} + def test_no_underscore(self): + assert de_slug("nodash") == "nodash"