"""Content-hash dedup and composite verdict engine. World Monitor pattern: SHA-256 content hash for dedup, tier-weighted composite verdict (PUBLISH/WATCH/ARCHIVE/DROP) on top of final_score. """ import hashlib import re from datetime import datetime, timezone from oracle.config import ( SOURCE_TIERS, TIER_BONUS, VERDICT_THRESHOLDS, HASH_LENGTH, CONTENT_HASH_PREFIX, ) def content_hash(title: str, url: str = "", body: str = "") -> str: """Deterministic content hash for dedup. Normalizes whitespace, lowercases, strips HTML tags, then hashes. Returns hex[:HASH_LENGTH] for compact storage. """ text = f"{title}|{body[:500]}|{url}" text = re.sub(r'\s+', ' ', text).strip().lower() text = re.sub(r'<[^>]+>', '', text) raw = hashlib.sha256(text.encode()).hexdigest() return f"{CONTENT_HASH_PREFIX}:{raw[:HASH_LENGTH]}" def check_duplicate(conn, title: str, url: str = "", body: str = "", cutoff_days: int = 7) -> bool: """Check if an entry with similar content_hash already exists within cutoff.""" h = content_hash(title, url, body) now = datetime.now(timezone.utc) cur = conn.cursor() cur.execute( "SELECT COUNT(*) FROM entries WHERE content_hash = ? AND first_seen > ?", (h, (now.timestamp() - cutoff_days * 86400)), ) count = cur.fetchone()[0] return count > 0 def get_source_tier(source: str) -> dict: """Return tier info for a source. Defaults to tier 2.""" return SOURCE_TIERS.get(source, {"tier": 2, "label": "SECONDARY", "description": "Unknown source"}) def tier_adjusted_score(base_score: float, source: str) -> float: """Apply tier bonus/penalty to a base signal score.""" tier_info = get_source_tier(source) tier_num = tier_info["tier"] bonus = TIER_BONUS.get(tier_num, 0.0) return round(base_score + bonus, 3) def compute_verdict(score: float, age_hours: float) -> str: """Compute composite verdict from score + age. Uses signal_score (0-10 scale). PUBLISH > WATCH > ARCHIVE > DROP. """ for verdict, thresholds in VERDICT_THRESHOLDS.items(): if score >= thresholds["min_score"] and age_hours <= thresholds["max_age_h"]: return verdict return "DROP" def age_hours(first_seen_iso: str) -> float: """Return age in hours from ISO timestamp.""" try: ts = first_seen_iso.replace("Z", "+00:00") first = datetime.fromisoformat(ts) now = datetime.now(timezone.utc) return max(0, (now - first).total_seconds() / 3600) except (ValueError, AttributeError): return 0.0 def apply_verdicts(conn): """Update verdict column for all entries that lack one. Uses signal_score (0-10 scale) + first_seen age to compute verdict. """ cur = conn.cursor() # Check if verdict column exists cur.execute("PRAGMA table_info(entries)") columns = {row[1] for row in cur.fetchall()} if "verdict" not in columns: print(" [verdict] column not found, skipping apply") return 0 # Reset all verdicts so they get recalculated cur.execute("UPDATE entries SET verdict = ''") conn.commit() # Fetch all entries with signal scores cur.execute("SELECT id, COALESCE(signal_score, 0), first_seen FROM entries") updated = 0 for row in cur.fetchall(): entry_id, signal_score, first_seen = row age = age_hours(first_seen) verdict = compute_verdict(signal_score, age) cur.execute("UPDATE entries SET verdict = ? WHERE id = ?", (verdict, entry_id)) updated += 1 conn.commit() return updated def backfill_hashes(conn, batch_size: int = 500) -> int: """Backfill content_hash for entries that lack one.""" cur = conn.cursor() cur.execute("PRAGMA table_info(entries)") columns = {row[1] for row in cur.fetchall()} if "content_hash" not in columns: print(" [dedup] content_hash column not found, skipping backfill") return 0 updated = 0 while True: cur.execute( "SELECT id, title, url, summary FROM entries " "WHERE content_hash IS NULL OR content_hash = '' " "LIMIT ?", (batch_size,), ) rows = cur.fetchall() if not rows: break for entry_id, title, url, summary in rows: body = "" if summary: import json try: s = json.loads(summary) body = s.get("one_liner", "") + " " + s.get("key_points", "") except (json.JSONDecodeError, TypeError): body = summary[:200] h = content_hash(title, url, body) cur.execute("UPDATE entries SET content_hash = ? WHERE id = ?", (h, entry_id)) updated += 1 conn.commit() return updated