telegram-rss-bot/validation/run_validation.py

#!/usr/bin/env python3
"""
Validation script — continuously fetches RSS feeds + ransomware.live victims,
classifies articles, appends everything to results.jsonl.
Runs independently from the bot using its own seen DBs.

Usage:
    python3 validation/run_validation.py
    Ctrl+C to stop
"""

import asyncio
import aiohttp
import json
import os
import sys
from datetime import datetime, timezone, timedelta
from pathlib import Path

ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(ROOT))

# Load .env from project root
env_file = ROOT / ".env"
if env_file.exists():
    with open(env_file) as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                k, v = line.split("=", 1)
                os.environ.setdefault(k.strip(), v.strip().strip("\"'"))

from rss_manager import RSSFeedManager
from content_classifier import ContentClassifier
from ransomware_fetcher import RansomwareFetcher
from bot_config import CATEGORY_FEEDS as CATEGORY_CONFIG, POLL_INTERVAL_SECONDS as POLL_INTERVAL, SEEN_RETENTION_DAYS

OUTPUT_FILE        = Path(__file__).parent / "results.jsonl"
SEEN_FILE          = str(Path(__file__).parent / "seen_validation.db")
SEEN_VICTIMS_FILE  = str(Path(__file__).parent / "seen_validation_victims.db")


def save_record(f, record: dict):
    f.write(json.dumps(record) + "\n")
    flags = []
    if record.get("cves"):             flags.append(f"CVEs: {', '.join(record['cves'][:3])}")
    if record.get("threat_actors"):    flags.append(f"Actors: {', '.join(record['threat_actors'][:2])}")
    if record.get("malware_families"): flags.append(f"Malware: {', '.join(record['malware_families'][:2])}")
    flag_str = f"  → {' | '.join(flags)}" if flags else ""
    category = record.get("category", "?").upper()
    print(f"  [{category:12s}] {record.get('title','')[:70]}{flag_str}")


async def poll_rss(classifier: ContentClassifier) -> int:
    all_articles = []
    for category, feeds_file in CATEGORY_CONFIG.items():
        feeds_path = str(ROOT / feeds_file)
        async with RSSFeedManager(feeds_file=feeds_path, seen_file=SEEN_FILE, feed_type=category) as manager:
            articles = await manager.fetch_all_feeds(initial_run=False)
            all_articles.extend(articles)

    if not all_articles:
        return 0

    unique = []
    seen_urls, seen_content = set(), set()
    for article in all_articles:
        fp = RSSFeedManager.get_article_fingerprints(article)
        url_key, content_key = fp["url_key"], fp["content_key"]
        if (url_key and url_key in seen_urls) or content_key in seen_content:
            continue
        if url_key:
            seen_urls.add(url_key)
        seen_content.add(content_key)
        unique.append(article)

    with open(OUTPUT_FILE, "a") as f:
        for article in unique:
            classified = classifier.classify_article(article)
            RSSFeedManager.mark_article_as_sent(classified, seen_file=SEEN_FILE)
            record = {
                "run_at":           datetime.now(timezone.utc).isoformat(),
                "type":             "article",
                "title":            classified.get("title"),
                "source":           classified.get("source"),
                "url":              classified.get("url"),
                "published":        classified.get("published_human"),
                "category":         classified.get("category"),
                "cves":             classified.get("cves", []),
                "threat_actors":    classified.get("threat_actors", []),
                "malware_families": classified.get("malware_families", []),
                "mitre_techniques": classified.get("mitre_techniques", []),
                "description":      classified.get("description", ""),
            }
            save_record(f, record)

    return len(unique)


async def poll_victims(fetcher: RansomwareFetcher) -> int:
    async with aiohttp.ClientSession() as session:
        new_victims = await fetcher.fetch_new_victims(session)

    if not new_victims:
        return 0

    with open(OUTPUT_FILE, "a") as f:
        for victim in new_victims:
            article = RansomwareFetcher.to_article(victim)
            fetcher.mark_seen(victim.get("id", ""))
            record = {
                "run_at":           datetime.now(timezone.utc).isoformat(),
                "type":             "victim",
                "title":            article.get("title"),
                "source":           "ransomware.live",
                "url":              article.get("url"),
                "published":        article.get("published_human"),
                "category":         article.get("category"),
                "cves":             [],
                "threat_actors":    [],
                "malware_families": article.get("malware_families", []),
                "mitre_techniques": [],
                "description":      article.get("description", ""),
            }
            save_record(f, record)

    return len(new_victims)


def prune_results_file():
    """Keep results.jsonl bounded to the same window dedup actually needs — this is a
    testing aid, not an archive, so nothing here needs to outlive SEEN_RETENTION_DAYS."""
    if not OUTPUT_FILE.exists():
        return
    cutoff = (datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)).isoformat()
    kept = []
    dropped = 0
    with open(OUTPUT_FILE) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                continue
            if record.get("run_at", "") >= cutoff:
                kept.append(line)
            else:
                dropped += 1
    if dropped:
        with open(OUTPUT_FILE, "w") as f:
            f.write("\n".join(kept) + ("\n" if kept else ""))
        print(f"  Pruned {dropped} record(s) older than {SEEN_RETENTION_DAYS} days")


async def main():
    classifier = ContentClassifier()

    ransomware_key = os.getenv("RANSOMWARE_LIVE_API_KEY", "")
    fetcher = RansomwareFetcher(ransomware_key, seen_file=SEEN_VICTIMS_FILE) if ransomware_key else None

    print(f"Validation monitor started — polling every {POLL_INTERVAL}s. Ctrl+C to stop.")
    print(f"RSS feeds:        {'enabled' if CATEGORY_CONFIG else 'disabled'}")
    print(f"ransomware.live:  {'enabled' if fetcher else 'disabled (no API key)'}")
    print(f"Output: {OUTPUT_FILE}\n")

    # Initial scan — mark everything currently visible as seen
    print(f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}] Initial scan...")
    for category, feeds_file in CATEGORY_CONFIG.items():
        feeds_path = str(ROOT / feeds_file)
        async with RSSFeedManager(feeds_file=feeds_path, seen_file=SEEN_FILE, feed_type=category) as manager:
            await manager.fetch_all_feeds(initial_run=True)
    if fetcher:
        async with aiohttp.ClientSession() as session:
            await fetcher.fetch_new_victims(session, initial_run=True)
    print("  Done. Watching for new items from now on.\n")

    while True:
        await asyncio.sleep(POLL_INTERVAL)
        ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
        print(f"[{ts}] Polling...")
        try:
            rss_count = await poll_rss(classifier)
            victim_count = await poll_victims(fetcher) if fetcher else 0
            total = rss_count + victim_count
            if total:
                print(f"  Saved {rss_count} articles + {victim_count} victims")
            else:
                print(f"  Nothing new")
            prune_results_file()
            print()
        except Exception as e:
            print(f"  Error: {e}\n")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopped.")