From ce3b813e8b1f692b0d3d6dee4ef447c37b4ab23f Mon Sep 17 00:00:00 2001 From: bot Date: Tue, 25 Aug 2026 12:19:54 +0300 Subject: [PATCH] Add cross-source duplicate detection, extend retention to 14 days MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exact URL/text matching missed the same story from a different outlet (different URL, different wording). Add fuzzy title matching via difflib, checked both within a single poll batch (the common case — two feeds returning the same story in the same 5-minute cycle) and against a rolling window of recently-sent titles in the seen-DB (the cross-cycle case, e.g. follow-up coverage a few hours later). seen_articles.db/seen_victims.db retention bumped from 7 to 14 days (bot_config.SEEN_RETENTION_DAYS) — the window the new title matching actually needs, and no reason to keep dedup history longer than that. validation/run_validation.py: switch to the shared bot_config mapping, and prune results.jsonl to the same 14-day window each poll instead of growing forever — it's a testing aid, not an archive. Also fixes a bug where a future-dated (bad clock/feed) article stayed "recent" indefinitely instead of aging out after 48 hours, and drops the legacy MD5 url-hash field that SHA-256 fingerprints replaced. Co-Authored-By: Claude Sonnet 5 --- rss_manager.py | 93 ++++++++++++++++++++++++++---------- validation/run_validation.py | 47 +++++++++++++----- 2 files changed, 103 insertions(+), 37 deletions(-) diff --git a/rss_manager.py b/rss_manager.py index 636282a..131160b 100644 --- a/rss_manager.py +++ b/rss_manager.py @@ -10,6 +10,7 @@ import logging import asyncio import aiohttp import sqlite3 +import difflib from datetime import datetime, timezone, timedelta import hashlib from typing import Dict, List, Optional, Tuple @@ -18,6 +19,8 @@ from html import escape from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from pathlib import Path +from bot_config import CATEGORY_EMOJIS, SEEN_RETENTION_DAYS, TITLE_DEDUP_WINDOW_HOURS, TITLE_SIMILARITY_THRESHOLD + # Configure logging logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', @@ -32,6 +35,7 @@ class RSSFeedManager: self.feed_type = feed_type # 'daily', 'blogs', or 'general' self.feeds = {} self.seen_articles = {} # Changed to dict to store timestamps + self.recent_titles: List[str] = [] # normalized titles sent recently, for cross-source dedup self.session = None self.first_run = True self.load_feeds() @@ -71,6 +75,7 @@ class RSSFeedManager: self._ensure_seen_db() self._migrate_legacy_seen_json() self._load_seen_from_sqlite() + self._load_recent_titles() def _ensure_seen_db(self): """Create seen-article database and indexes if missing.""" @@ -87,10 +92,34 @@ class RSSFeedManager: conn.execute( "CREATE INDEX IF NOT EXISTS idx_seen_articles_seen_at ON seen_articles (seen_at)" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS sent_titles ( + title_norm TEXT NOT NULL, + seen_at TEXT NOT NULL + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_sent_titles_seen_at ON sent_titles (seen_at)" + ) conn.commit() except Exception as e: logger.error(f"Error initializing seen-article database: {e}") + def _load_recent_titles(self): + """Load normalized titles sent within the cross-source dedup window.""" + try: + cutoff_iso = (datetime.now(timezone.utc) - timedelta(hours=TITLE_DEDUP_WINDOW_HOURS)).isoformat() + with sqlite3.connect(self.seen_file) as conn: + cursor = conn.execute( + "SELECT title_norm FROM sent_titles WHERE seen_at >= ?", (cutoff_iso,) + ) + self.recent_titles = [row[0] for row in cursor.fetchall()] + except Exception as e: + logger.error(f"Error loading recent titles: {e}") + self.recent_titles = [] + def _load_seen_from_sqlite(self): """Load seen keys from SQLite into memory.""" try: @@ -149,7 +178,7 @@ class RSSFeedManager: def save_seen_articles(self): """Save seen article keys with retention in SQLite.""" try: - cutoff_time = datetime.now(timezone.utc) - timedelta(days=7) + cutoff_time = datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS) cutoff_iso = cutoff_time.isoformat() cleaned_articles = {} with sqlite3.connect(self.seen_file) as conn: @@ -166,13 +195,23 @@ class RSSFeedManager: logger.info(f"Saved {len(self.seen_articles)} seen articles (cleaned old entries)") except Exception as e: logger.error(f"Error saving seen articles: {e}") - - def generate_article_hash(self, article: Dict) -> str: - """Generate unique hash for article based on URL only for better duplicate detection""" - # Use only the URL for hash to catch duplicates across different RSS sources - # Different sources may have same article with different titles/dates - url = article.get('link', '') - return hashlib.md5(url.encode()).hexdigest() + + def _save_sent_title(self, title: str, seen_at: str): + """Record a sent article's normalized title for cross-source dedup, pruning old rows.""" + title_norm = self.normalize_text(title) + if not title_norm: + return + try: + cutoff_iso = (datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)).isoformat() + with sqlite3.connect(self.seen_file) as conn: + conn.execute("DELETE FROM sent_titles WHERE seen_at < ?", (cutoff_iso,)) + conn.execute( + "INSERT INTO sent_titles (title_norm, seen_at) VALUES (?, ?)", + (title_norm, seen_at), + ) + conn.commit() + except Exception as e: + logger.error(f"Error saving sent title: {e}") @staticmethod def canonicalize_url(url: str) -> str: @@ -220,22 +259,36 @@ class RSSFeedManager: } def is_article_recent(self, published_dt: datetime) -> bool: - """Allow articles published within the last 48 hours.""" + """Allow articles published within the last 48 hours (with brief clock-skew tolerance).""" if published_dt.tzinfo is None: published_dt = published_dt.replace(tzinfo=timezone.utc) - age = datetime.now(timezone.utc) - published_dt.astimezone(timezone.utc) - return age.total_seconds() <= 48 * 3600 + age_seconds = (datetime.now(timezone.utc) - published_dt.astimezone(timezone.utc)).total_seconds() + return -300 <= age_seconds <= 48 * 3600 + + @staticmethod + def titles_are_near_duplicate(title_norm_a: str, title_norm_b: str) -> bool: + """Fuzzy title match, tolerant of different outlets' wording for the same story.""" + return difflib.SequenceMatcher(None, title_norm_a, title_norm_b).ratio() >= TITLE_SIMILARITY_THRESHOLD + + def is_near_duplicate_title(self, title: str) -> bool: + """Fuzzy-match against recently sent titles to catch the same story from a different source.""" + title_norm = self.normalize_text(title) + if not title_norm: + return False + return any(self.titles_are_near_duplicate(title_norm, other) for other in self.recent_titles) def is_duplicate_article(self, article: Dict) -> bool: - """Check URL/content fingerprint duplicates against sent-history.""" + """Check URL/content fingerprint duplicates, plus fuzzy title matches against recent cross-source sends.""" fingerprints = self.get_article_fingerprints(article) url_key = fingerprints['url_key'] content_key = fingerprints['content_key'] - return (url_key and url_key in self.seen_articles) or (content_key in self.seen_articles) + if (url_key and url_key in self.seen_articles) or (content_key in self.seen_articles): + return True + return self.is_near_duplicate_title(article.get('title', '')) @classmethod def mark_article_as_sent(cls, article: Dict, seen_file: str = "seen_articles.db"): - """Persist fingerprints only after successful delivery.""" + """Persist fingerprints and title only after successful delivery.""" manager = cls(feeds_file="", seen_file=seen_file, feed_type=article.get('feed_type', 'general')) fingerprints = cls.get_article_fingerprints(article) now_iso = datetime.now(timezone.utc).isoformat() @@ -244,6 +297,7 @@ class RSSFeedManager: manager.seen_articles[fingerprints['url_key']] = now_iso manager.seen_articles[fingerprints['content_key']] = now_iso manager.save_seen_articles() + manager._save_sent_title(article.get('title', ''), now_iso) def clean_html(self, text: str) -> str: """Clean HTML tags and decode entities""" @@ -322,7 +376,6 @@ class RSSFeedManager: content = entry.description article = { - 'hash': self.generate_article_hash(entry), 'title': self.clean_html(getattr(entry, 'title', 'No Title')), 'description': self.clean_html(content), 'url': self.canonicalize_url(getattr(entry, 'link', '')), @@ -449,15 +502,7 @@ class RSSFeedManager: @staticmethod def format_telegram_message(article: Dict) -> Tuple[str, Optional[str]]: """Format article for Telegram message""" - category_emojis = { - 'news': '📰', - 'malware': '🦠', - 'threat_intel': '🛰️', - 'osint': '🕵️', - 'research': '🔬' - } - - emoji = category_emojis.get(article.get('category', ''), '📰') + emoji = CATEGORY_EMOJIS.get(article.get('category', ''), '📰') title = escape(article.get('title', 'No Title')) description = escape(article.get('description', '')) source = escape(article.get('source', 'Unknown Source')) diff --git a/validation/run_validation.py b/validation/run_validation.py index 5d82ed8..b8226b5 100644 --- a/validation/run_validation.py +++ b/validation/run_validation.py @@ -14,7 +14,7 @@ import aiohttp import json import os import sys -from datetime import datetime, timezone +from datetime import datetime, timezone, timedelta from pathlib import Path ROOT = Path(__file__).parent.parent @@ -33,19 +33,11 @@ if env_file.exists(): from rss_manager import RSSFeedManager from content_classifier import ContentClassifier from ransomware_fetcher import RansomwareFetcher - -CATEGORY_CONFIG = { - "news": "feeds/news_feeds.json", - "malware": "feeds/malware_feeds.json", - "threat_intel": "feeds/threat_intel_feeds.json", - "osint": "feeds/osint_feeds.json", - "research": "feeds/research_feeds.json", -} +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") -POLL_INTERVAL = 300 def save_record(f, record: dict): @@ -123,7 +115,7 @@ async def poll_victims(fetcher: RansomwareFetcher) -> int: "source": "ransomware.live", "url": article.get("url"), "published": article.get("published_human"), - "category": "malware", + "category": article.get("category"), "cves": [], "threat_actors": [], "malware_families": article.get("malware_families", []), @@ -135,6 +127,33 @@ async def poll_victims(fetcher: RansomwareFetcher) -> int: 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() @@ -166,9 +185,11 @@ async def main(): 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\n") + print(f" Saved {rss_count} articles + {victim_count} victims") else: - print(f" Nothing new\n") + print(f" Nothing new") + prune_results_file() + print() except Exception as e: print(f" Error: {e}\n")