athena-oracle/pipeline.py

#!/usr/bin/env python3
"""
AI Research Oracle — Pipeline Orchestrator.

Runs source adapters, deduplicates, stores to unified SQLite DB.
Designed so adding a new adapter is one line of registration.

Usage:
    python3 pipeline.py                          # run all enabled adapters
    python3 pipeline.py --sources github,arxiv   # specific sources
    python3 pipeline.py --limit 15               # per-source limit
    python3 pipeline.py --dry-run                # fetch but don't store
    python3 pipeline.py --verify                 # spot-check N entries
"""

import argparse
import json
import os
import sqlite3
import sys
import time
from datetime import datetime, timezone

# Allow running from project root
sys.path.insert(0, os.path.dirname(__file__))

from adapters import SourceAdapter

# Adapter registry — add new adapters here (one line each)
ADAPTERS = {
    "github": lambda: __import__("adapters.github", fromlist=["GitHubAdapter"]).GitHubAdapter(),
    "arxiv": lambda: __import__("adapters.arxiv", fromlist=["ArxivAdapter"]).ArxivAdapter(),
    "reddit": lambda: __import__("adapters.reddit", fromlist=["RedditAdapter"]).RedditAdapter(),
    "hackernews": lambda: __import__("adapters.hackernews", fromlist=["HackerNewsAdapter"]).HackerNewsAdapter(),
    "huggingface": lambda: __import__("adapters.huggingface", fromlist=["HuggingFaceAdapter"]).HuggingFaceAdapter(),
    "rss": lambda: __import__("adapters.rss_feeds", fromlist=["RSSFeedsAdapter"]).RSSFeedsAdapter(),
}

# Default enabled sources
ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"]


def init_db(db_path: str, schema_path: str) -> sqlite3.Connection:
    """Initialize or open the database."""
    conn = sqlite3.connect(db_path)
    if os.path.exists(schema_path):
        with open(schema_path) as f:
            conn.executescript(f.read())
        conn.commit()
    return conn


def store_entries(conn: sqlite3.Connection, entries: list[dict]) -> int:
    """Store entries using INSERT OR REPLACE (dedup by source+source_id)."""
    cur = conn.cursor()
    stored = 0
    for entry in entries:
        try:
            cur.execute("""
                INSERT OR REPLACE INTO entries
                (source, source_id, url, title, extracted_text, summary,
                 category_tags, signal_score, raw_metadata, first_seen, last_updated)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                entry["source"], entry["source_id"], entry["url"], entry["title"],
                entry["extracted_text"] or "", entry["summary"],  # None → NULL in DB
                entry["category_tags"], entry["signal_score"],
                entry["raw_metadata"], entry["first_seen"], entry["last_updated"],
            ))
            stored += 1
        except Exception as e:
            print(f"  ⚠ DB error on {entry.get('source', '?')}/{entry.get('source_id', '?')}: {e}")
    conn.commit()
    return stored


def verify_entries(conn: sqlite3.Connection, source: str, sample_size: int = 3):
    """Spot-check N entries from a source against live data.

    This is a standing quality gate: after each adapter run, verify a
    random sample of high-signal entries match the live source.
    Protects against bad batches propagating to the reasoning layer.
    """
    import urllib.request
    import random

    cur = conn.cursor()
    cur.execute("""
        SELECT id, source, source_id, url, title, raw_metadata, signal_score
        FROM entries WHERE source = ?
        ORDER BY signal_score DESC
        LIMIT ?
    """, (source, sample_size))

    rows = cur.fetchall()
    if not rows:
        print(f"  No entries to verify for {source}")
        return True

    passed = 0
    for row in rows:
        eid, src, sid, url, title, meta_json, score = row
        meta = json.loads(meta_json) if meta_json else {}

        # Source-specific verification
        if src == "github":
            repo = meta.get("full_name", "")
            db_stars = meta.get("stars", 0)
            if repo:
                try:
                    api_url = f"https://api.github.com/repos/{repo}"
                    req = urllib.request.Request(api_url, headers={
                        "Accept": "application/vnd.github.v3+json",
                        "User-Agent": "ai-oracle/0.1",
                    })
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        live = json.loads(resp.read().decode("utf-8"))
                    live_stars = live.get("stargazers_count", 0)
                    drift = abs(live_stars - db_stars)
                    # Allow up to 1% drift or 100 stars (whichever larger)
                    threshold = max(int(db_stars * 0.01), 100)
                    if drift <= threshold:
                        print(f"  ✓ [{eid}] {title[:60]}... stars={db_stars} drift={drift}")
                        passed += 1
                    else:
                        print(f"  ✗ [{eid}] {title[:60]}... stars={db_stars} vs live={live_stars} DRIFT={drift}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        elif src == "arxiv":
            arxiv_id = meta.get("arxiv_id", "")
            if arxiv_id:
                # Clean version suffix for URL
                clean_id = arxiv_id.split("v")[0]
                try:
                    req = urllib.request.Request(
                        f"https://arxiv.org/abs/{clean_id}",
                        headers={"User-Agent": "ai-oracle/0.1"}
                    )
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        status = resp.status
                    if status == 200:
                        print(f"  ✓ [{eid}] {title[:60]}... arxiv.org/abs/{clean_id} exists")
                        passed += 1
                    else:
                        print(f"  ? [{eid}] {title[:60]}... HTTP {status}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        elif src == "hackernews":
            hn_id = meta.get("id", "")
            if hn_id:
                try:
                    req = urllib.request.Request(
                        f"https://hacker-news.firebaseio.com/v0/item/{hn_id}.json",
                        headers={"User-Agent": "ai-oracle/0.1"},
                    )
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        live = json.loads(resp.read().decode("utf-8"))
                    live_score = live.get("score", 0)
                    db_score = meta.get("score", 0)
                    drift = abs(live_score - db_score)
                    if drift <= 20:
                        print(f"  ✓ [{eid}] {title[:60]}... hn_score={db_score} drift={drift}")
                        passed += 1
                    else:
                        print(f"  ? [{eid}] {title[:60]}... hn_score={db_score} vs live={live_score} DRIFT={drift}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        elif src == "huggingface":
            model_id = meta.get("model_id", "")
            if model_id:
                try:
                    req = urllib.request.Request(
                        f"https://huggingface.co/api/models/{model_id}",
                        headers={"User-Agent": "athena/0.1"},
                    )
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        live = json.loads(resp.read().decode("utf-8"))
                    live_likes = live.get("likes", 0)
                    db_likes = meta.get("likes", 0)
                    drift = abs(live_likes - db_likes)
                    if drift <= 10:  # HF likes update in real-time, allow small drift
                        print(f"  ✓ [{eid}] {title[:60]}... hf_likes={db_likes} drift={drift}")
                        passed += 1
                    else:
                        print(f"  ? [{eid}] {title[:60]}... hf_likes={db_likes} vs live={live_likes} DRIFT={drift}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        elif src == "reddit":
            # For Reddit, just check the URL is reachable (no easy verification of scores via RSS)
            if url:
                try:
                    req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"})
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        status = resp.status
                    if status in (200, 302):
                        print(f"  ✓ [{eid}] {title[:60]}... URL reachable")
                        passed += 1
                    else:
                        print(f"  ? [{eid}] {title[:60]}... HTTP {status}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        elif src == "rss":
            # For RSS, verify the article URL is reachable
            if url:
                try:
                    req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"})
                    with urllib.request.urlopen(req, timeout=10) as resp:
                        status = resp.status
                    if status in (200, 301, 302):
                        print(f"  ✓ [{eid}] {title[:60]}... URL reachable (HTTP {status})")
                        passed += 1
                    else:
                        print(f"  ? [{eid}] {title[:60]}... HTTP {status}")
                except Exception as e:
                    print(f"  ? [{eid}] {title[:60]}... verify failed: {e}")

        time.sleep(0.5)  # polite spacing

    return passed > 0


def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: bool = False, verify: bool = False):
    """Run the ingestion pipeline."""
    sources = sources or ENABLED_SOURCES
    now = datetime.now(timezone.utc)
    print(f"=== AI Research Oracle Pipeline ===")
    print(f"  Sources: {', '.join(sources)}")
    print(f"  Limit: {limit}/source")
    print(f"  Dry run: {dry_run}")
    print()

    db_path = os.path.join(os.path.dirname(__file__), "oracle.db")
    schema_path = os.path.join(os.path.dirname(__file__), "schema.sql")

    all_entries = []
    source_stats = {}

    for source_name in sources:
        if source_name not in ADAPTERS:
            print(f"  ⚠ Unknown source: {source_name} (available: {', '.join(ADAPTERS.keys())})")
            continue

        print(f"  [{source_name}]")
        adapter = ADAPTERS[source_name]()

        try:
            entries = adapter.fetch(limit=limit)
        except Exception as e:
            print(f"  ✗ {source_name} failed: {e}")
            source_stats[source_name] = {"fetched": 0, "stored": 0, "error": str(e)}
            continue

        # Add adapter_version to metadata
        for entry in entries:
            meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"]
            meta["adapter_version"] = "0.1"
            entry["raw_metadata"] = json.dumps(meta)

        all_entries.extend(entries)
        source_stats[source_name] = {"fetched": len(entries), "stored": 0}
        print(f"    Fetched: {len(entries)} entries")

        # Small spacing between sources
        time.sleep(1)

    # Store
    if not dry_run and all_entries:
        conn = init_db(db_path, schema_path)
        stored = store_entries(conn, all_entries)

        # Update per-source stored counts
        for entry in all_entries:
            src = entry["source"]
            if src in source_stats:
                source_stats[src]["stored"] += 1

        # Verification
        if verify:
            print(f"\n  [Verification]")
            for src in sources:
                if source_stats.get(src, {}).get("stored", 0) > 0:
                    print(f"    Checking {src}...")
                    verify_entries(conn, src, sample_size=3)
            print()

        # Record run log (failure visibility + growth control) — before conn.close()
        ok = [s for s, st in source_stats.items() if not st.get("error")]
        failed = [s for s, st in source_stats.items() if st.get("error")]
        # Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but
        # is still a degraded run — record it so run_log can tell
        # "intermittent vs consistently-broken" apart over time.
        zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")]
        notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")]
        if zero:
            notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}")
        notes = "; ".join(notes_parts) or "all sources ok"
        try:
            conn.execute("""
                INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
                VALUES (?, ?, ?, ?, ?)
            """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes))
            conn.commit()
        except Exception as e:
            print(f"  ⚠ run_log write failed: {e}")

        conn.close()
        print(f"  Total stored: {stored} entries")
    else:
        print(f"  Total fetched: {len(all_entries)} entries (dry run, not stored)")

    # Summary
    print(f"\n  Source summary:")
    for src, stats in source_stats.items():
        error = stats.get("error", "")
        error_str = f" ✗ {error}" if error else ""
        print(f"    {src}: {stats['fetched']} fetched, {stats.get('stored', '—')} stored{error_str}")

    # Top entries across all sources
    if all_entries:
        print(f"\n  Recent entries by signal score (per-source ranking, NOT cross-source comparable):")
        print(f"  (Note: GitHub scores use actual star counts; arXiv/reddit use estimated heuristics)")
        sorted_entries = sorted(all_entries, key=lambda e: e["signal_score"], reverse=True)
        for i, entry in enumerate(sorted_entries[:5]):
            meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"]
            score_type = meta.get("score_type", "?")
            print(f"    [{i+1}] {entry['source'].upper():6} ({score_type:8}) score={entry['signal_score']:.2f} {entry['title'][:75]}")

    print(f"\n  Done.")
    return all_entries


def main():
    parser = argparse.ArgumentParser(description="AI Research Oracle Pipeline")
    parser.add_argument("--sources", default=None, help="Comma-separated sources (default: github,arxiv)")
    parser.add_argument("--limit", type=int, default=20, help="Entries per source")
    parser.add_argument("--dry-run", action="store_true", help="Fetch but don't store")
    parser.add_argument("--verify", action="store_true", help="Spot-check entries against live sources")
    args = parser.parse_args()

    sources = args.sources.split(",") if args.sources else None
    run_pipeline(sources=sources, limit=args.limit, dry_run=args.dry_run, verify=args.verify)


if __name__ == "__main__":
    main()