(
snapshots: List[Path],
kev_full: List[Dict] | None = None,
*,
threshold: float,
max_movers: int,
recent_days: int,
)
| 61 | |
| 62 | |
| 63 | def build_diff( |
| 64 | snapshots: List[Path], |
| 65 | kev_full: List[Dict] | None = None, |
| 66 | *, |
| 67 | threshold: float, |
| 68 | max_movers: int, |
| 69 | recent_days: int, |
| 70 | ) -> Tuple[Dict, Path | None]: |
| 71 | if not snapshots: |
| 72 | return {}, None |
| 73 | latest_path = snapshots[-1] |
| 74 | latest = load_snapshot(latest_path) |
| 75 | latest_date = latest.get("generated") or latest_path.stem |
| 76 | |
| 77 | if len(snapshots) >= 2: |
| 78 | prev = load_snapshot(snapshots[-2]) |
| 79 | kev_diff = diff_lists(prev.get("kev_top", []), latest.get("kev_top", [])) |
| 80 | high_epss_diff = diff_lists(prev.get("high_epss", []), latest.get("high_epss", [])) |
| 81 | else: |
| 82 | prev = {} |
| 83 | kev_diff = {"new": latest.get("kev_top", []), "removed": []} |
| 84 | high_epss_diff = {"new": latest.get("high_epss", []), "removed": []} |
| 85 | |
| 86 | prev_epss_lookup = {row["cve"]: row for row in (prev.get("high_epss", []) if prev else [])} |
| 87 | curr_epss_lookup = {row["cve"]: row for row in latest.get("high_epss", [])} |
| 88 | epss_movers = compute_epss_movers(prev_epss_lookup, curr_epss_lookup, max_movers) |
| 89 | |
| 90 | kev_recent = filter_recent_kev(kev_full or latest.get("kev_top", []), recent_days=recent_days) |
| 91 | |
| 92 | diff_outputs = { |
| 93 | "generated": latest_date, |
| 94 | "new_kev_entries": kev_recent, |
| 95 | "removed_kev_entries": kev_diff["removed"], |
| 96 | "new_high_epss": [row for row in high_epss_diff["new"] if (row.get("epss") or 0) >= threshold], |
| 97 | "removed_high_epss": high_epss_diff["removed"], |
| 98 | "epss_movers": epss_movers, |
| 99 | } |
| 100 | |
| 101 | target = API_DIR / "diff" / f"{latest_date}.json" |
| 102 | ensure_dirs(target.parent) |
| 103 | save_json(target, diff_outputs) |
| 104 | # also write a stable latest pointer |
| 105 | save_json(target.parent / "latest.json", diff_outputs) |
| 106 | |
| 107 | return diff_outputs, target |
| 108 | |
| 109 | |
| 110 | def prune_snapshots(snapshots: List[Path], *, lookback_days: int) -> None: |
no test coverage detected