更新: 109 个文件 - 2026-03-18 10:55:52

scripts/lab/render.py

@@ -236,6 +236,79 @@ def _latest_advisories(advisories: List[Dict[str, Any]]) -> List[Dict[str, Any]]
     return [annotate_with_latest_run(item, run_map.get(item.get("canonical_id"))) for item in advisories]
 
 
+def _latest_run_map(runs: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
+    latest: Dict[str, Dict[str, Any]] = {}
+    for item in runs:
+        advisory_id = item.get("advisory_id")
+        if not advisory_id:
+            continue
+        previous = latest.get(advisory_id)
+        if previous is None or (item.get("finished_at") or "") >= (previous.get("finished_at") or ""):
+            latest[advisory_id] = item
+    return latest
+
+
+def _synthetic_advisory_from_run(run: Dict[str, Any], source_system_map: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
+    system_id = run.get("system_id") or "unknown"
+    system_meta = source_system_map.get(system_id, {})
+    browser_evidence = run.get("browser_evidence") or {
+        "required": False,
+        "present": bool(run.get("browser_refs")),
+        "refs": run.get("browser_refs", []),
+    }
+    return {
+        "canonical_id": run.get("advisory_id"),
+        "system_id": system_id,
+        "display_name": system_meta.get("display_name", system_id),
+        "title": run.get("advisory_title") or run.get("advisory_id") or run.get("run_id"),
+        "summary": run.get("blocked_reason") or f"Derived from latest run {run.get('run_id')}",
+        "category": system_meta.get("category"),
+        "aliases": [],
+        "secure_code_topics": system_meta.get("secure_code_topics", []),
+        "verification_status": run.get("verification_status"),
+        "verification_mode": run.get("verification_mode"),
+        "last_verified_at": run.get("finished_at"),
+        "last_run_id": run.get("run_id"),
+        "browser_evidence": browser_evidence,
+        "repro_profile_id": run.get("repro_profile_id"),
+        "artifact_mode": run.get("artifact_mode"),
+        "blocked_reason": run.get("blocked_reason"),
+        "published_at": run.get("started_at") or run.get("finished_at"),
+        "updated_at": run.get("finished_at") or run.get("started_at"),
+        "official_source_url": "",
+        "secondary_source_urls": [],
+    }
+
+
+def _merge_latest_advisories(
+    advisories: List[Dict[str, Any]],
+    runs: List[Dict[str, Any]],
+    source_system_map: Dict[str, Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    run_map = _latest_run_map(runs)
+    merged: Dict[str, Dict[str, Any]] = {}
+
+    for item in advisories:
+        canonical_id = item.get("canonical_id")
+        if not canonical_id:
+            continue
+        merged[canonical_id] = annotate_with_latest_run(item, run_map.get(canonical_id))
+
+    for advisory_id, run in run_map.items():
+        if advisory_id in merged:
+            continue
+        merged[advisory_id] = annotate_with_latest_run(_synthetic_advisory_from_run(run, source_system_map), run)
+
+    return sorted(
+        merged.values(),
+        key=lambda item: (
+            item.get("updated_at") or item.get("published_at") or "",
+            item.get("canonical_id") or "",
+        ),
+        reverse=True,
+    )
+
+
 def _build_completeness(
     advisories: List[Dict[str, Any]],
     runs: List[Dict[str, Any]],
@@ -1202,7 +1275,7 @@ def render_dashboard() -> Dict[str, str]:
     source_map = read_yaml(SOURCE_MAP_PATH, default={}) or {}
     repro_map = read_yaml(REPRO_MAP_PATH, default={}) or {}
     source_system_map = {item["system_id"]: item for item in source_map.get("systems", []) if item.get("system_id")}
-    merged_advisories = _latest_advisories(advisory_records)
+    merged_advisories = _merge_latest_advisories(advisory_records, runs, source_system_map)
     advisory_map = {item["canonical_id"]: item for item in merged_advisories if item.get("canonical_id")}
     profile_map = load_profiles()