feat: pipeline merge+validate — nuovo funnel SO (−6.400 righe nette)

.github/workflows/observatory.yml

@@ -9,7 +9,6 @@ on:
 permissions:
   contents: write
   id-token: write
-  issues: write
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
@@ -48,13 +47,6 @@ jobs:
       # ═══════════════════════════════════════════════════════════════════
       # STEP 1 — INVENTORY BUILD
       # ═══════════════════════════════════════════════════════════════════
-      - name: Download previous report (for signals delta)
-        if: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER != '' && env.GCP_SERVICE_ACCOUNT != '' && env.CATALOG_INVENTORY_GCS_PREFIX != '' }}
-        run: |
-          prefix='${{ env.CATALOG_INVENTORY_GCS_PREFIX }}'
-          prefix="${prefix%/}"
-          gcloud storage cp "$prefix/catalog_inventory_report.json" previous_report.json || echo "{}" > previous_report.json
-
       - name: Download previous inventory (for source-check merge)
         if: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER != '' && env.GCP_SERVICE_ACCOUNT != '' && env.CATALOG_INVENTORY_GCS_PREFIX != '' }}
         run: |
@@ -71,97 +63,36 @@ jobs:
             --workers 16 \
             --skip-red-sources
 
-      - name: Build catalog signals + diff
-        run: |
-          python scripts/build_catalog_signals.py \
-            --report data/catalog_inventory/generated/catalog_inventory_report.json \
-            --previous previous_report.json \
-            --radar data/radar/radar_summary.json \
-            --out data/catalog/catalog_signals.json
-
-          if python -c "import json,sys; d=json.load(open('previous_report.json')); sys.exit(0 if d.get('sources') else 1)"; then
-            python scripts/catalog_diff.py previous_report.json \
-              data/catalog_inventory/generated/catalog_inventory_report.json --output diff.md
-          else
-            echo "NO_BASELINE" > diff.md
-          fi
-
-      - name: Commit catalog signals
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: |
-          git config user.name "github-actions[bot]"
-          git config user.email "github-actions[bot]@users.noreply.github.com"
-          git add data/catalog/catalog_signals.json
-          if git diff --cached --quiet; then
-            echo "Nessuna variazione nei segnali."
-          else
-            git commit -m "chore(so): aggiorna catalog_signals e watch report [ci skip]"
-            git push
-          fi
-
-      - name: Create catalog alert issue
-        env:
-          GH_TOKEN: ${{ github.token }}
-          GCS_PREFIX: ${{ secrets.CATALOG_INVENTORY_GCS_PREFIX }}
-        run: |
-          python scripts/gha/build_issue_body.py --gcs-prefix "$GCS_PREFIX"
-          if [ ! -f issue_title.txt ]; then
-            echo "Nessuna variazione o baseline assente."
-            exit 0
-          fi
-          LABEL_ARGS=$(python3 -c "import json,sys; print(' '.join('--label '+l for l in json.load(open('issue_labels.json'))))")
-          EXISTING=$(gh issue list --label "catalog-alert" --state open --json number,title --jq \
-            ".[] | select(.title | startswith(\"[Catalog]\")) | .number" | head -1)
-          if [ -n "$EXISTING" ]; then
-            echo "Aggiorno issue #$EXISTING."
-            gh issue comment "$EXISTING" --body-file issue_body.md
-          else
-            eval "gh issue create --title \"$(cat issue_title.txt)\" --body-file issue_body.md $LABEL_ARGS"
-          fi
-
       # ═══════════════════════════════════════════════════════════════════
-      # STEP 3 — SOURCE-CHECK (sempre)
+      # STEP 2 — PIPELINE (merge + validate)
       # ═══════════════════════════════════════════════════════════════════
-      - name: Download source-check results from GCS (for merge)
-        if: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER != '' && env.GCP_SERVICE_ACCOUNT != '' && env.CATALOG_INVENTORY_GCS_PREFIX != '' }}
+      - name: Clean old artifacts
         run: |
-          prefix='${{ env.CATALOG_INVENTORY_GCS_PREFIX }}'
-          prefix="${prefix%/}"
-          mkdir -p data/catalog_inventory/generated
-          gcloud storage cp "$prefix/source-check/source_check_results.parquet" \
-            data/catalog_inventory/generated/source_check_results.parquet 2>/dev/null || \
-            echo "No previous results — starting fresh"
+          rm -f data/catalog/CATALOG_WATCH_REPORT.md
 
-      - name: Run bulk source-check
+      - name: Run pipeline (merge + validate)
         run: |
-          python scripts/bulk_source_check.py \
-            --skip-red-sources \
-            --no-sdmx-years \
-            --circuit-fail-threshold 2 \
-            --max-items 5000 \
+          python scripts/pipeline/run_pipeline.py \
             --workers 16
 
       # ═══════════════════════════════════════════════════════════════════
-      # STEP 3b — BUILD SOURCE REPORTS
+      # STEP 3 — SOURCE REPORTS
       # ═══════════════════════════════════════════════════════════════════
       - name: Build source reports
         run: |
           python scripts/build_source_reports.py
 
-      - name: Commit source reports + parquet
+      - name: Commit results
         env:
           GH_TOKEN: ${{ github.token }}
         run: |
           git config user.name "github-actions[bot]"
           git config user.email "github-actions[bot]@users.noreply.github.com"
-          git add data/reports/
-          # Parquet: force-add perché data/catalog_inventory/generated/ è in .gitignore
+          git add data/reports/ data/pipeline/validated.parquet data/pipeline/summary.json
           git add -f data/catalog_inventory/generated/catalog_inventory_latest.parquet \
-                     data/catalog_inventory/generated/source_check_results.parquet \
                      data/catalog_inventory/generated/catalog_inventory_report.json
           if git diff --cached --quiet; then
-            echo "Nessuna variazione nei report."
+            echo "Nessuna variazione."
           else
             git commit -m "chore(so): aggiorna reports e parquet [ci skip]"
             git push
@@ -170,18 +101,21 @@ jobs:
       # ═══════════════════════════════════════════════════════════════════
       # STEP 4 — UPLOAD GCS
       # ═══════════════════════════════════════════════════════════════════
-      - name: Upload source-check results to GCS
+      - name: Upload validated results to GCS
         if: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER != '' && env.GCP_SERVICE_ACCOUNT != '' && env.CATALOG_INVENTORY_GCS_PREFIX != '' }}
         run: |
           stamp=$(python3 -c "from datetime import datetime, timezone; print(datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S'))")
           prefix='${{ env.CATALOG_INVENTORY_GCS_PREFIX }}'
           prefix="${prefix%/}"
           python scripts/gha/gcs_upload.py \
-            data/catalog_inventory/generated/source_check_results.parquet \
-            "$prefix/source-check/source_check_results.parquet"
+            data/pipeline/validated.parquet \
+            "$prefix/pipeline/validated.parquet"
+          python scripts/gha/gcs_upload.py \
+            data/pipeline/validated.parquet \
+            "$prefix/pipeline/snapshots/validated_${stamp}.parquet"
           python scripts/gha/gcs_upload.py \
-            data/catalog_inventory/generated/source_check_results.parquet \
-            "$prefix/source-check/snapshots/source_check_${stamp}.parquet"
+            data/pipeline/summary.json \
+            "$prefix/pipeline/summary_${stamp}.json"
 
       - name: Upload inventory snapshot to GCS
         if: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER != '' && env.GCP_SERVICE_ACCOUNT != '' && env.CATALOG_INVENTORY_GCS_PREFIX != '' }}
@@ -207,20 +141,19 @@ jobs:
             "$prefix/snapshots/catalog_inventory_report_${stamp}.json"
 
       # ═══════════════════════════════════════════════════════════════════
-      # STEP 5 — ARTIFACTS + SUMMARIES
+      # STEP 5 — ARTIFACTS
       # ═══════════════════════════════════════════════════════════════════
       - name: Upload workflow artifacts
         uses: actions/upload-artifact@v7
         with:
           name: observatory-results
           path: |
-            data/catalog_inventory/generated/source_check_results.parquet
+            data/pipeline/validated.parquet
+            data/pipeline/summary.json
+            data/catalog_inventory/generated/catalog_inventory_latest.parquet
             data/radar/radar_summary.json
-            data/radar/radar_history.json
 
-      - name: Publish summaries
+      - name: Publish radar summary
         run: |
           python scripts/gha/publish_radar_summary.py
           cat radar_summary.md >> "$GITHUB_STEP_SUMMARY"
-          python scripts/gha/publish_source_check_summary.py
-          cat source_check_summary.md >> "$GITHUB_STEP_SUMMARY"

.gitignore

@@ -35,3 +35,4 @@ observatory-results/
 # Generated by CI — bridge verso data-advocacy
 data/health/
 data/compliance/
+data/pipeline/

CONTRIBUTING.md

@@ -13,17 +13,18 @@ Risponde a una domanda: **questa fonte vale il tempo del Lab?**
 Il funnel del repo:
 
 ```
-radar ── gate ── catalog-watch ── catalog-inventory ── source-check
-             └── radar-only
+radar ── gate ── catalog-watch ── catalog-inventory ── pipeline (merge → validate)
+              └── radar-only
 ```
 
 Qui stanno:
 
 - `sources_registry.yaml` — registro di tutte le fonti osservate
-- `scripts/` — radar check, inventory, source-check, catalog diff
+- `scripts/` — radar check, inventory, pipeline, report
+- `scripts/pipeline/` — merge + validate (produce `validated.parquet`)
 - `skills/` — guide operative per agenti (source-check, inventory-triage, portal-scout)
 - `so_mcp/` — layer MCP read-only sugli artifact
-- `data/` — artifact versionati: radar_summary, radar_history, catalog_signals
+- `data/` — artifact versionati: radar_summary, radar_history
 - workflow CI: `radar.yml` (daily), `observatory.yml` (weekly)
 
 Qui non stanno:
@@ -42,7 +43,7 @@ il trattamento:
 | Modalità | Cosa succede | Frequenza |
 |---|---|---|
 | `radar-only` | Solo health check HTTP | Daily (radar.yml) |
-| `catalog-watch` | Radar + inventory + source-check | Daily radar + weekly observatory |
+| `catalog-watch` | Radar + inventory + pipeline (merge→validate) | Daily radar + weekly observatory |
 
 ### Radar (daily)
 
@@ -54,9 +55,9 @@ Probe HTTP leggero su ogni fonte. Produce:
 ### Observatory (weekly, lunedì)
 
 1. Build inventory parquet per fonti `catalog-watch`
-2. Calcola segnali di drift
-3. Scoring item-level (source-check)
-4. Upload su GCS + issue alert in caso di variazioni
+2. Pipeline merge + validate → `validated.parquet`
+3. Report per fonte + dashboard
+4. Upload su GCS
 
 ## Setup locale
 
@@ -75,13 +76,16 @@ pip install -e ../lab-connectors
 
 ```bash
 # Radar check manuale
-python scripts/radar_check.py
+so-radar-check
 
 # Catalog inventory
 python scripts/build_catalog_inventory.py --out-dir data/catalog_inventory/generated --workers 4
 
-# Source-check incrementale
-python scripts/bulk_source_check.py --skip-red-sources --max-items 200 --workers 8
+# Pipeline merge + validate
+so-run-pipeline --workers 4
+
+# Build reports
+so-build-reports
 
 # Test
 pytest tests/
@@ -121,9 +125,9 @@ Vedi [`.github`](https://github.com/dataciviclab/.github) per orientarti.
 ## Riferimenti
 
 - [README.md](README.md) — panoramica del repo
-- [docs/runbook.md](docs/runbook.md) — guida operativa radar, inventory, source-check
+- [docs/runbook.md](docs/runbook.md) — guida operativa radar, inventory, pipeline
 - [docs/architecture.md](docs/architecture.md) — architettura del sistema
 - [docs/catalog_watch_measurement_policy.md](docs/catalog_watch_measurement_policy.md) — policy di misura
 - [skills/](skills/) — guide operative per agenti
 - [`lab-connectors`](https://github.com/dataciviclab/lab-connectors) — dipendenza condivisa
-- [`dataset-incubator`](https://github.com/dataciviclab/dataset-incubator) — downstream: qui finiscono i source-check che diventano candidate
+- [`dataset-incubator`](https://github.com/dataciviclab/dataset-incubator) — downstream: qui finiscono i validated che diventano candidate