"""Probe diff — compare current vs previous extraction results.
Detects: added, removed, and changed records between two probe runs.
Useful for change detection and drift monitoring.
"""
from __future__ import annotations
import json
from pathlib import Path
[docs]
def diff_probes(current_dir: Path, previous_dir: Path) -> list[dict]:
"""Compare all probe JSON files between two output directories.
Returns a list of diff records: {probe, added, removed, changed, unchanged}
"""
results = []
current_files = {f.relative_to(current_dir): f for f in current_dir.rglob("*.json")}
previous_files = {f.relative_to(previous_dir): f for f in previous_dir.rglob("*.json")}
all_keys = sorted(set(current_files.keys()) | set(previous_files.keys()))
for key in all_keys:
cur_file = current_files.get(key)
prev_file = previous_files.get(key)
if cur_file and not prev_file:
cur_data = _load(cur_file)
results.append({
"probe": str(key),
"status": "new",
"current_records": cur_data.get("record_count", 0),
"previous_records": 0,
"added": cur_data.get("record_count", 0),
"removed": 0,
"changed": 0,
})
elif prev_file and not cur_file:
prev_data = _load(prev_file)
results.append({
"probe": str(key),
"status": "removed",
"current_records": 0,
"previous_records": prev_data.get("record_count", 0),
"added": 0,
"removed": prev_data.get("record_count", 0),
"changed": 0,
})
elif cur_file and prev_file:
cur_data = _load(cur_file)
prev_data = _load(prev_file)
diff = _diff_records(
cur_data.get("records", []),
prev_data.get("records", []),
)
results.append({
"probe": str(key),
"status": "changed" if diff["added"] or diff["removed"] or diff["changed"] else "unchanged",
"current_records": cur_data.get("record_count", 0),
"previous_records": prev_data.get("record_count", 0),
**diff,
})
return results
def _load(path: Path) -> dict:
try:
return json.loads(path.read_text())
except Exception:
return {}
def _diff_records(current: list[dict], previous: list[dict]) -> dict:
"""Smart diff: detects added, removed, and changed records by primary key."""
def _key(record: dict) -> str:
for k in ("table_name", "name", "view_name", "column_name", "index_name",
"constraint_name", "probe", "id", "key", "path"):
if k in record:
return f"{k}={record[k]}"
return json.dumps(record, sort_keys=True, default=str)
cur_by_key = {_key(r): r for r in current}
prev_by_key = {_key(r): r for r in previous}
added = len(set(cur_by_key) - set(prev_by_key))
removed = len(set(prev_by_key) - set(cur_by_key))
common = set(cur_by_key) & set(prev_by_key)
changed = sum(
1 for k in common
if json.dumps(cur_by_key[k], sort_keys=True, default=str)
!= json.dumps(prev_by_key[k], sort_keys=True, default=str)
)
unchanged = len(common) - changed
return {"added": added, "removed": removed, "changed": changed, "unchanged": unchanged}