telegram-rss-bot/validation/review.py

#!/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 --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"


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--category", help="Filter by category (news, malware, threat_intel, osint, research)")
    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.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)

    return records


def print_record(r):
    print(f"\n{'─' * 80}")
    print(f"[{r.get('category', '?').upper()}] {r.get('title', 'No title')}")
    print(f"   📡 {r.get('source')}  ·  {r.get('published')}")

    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)
    print(f"\n=== Validation Results ({total} articles) ===")

    for r in records:
        print_record(r)

    print(f"\n{'─' * 80}")
    print(f"Total: {total} articles")


if __name__ == "__main__":
    main()