feat: adapter Explorer->generatore (scripts/adapt_explorer.py) + test; README avvio rapido con adattamento

Author: superpios

README.md

@@ -22,7 +22,9 @@ Nessuna pista dimostra, suggerisce o implica illecito, spreco, frode o responsab
 ## Avvio rapido
 ```bash
 pip install -r requirements.txt
-# Copia le tabelle relations/ dall'Explorer in data/input/
+# 1) Adatta le tabelle di relazione dell'Explorer nel formato atteso dal generatore
+python scripts/adapt_explorer.py --relations <EXPLORE>/data/relations --output data/input
+# 2) Applica le regole (deterministico, fail-closed)
 python scripts/apply_rules.py --input data/input --output data/leads --rules rules/rules_v0.1.yaml
 ```
 

scripts/adapt_explorer.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Adattatore: trasforma le tabelle di relazione di investigative-explorer-dvns
+nel formato di input atteso da scripts/apply_rules.py.
+
+Mappatura documentata e revisionabile (vedi REGOLE_SEGNALAZIONE.md):
+  persona_incarico_ente__incarichi_nominativi_shard.csv
+     -> person_name = subject_key, entity_id = object_key, year = period[:4]
+  awards__affidamenti_diretti.csv
+     -> awardee = subject_key, entity_id = object_key, award_date = period,
+        procedure_type = "affidamento diretto" (il dataset e' gia' filtrato)
+  cig_ente__affidamenti_diretti.csv
+     -> cig = subject_key, subject_id = object_key
+
+La provenienza (source_dataset, source_record_id) e' preservata cosi' com'e'.
+"""
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import pandas as pd
+
+SOURCES = {
+    "persona_incarico_ente__incarichi_nominativi_shard.csv": "incarichi.csv",
+    "awards__affidamenti_diretti.csv": "affidamenti_diretti.csv",
+    "cig_ente__affidamenti_diretti.csv": "cig_enti.csv",
+}
+
+
+def _src(df: pd.DataFrame) -> dict[str, object]:
+    return {
+        "source_dataset": df.get("source_dataset", ""),
+        "source_record_id": df.get("source_record_id", ""),
+    }
+
+
+def adapt_persona(df: pd.DataFrame) -> pd.DataFrame:
+    out = pd.DataFrame()
+    out["person_name"] = df["subject_key"]
+    out["entity_id"] = df["object_key"]
+    out["year"] = df["period"].astype(str).str[:4]
+    out["source_dataset"] = _src(df)["source_dataset"]
+    out["source_record_id"] = _src(df)["source_record_id"]
+    return out
+
+
+def adapt_awards(df: pd.DataFrame) -> pd.DataFrame:
+    out = pd.DataFrame()
+    out["awardee"] = df["subject_key"]
+    out["entity_id"] = df["object_key"]
+    out["award_date"] = df["period"].astype(str)
+    out["procedure_type"] = "affidamento diretto"
+    out["source_dataset"] = _src(df)["source_dataset"]
+    out["source_record_id"] = _src(df)["source_record_id"]
+    return out
+
+
+def adapt_cig(df: pd.DataFrame) -> pd.DataFrame:
+    out = pd.DataFrame()
+    out["cig"] = df["subject_key"]
+    out["subject_id"] = df["object_key"]
+    out["source_dataset"] = _src(df)["source_dataset"]
+    out["source_record_id"] = _src(df)["source_record_id"]
+    return out
+
+
+ADAPTERS = {
+    "persona_incarico_ente__incarichi_nominativi_shard.csv": adapt_persona,
+    "awards__affidamenti_diretti.csv": adapt_awards,
+    "cig_ente__affidamenti_diretti.csv": adapt_cig,
+}
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="Adatta le relazioni dell'Explorer al formato del generatore.")
+    ap.add_argument("--relations", required=True, type=Path, help="Cartella data/relations dell'Explorer")
+    ap.add_argument("--output", required=True, type=Path, help="Cartella data/input del generatore")
+    args = ap.parse_args()
+    args.output.mkdir(parents=True, exist_ok=True)
+    for src, dst in SOURCES.items():
+        p = args.relations / src
+        if not p.exists():
+            print(f"salto (assente): {src}")
+            continue
+        df = pd.read_csv(p, dtype=str, keep_default_na=False)
+        out = ADAPTERS[src](df)
+        out.to_csv(args.output / dst, index=False)
+        print(f"{src} -> {dst}: {len(out)} righe")
+
+
+if __name__ == "__main__":
+    main()

tests/test_adapter.py

@@ -0,0 +1,68 @@
+import sys
+from pathlib import Path
+
+import pandas as pd
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+
+import adapt_explorer  # noqa: E402
+
+
+def test_adapt_persona():
+    df = pd.DataFrame(
+        [
+            {
+                "subject_key": "MARIO ROSSI",
+                "object_key": "COMUNE X",
+                "period": "2025-03-01",
+                "source_dataset": "ds",
+                "source_record_id": "r1",
+            }
+        ]
+    )
+    out = adapt_explorer.adapt_persona(df)
+    assert list(out.columns) == [
+        "person_name",
+        "entity_id",
+        "year",
+        "source_dataset",
+        "source_record_id",
+    ]
+    assert out.iloc[0]["person_name"] == "MARIO ROSSI"
+    assert out.iloc[0]["entity_id"] == "COMUNE X"
+    assert out.iloc[0]["year"] == "2025"
+
+
+def test_adapt_awards():
+    df = pd.DataFrame(
+        [
+            {
+                "subject_key": "AZ",
+                "object_key": "ENT",
+                "period": "2024-05-01",
+                "source_dataset": "ds",
+                "source_record_id": "r2",
+            }
+        ]
+    )
+    out = adapt_explorer.adapt_awards(df)
+    assert out.iloc[0]["awardee"] == "AZ"
+    assert out.iloc[0]["entity_id"] == "ENT"
+    assert out.iloc[0]["award_date"] == "2024-05-01"
+    assert out.iloc[0]["procedure_type"] == "affidamento diretto"
+
+
+def test_adapt_cig():
+    df = pd.DataFrame(
+        [
+            {
+                "subject_key": "CIG1",
+                "object_key": "ENT",
+                "source_dataset": "ds",
+                "source_record_id": "r3",
+            }
+        ]
+    )
+    out = adapt_explorer.adapt_cig(df)
+    assert out.iloc[0]["cig"] == "CIG1"
+    assert out.iloc[0]["subject_id"] == "ENT"