Ransomware.live integration, 48h window, CISA feeds, validation updates, README

validation/review.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""
+Review validation results — pretty-prints results.jsonl with optional filters.
+
+Usage:
+    python3 validation/review.py                        # all results
+    python3 validation/review.py --category research   # filter by category
+    python3 validation/review.py --severity critical   # filter by severity
+    python3 validation/review.py --has cves            # only articles with CVEs
+    python3 validation/review.py --has malware         # only articles with malware hits
+    python3 validation/review.py --has actors          # only articles with threat actors
+    python3 validation/review.py --today               # only today's run
+"""
+
+import json
+import sys
+import argparse
+from pathlib import Path
+from datetime import datetime, timezone
+
+RESULTS_FILE = Path(__file__).parent / "results.jsonl"
+
+SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unknown": 4}
+
+
+def parse_args():
+    p = argparse.ArgumentParser()
+    p.add_argument("--category", help="Filter by category (news, malware, threat_intel, osint, research)")
+    p.add_argument("--severity", help="Filter by severity (critical, high, medium, low)")
+    p.add_argument("--has", choices=["cves", "malware", "actors"], help="Only show articles with these detections")
+    p.add_argument("--today", action="store_true", help="Only show results from today")
+    p.add_argument("--limit", type=int, default=0, help="Max articles to show (0 = all)")
+    return p.parse_args()
+
+
+def load_results(args):
+    if not RESULTS_FILE.exists():
+        print("No results.jsonl found — run run_validation.py first.")
+        sys.exit(0)
+
+    today = datetime.now(timezone.utc).date().isoformat()
+    records = []
+
+    with open(RESULTS_FILE) as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                r = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+
+            if args.today and not r.get("run_at", "").startswith(today):
+                continue
+            if args.category and r.get("category") != args.category:
+                continue
+            if args.severity and r.get("severity") != args.severity:
+                continue
+            if args.has == "cves" and not r.get("cves"):
+                continue
+            if args.has == "malware" and not r.get("malware_families"):
+                continue
+            if args.has == "actors" and not r.get("threat_actors"):
+                continue
+
+            records.append(r)
+
+    records.sort(key=lambda x: SEVERITY_ORDER.get(x.get("severity", "unknown"), 4))
+    return records
+
+
+def print_record(r):
+    sev = r.get("severity", "unknown").upper()
+    sev_icons = {"CRITICAL": "🚨", "HIGH": "🔴", "MEDIUM": "🟠", "LOW": "🟡"}
+    icon = sev_icons.get(sev, "⚪")
+
+    print(f"\n{'─' * 80}")
+    print(f"{icon} [{r.get('category', '?').upper()}] {r.get('title', 'No title')}")
+    print(f"   📡 {r.get('source')}  ·  {r.get('published')}  ·  Score: {r.get('quality_score', '?')}/100")
+
+    if r.get("cves"):
+        print(f"   🆔 {', '.join(r['cves'])}")
+    if r.get("threat_actors"):
+        print(f"   👤 {', '.join(r['threat_actors'])}")
+    if r.get("malware_families"):
+        print(f"   🦠 {', '.join(r['malware_families'])}")
+    if r.get("mitre_techniques"):
+        print(f"   🎯 {', '.join(r['mitre_techniques'][:5])}")
+    if r.get("description"):
+        print(f"   {r['description'][:200]}...")
+    print(f"   🔗 {r.get('url', '')}")
+
+
+def main():
+    args = parse_args()
+    records = load_results(args)
+
+    if not records:
+        print("No records match the filters.")
+        return
+
+    if args.limit:
+        records = records[:args.limit]
+
+    total = len(records)
+    sev_counts = {}
+    for r in records:
+        s = r.get("severity", "unknown")
+        sev_counts[s] = sev_counts.get(s, 0) + 1
+
+    print(f"\n=== Validation Results ({total} articles) ===")
+    print("  " + "  ".join(f"{s.upper()}: {c}" for s, c in sorted(sev_counts.items(), key=lambda x: SEVERITY_ORDER.get(x[0], 4))))
+
+    for r in records:
+        print_record(r)
+
+    print(f"\n{'─' * 80}")
+    print(f"Total: {total} articles")
+
+
+if __name__ == "__main__":
+    main()