ci: automate freshdata on every pull request

.github/workflows/automation.yml

@@ -0,0 +1,87 @@
+# Batch profile + clean every fixture on each PR. Fails if any dataset errors.
+name: Freshdata automation
+
+on:
+  pull_request:
+  workflow_dispatch:
+
+concurrency:
+  group: automation-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  automate:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[dev]"
+
+      - name: Run fixture automation
+        run: |
+          python scripts/automate_freshdata.py \
+            --output-dir scripts/.automation_out \
+            --report scripts/.automation_out/summary.json
+
+      - name: Job summary
+        if: always()
+        run: |
+          python - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+
+          summary_path = Path("scripts/.automation_out/summary.json")
+          if not summary_path.exists():
+              print("No automation summary found.")
+              raise SystemExit(0)
+
+          data = json.loads(summary_path.read_text(encoding="utf-8"))
+          lines = [
+              "## Freshdata automation",
+              "",
+              f"- **Datasets:** {data.get('total', 0)}",
+              f"- **OK:** {data.get('ok', 0)}",
+              f"- **Failed:** {data.get('fail', 0)}",
+              f"- **Steps:** `{', '.join(data.get('steps', []))}`",
+              "",
+          ]
+
+          out_dir = Path(data.get("output_dir", "scripts/.automation_out"))
+          for report_name in data.get("reports", []):
+              report = json.loads((out_dir / report_name).read_text(encoding="utf-8"))
+              ds = report["dataset"]
+              before = report.get("before_shape")
+              after = report.get("after_shape")
+              err = report.get("error")
+              if err:
+                  lines.append(f"- ❌ `{ds}` — {err}")
+              else:
+                  lines.append(f"- ✅ `{ds}` {before} → {after} ({report.get('seconds', '?')}s)")
+
+          body = "\n".join(lines)
+          print(body)
+          summary_file = os.environ.get("GITHUB_STEP_SUMMARY")
+          if summary_file:
+              Path(summary_file).write_text(body + "\n", encoding="utf-8")
+          PY
+
+      - name: Upload automation artifacts
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: freshdata-automation
+          path: scripts/.automation_out/
+          if-no-files-found: warn

.gitignore

@@ -15,8 +15,8 @@ htmlcov/
 .DS_Store
 .ipynb_checkpoints/
 
-# AI-assistant working artifacts (plans, task reports) — never committed
-.superpowers/
+# Automation runner output (CI uploads as artifact)
+scripts/.automation_out/
 
 # MkDocs build output
 site/

Makefile

@@ -5,7 +5,7 @@ PY ?= python
 # training-* targets are matched by the pattern rule below (pattern rules
 # cannot be .PHONY; the delegated targets are .PHONY inside training/Makefile).
 .PHONY: help benchmark benchmark-ci benchmark-report benchmark-fixtures benchmark-test \
-        cleanbench-full truthbench-release truthbench-pr performance-ci \
+        cleanbench-full truthbench-release truthbench-pr automation-pr performance-ci \
         performance-baseline performance-profile performance-report coverage-report
 
 help:
@@ -18,6 +18,7 @@ help:
 	@echo "  cleanbench-full     Full CleanBench T1-T5 with release gates + site report"
 	@echo "  truthbench-release  Official TruthBench release verification (fail-closed)"
 	@echo "  truthbench-pr       TruthBench PR ratchet (fails only on regressions)"
+	@echo "  automation-pr       Profile + clean all fixtures (PR automation gate)"
 	@echo "  performance-ci      Run the CI-safe performance contract suite"
 	@echo "  performance-baseline Run the performance investigation matrix"
 	@echo "  performance-profile Profile one 100k-row performance case"
@@ -56,6 +57,10 @@ truthbench-pr:
 	  --repeats 2 \
 	  --check-regressions
 
+# Profile + clean every committed fixture; same gate as .github/workflows/automation.yml.
+automation-pr:
+	$(PY) scripts/automate_freshdata.py --output-dir scripts/.automation_out
+
 # Phase-5 training pipeline targets delegate to training/Makefile.
 training-%:
 	$(MAKE) -C training PY=$(PY) $@

scripts/automate_freshdata.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Batch profile + clean runner for local fixtures and CSV inputs."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+from pathlib import Path
+
+import pandas as pd
+
+import freshdata as fd
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "tests"))
+
+from expectations import ALL_FIXTURES, load_fixture  # noqa: E402
+
+
+def parse_steps(raw: str) -> list[str]:
+    return [part.strip() for part in raw.split(",") if part.strip()]
+
+
+def run_steps(df: pd.DataFrame, steps: list[str]) -> tuple[pd.DataFrame, dict]:
+    """Run freshdata steps; only mutating steps update the working frame."""
+    out = df.copy()
+    meta: dict = {}
+    for name in steps:
+        fn = getattr(fd, name, None)
+        if not callable(fn):
+            raise ValueError(f"unknown step: {name}")
+        t0 = time.perf_counter()
+        if name == "profile":
+            prof = fn(out)
+            meta["profile"] = {
+                "columns": len(getattr(prof, "columns", prof)),
+                "rows": len(out),
+            }
+        elif name == "clean":
+            result = fn(out, return_report=True)
+            out = result[0] if isinstance(result, tuple) else result
+            if isinstance(result, tuple) and len(result) > 1:
+                report = result[1]
+                meta["clean"] = {
+                    "actions": len(getattr(report, "actions", report)),
+                }
+        else:
+            result = fn(out)
+            out = result[0] if isinstance(result, tuple) else result
+        meta.setdefault("timing", {})[name] = round(time.perf_counter() - t0, 4)
+    return out, meta
+
+
+def collect_csvs(inp: str | None) -> list[tuple[str, pd.DataFrame]]:
+    if inp:
+        path = Path(inp)
+        if path.is_file() and path.suffix.lower() == ".csv":
+            return [(path.stem, pd.read_csv(path))]
+        if path.is_dir():
+            return [(item.stem, pd.read_csv(item)) for item in sorted(path.glob("*.csv"))]
+        return []
+    return [(str(name), load_fixture(name)) for name in ALL_FIXTURES]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Reusable FreshData automation runner")
+    parser.add_argument("--input", help="CSV file path or directory of CSV files")
+    parser.add_argument(
+        "--output-dir",
+        default=str(ROOT / "scripts" / ".automation_out"),
+    )
+    parser.add_argument("--steps", default="profile,clean", help="Comma-separated steps")
+    parser.add_argument("--report", help="Optional summary report path")
+    args = parser.parse_args()
+
+    out_dir = Path(args.output_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    steps = parse_steps(args.steps)
+    datasets = collect_csvs(args.input)
+    reports: list[dict] = []
+    ok = fail = 0
+
+    for name, df in datasets:
+        t0 = time.perf_counter()
+        rec = {
+            "dataset": name,
+            "steps": steps,
+            "before_shape": list(df.shape),
+            "after_shape": None,
+            "seconds": None,
+            "error": None,
+        }
+        try:
+            cleaned, step_meta = run_steps(df, steps)
+            rec["after_shape"] = list(cleaned.shape)
+            rec["step_meta"] = step_meta
+            cleaned.to_csv(out_dir / f"{Path(name).stem}_cleaned.csv", index=False)
+            ok += 1
+        except Exception as exc:  # noqa: BLE001
+            rec["error"] = f"{type(exc).__name__}: {exc}"
+            fail += 1
+        rec["seconds"] = round(time.perf_counter() - t0, 6)
+        report_path = out_dir / f"{Path(name).stem}_report.json"
+        report_path.write_text(json.dumps(rec, indent=2), encoding="utf-8")
+        reports.append(rec)
+
+    report_names = [f"{Path(row['dataset']).stem}_report.json" for row in reports]
+    summary = {
+        "total": len(reports),
+        "ok": ok,
+        "fail": fail,
+        "steps": steps,
+        "output_dir": str(out_dir),
+        "reports": report_names,
+    }
+    (out_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
+    if args.report:
+        report_out = Path(args.report)
+        report_out.parent.mkdir(parents=True, exist_ok=True)
+        report_out.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+    return 0 if fail == 0 else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())