athena-oracle/oracle/dedup.py

"""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