"""CLI entry point for the AI Research Oracle. Single command with subcommands for all operations. Replaces scattered root scripts with one unified interface. Usage: python -m oracle [args] Commands: ingest Run the ingestion pipeline (fetch + store + score) summarize Generate summaries for unscored entries query Query the database (top, search, recent, stats, snapshot) render Render static site from clickability index archive Soft-cap archival of old entries themes Theme-based trend tracking recency Recency guard analysis health System health check """ import argparse import json import os import sys from datetime import datetime, timezone # Ensure project root is on path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from oracle.config import DB_PATH, SCHEMA_PATH, ENABLED_SOURCES, DEFAULT_LIMIT, SOURCE_TIERS, VERDICT_THRESHOLDS from oracle.db import get_connection, init_db, get_stats, query_top, query_recent, query_search, query_by_tag, migrate_world_monitor from oracle.scoring import attach_scoring, migrate as scoring_migrate from oracle.summarize import run_summarization from oracle.render import render as render_site from oracle.archive import archive as archive_entries from oracle.themes import scan as theme_scan from oracle.clickability import fetch_items, compute_index, decay_index from oracle.recency import filter_fresh from oracle.dedup import content_hash, compute_verdict, tier_adjusted_score, backfill_hashes, apply_verdicts, get_source_tier def cmd_ingest(args): """Run the ingestion pipeline.""" import time import sqlite3 import signal from adapters import SourceAdapter from adapters._store import upsert_entries sources = args.sources.split(",") if args.sources else ENABLED_SOURCES now = datetime.now(timezone.utc) print(f"=== AI Research Oracle Pipeline ===") print(f" Sources: {', '.join(sources)}") print(f" Limit: {args.limit}/source") print(f" Dry run: {args.dry_run}") print() # Import adapters dynamically adapter_modules = { "github": "adapters.github", "arxiv": "adapters.arxiv", "reddit": "adapters.reddit", "hackernews": "adapters.hackernews", "huggingface": "adapters.huggingface", "rss": "adapters.rss_feeds", } adapter_classes = { "github": "GitHubAdapter", "arxiv": "ArxivAdapter", "reddit": "RedditAdapter", "hackernews": "HackerNewsAdapter", "huggingface": "HuggingFaceAdapter", "rss": "RSSFeedsAdapter", } db_path = str(DB_PATH) schema_path = str(SCHEMA_PATH) all_entries = [] source_stats = {} for source_name in sources: if source_name not in adapter_modules: print(f" ⚠ Unknown source: {source_name}") continue print(f" [{source_name}]") mod = __import__(adapter_modules[source_name], fromlist=[adapter_classes[source_name]]) adapter = getattr(mod, adapter_classes[source_name])() # Fetch with per-adapter timeout (prevents blocking on slow endpoints) entries = [] error = None try: entries = adapter.fetch(limit=args.limit, timeout=10) except TypeError: # Old adapter signature without timeout param — use thread-based fallback import threading result = {"entries": [], "error": None} def _fetch(): try: result["entries"] = adapter.fetch(limit=args.limit) except Exception as e: result["error"] = str(e) t = threading.Thread(target=_fetch, daemon=True) t.start() t.join(timeout=10) if t.is_alive(): error = f"timeout after 10s" else: entries = result["entries"] error = result["error"] except Exception as e: error = str(e) if error: print(f" ✗ {source_name} failed: {error}") source_stats[source_name] = {"fetched": 0, "stored": 0, "error": error} continue for entry in entries: meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"] meta["adapter_version"] = "1.0" 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") time.sleep(0.5) if not args.dry_run and all_entries: conn = init_db(db_path, schema_path) stored = upsert_entries(conn, all_entries) for entry in all_entries: src = entry["source"] if src in source_stats: source_stats[src]["stored"] += 1 # Score new entries try: attach_scoring(db_path) except Exception as e: print(f" ⚠ scoring attach failed: {e}") conn.close() print(f" Total stored: {stored} entries") else: print(f" Total fetched: {len(all_entries)} entries (dry run)") 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}") if all_entries: print(f"\n Top entries by signal score:") sorted_entries = sorted(all_entries, key=lambda e: e["signal_score"], reverse=True) for i, entry in enumerate(sorted_entries[:5]): print(f" [{i+1}] {entry['source'].upper():6} score={entry['signal_score']:.2f} {entry['title'][:70]}") print(f"\n Done.") def cmd_summarize(args): """Run the summarization engine.""" run_summarization(source=args.source, limit=args.limit) def cmd_query(args): """Query the database.""" conn = get_connection() qc = getattr(args, "query_command", None) if qc == "top": entries = query_top(conn, n=args.n, source=args.source, min_score=args.min_score) print(f"Top {len(entries)} entries:") _print_entries(entries) elif qc == "search": entries = query_search(conn, args.query_text, limit=args.limit) print(f"Search results for '{args.query_text}':") _print_entries(entries) elif qc == "recent": entries = query_recent(conn, hours=args.hours) print(f"Entries from last {args.hours}h:") _print_entries(entries) elif qc == "by-tag": entries = query_by_tag(conn, args.tag, limit=args.limit) print(f"Entries tagged '{args.tag}':") _print_entries(entries) elif qc == "stats": stats = get_stats(conn) print(f"Database: {DB_PATH}") print(f"Total entries: {stats['total_entries']}") print(f"Summarized: {stats['summarized']}") print(f"Pending summary: {stats['pending_summary']}") if stats.get("buckets"): print(f"\nBucket distribution:") for bucket, count in sorted(stats["buckets"].items(), key=lambda x: -x[1]): print(f" {bucket}: {count}") elif qc == "snapshot": stats = get_stats(conn) top = query_top(conn, n=10) print("=" * 70) print("AI RESEARCH ORACLE — SNAPSHOT") print("=" * 70) print(f"Time: {datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}") print(f"Total entries: {stats['total_entries']}") print() print("Source breakdown:") for src, s in stats["sources"].items(): print(f" {src}: {s['cnt']} entries, avg score {s['avg_score']}") print() print("Top 10 by signal score:") _print_entries(top) conn.close() def _print_entries(entries): """Pretty-print a list of entries.""" if not entries: print(" (no results)") return for i, e in enumerate(entries, 1): summary = json.loads(e["summary"]) if e.get("summary") else {} meta = json.loads(e["raw_metadata"]) if e.get("raw_metadata") else {} print(f" [{i}] {e['source'].upper():10} | score={e['signal_score']:.2f}") print(f" {e['title'][:80]}") if summary.get("one_liner"): print(f" → {summary['one_liner'][:100]}") print() def cmd_render(args): """Render static site — single variant or all variants.""" from oracle.variants import load_variant, list_variants, render_variant if args.list: print("Available variants:") for v in list_variants(): print(f" {v}") return if args.all_variants: variants = list_variants() if not variants: print("No variants found in variants/") return print(f"=== Rendering all {len(variants)} variants ===\n") for vname in variants: try: config = load_variant(vname) render_variant(config, dry_run=args.dry_run, webroot=args.webroot) except Exception as e: print(f"[variant:{vname}] ERROR: {e}") print() return # Single variant (default = 'default' if not specified) vname = args.variant or "default" try: config = load_variant(vname) render_variant(config, dry_run=args.dry_run, webroot=args.webroot) except FileNotFoundError as e: print(f"Error: {e}") def cmd_archive(args): """Archive old entries.""" archive_entries(days=args.days, cap=args.cap, dry_run=args.dry_run) def cmd_themes(args): """Run theme scan.""" results = theme_scan(history=args.history) print(f"=== Theme trend scan ===") print(f" Fresh entries this cycle: {results['fresh_count']}") if results["new_arrivals"]: print(" NEW theme arrivals this cycle:") for theme, count in results["new_arrivals"].items(): print(f" {theme}: +{count}") else: print(" NEW theme arrivals this cycle: 0") print(f" Cumulative totals: {results['cumulative']}") if args.history and results.get("history"): print("\n Per-cycle history:") for day, theme, count in results["history"]: print(f" {day} {theme}: {count}") def cmd_dedup(args): """World Monitor migration: tiers, hashes, verdicts.""" print("=== World Monitor Migration ===\n") conn = get_connection() # Show source tier config if args.show_tiers: print("Source tier configuration:") for src, info in SOURCE_TIERS.items(): print(f" {src:12} Tier {info['tier']} ({info['label']}) - {info['description']}") print() # Show verdict thresholds if args.show_verdicts: print("Verdict thresholds:") for verdict, thresholds in VERDICT_THRESHOLDS.items(): print(f" {verdict:8} score >= {thresholds['min_score']}, age <= {thresholds['max_age_h']}h") print() # Run migration if args.migrate: print("Running schema migration + backfill...\n") result = migrate_world_monitor(conn) print(f" Columns added: {result['columns_added']}") print(f" Hashes backfilled: {result['hashes_backfilled']}") print(f" Verdicts set: {result['verdicts_set']}") print() # Show verdict distribution cur = conn.cursor() try: cur.execute("PRAGMA table_info(entries)") columns = {r[1] for r in cur.fetchall()} except Exception: conn.close() return if "verdict" in columns: cur.execute("SELECT verdict, COUNT(*) as cnt FROM entries WHERE verdict != '' GROUP BY verdict ORDER BY cnt DESC") rows = cur.fetchall() if rows: print("Verdict distribution:") for r in rows: print(f" {r['verdict']:8} {r['cnt']}") print() if "content_hash" in columns: cur.execute("SELECT COUNT(*) FROM entries WHERE content_hash != '' AND content_hash IS NOT NULL") hashed = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM entries") total = cur.fetchone()[0] print(f"Content hashes: {hashed}/{total} entries hashed") if "source_tier" in columns: cur.execute("SELECT source_tier, COUNT(*) as cnt FROM entries GROUP BY source_tier ORDER BY source_tier") rows = cur.fetchall() if rows: print(f"\nSource tier distribution:") for r in rows: print(f" Tier {r['source_tier']}: {r['cnt']}") conn.close() print() def cmd_brief(args): """Generate morning intelligence briefing.""" from oracle.brief import generate_brief, generate_brief_json, get_connection db = args.db or str(DB_PATH) conn = get_connection(db) try: brief = generate_brief_json(conn, max_items=args.max_items, max_age_h=args.max_age_h) if args.json: import json as _json print(_json.dumps(brief, indent=2, default=str)) elif args.output: generate_brief(conn, max_items=args.max_items, max_age_h=args.max_age_h, output_path=args.output) print(f"Saved: {args.output}") else: print(brief.get("markdown", brief.get("error", "No brief generated"))) finally: conn.close() def cmd_weekly(args): """Generate weekly review.""" from oracle.weekly import generate_weekly, get_connection db = args.db or str(DB_PATH) conn = get_connection(db) try: review = generate_weekly(conn, days=args.days) if args.json: import json as _json print(_json.dumps({k: v for k, v in review.items() if k != "markdown"}, indent=2, default=str)) elif args.output: generate_weekly(conn, days=args.days, output_path=args.output) print(f"Saved: {args.output}") else: print(review.get("markdown", "No review generated")) finally: conn.close() def cmd_health(args): """System health check.""" print("=== System Health Check ===\n") # Database try: conn = get_connection() stats = get_stats(conn) conn.close() print(f"✓ Database: {DB_PATH}") print(f" Total entries: {stats['total_entries']}") print(f" Summarized: {stats['summarized']}") print(f" Pending: {stats['pending_summary']}") except Exception as e: print(f"✗ Database error: {e}") # Schema columns try: conn = get_connection() cur = conn.cursor() cur.execute("PRAGMA table_info(entries)") columns = [r[1] for r in cur.fetchall()] conn.close() print(f" Schema columns: {len(columns)}") except Exception as e: print(f" Schema check: {e}") # Run log try: conn = get_connection() cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM run_log") runs = cur.fetchone()[0] cur.execute("SELECT run_time, failure_class FROM run_log ORDER BY id DESC LIMIT 3") recent = cur.fetchall() conn.close() print(f" Pipeline runs logged: {runs}") for r in recent: print(f" {r[0]} class={r[1]}") except Exception as e: print(f" Run log: {e}") print() def main(): parser = argparse.ArgumentParser( description="AI Research Oracle — Unified CLI", formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = parser.add_subparsers(dest="command", help="Available commands") # ingest p_ingest = sub.add_parser("ingest", help="Run the ingestion pipeline") p_ingest.add_argument("--sources", default=None, help="Comma-separated sources") p_ingest.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="Entries per source") p_ingest.add_argument("--dry-run", action="store_true", help="Fetch but don't store") # summarize p_summarize = sub.add_parser("summarize", help="Generate summaries") p_summarize.add_argument("--source", default=None, help="Filter by source") p_summarize.add_argument("--limit", type=int, default=0, help="Max entries (0=all)") # query (nested subcommands) p_query = sub.add_parser("query", help="Query the database") query_sub = p_query.add_subparsers(dest="query_command") p_top = query_sub.add_parser("top", help="Top N entries") p_top.add_argument("n", type=int, nargs="?", default=10) p_top.add_argument("--source", default=None) p_top.add_argument("--min-score", type=float, default=0) p_search = query_sub.add_parser("search", help="Keyword search") p_search.add_argument("query_text") p_search.add_argument("--limit", type=int, default=20) p_recent = query_sub.add_parser("recent", help="Recent entries") p_recent.add_argument("--hours", type=int, default=24) p_tag = query_sub.add_parser("by-tag", help="Entries by tag") p_tag.add_argument("tag") p_tag.add_argument("--limit", type=int, default=20) query_sub.add_parser("stats", help="Database statistics") query_sub.add_parser("snapshot", help="Full snapshot") # render p_render = sub.add_parser("render", help="Render static site") p_render.add_argument("--variant", default=None, help="Variant name (default, research, devops, brief)") p_render.add_argument("--all", dest="all_variants", action="store_true", help="Render all variants") p_render.add_argument("--list", action="store_true", help="List available variants") p_render.add_argument("--dry-run", action="store_true") p_render.add_argument("--webroot", default=None, help="Output directory") # archive p_archive = sub.add_parser("archive", help="Archive old entries") p_archive.add_argument("--days", type=int, default=30) p_archive.add_argument("--cap", type=int, default=5000) p_archive.add_argument("--dry-run", action="store_true") # themes p_themes = sub.add_parser("themes", help="Theme trend tracking") p_themes.add_argument("--history", action="store_true") # dedup (World Monitor migration) p_dedup = sub.add_parser("dedup", help="World Monitor: tiers, hashes, verdicts") p_dedup.add_argument("--migrate", action="store_true", help="Run schema migration + backfill") p_dedup.add_argument("--show-tiers", action="store_true", help="Show source tier config") p_dedup.add_argument("--show-verdicts", action="store_true", help="Show verdict thresholds") # brief p_brief = sub.add_parser("brief", help="Generate morning intelligence briefing") p_brief.add_argument("--db", default=None, help="Database path") p_brief.add_argument("--max-items", type=int, default=8, help="Max items in brief") p_brief.add_argument("--max-age-h", type=float, default=48, help="Max age in hours") p_brief.add_argument("--output", "-o", help="Output path for markdown brief") p_brief.add_argument("--json", action="store_true", help="Output as JSON") # weekly p_weekly = sub.add_parser("weekly", help="Generate weekly review") p_weekly.add_argument("--db", default=None, help="Database path") p_weekly.add_argument("--days", type=int, default=7, help="Days to review") p_weekly.add_argument("--output", "-o", help="Output path for markdown review") p_weekly.add_argument("--json", action="store_true", help="Output as JSON") # metrics p_metrics = subparsers.add_parser("metrics", help="Pipeline metrics and adapter health") p_metrics.add_argument("--db", default="oracle.db") p_metrics.add_argument("--days", type=int, default=7) p_metrics.add_argument("--adapters", action="store_true", help="Show adapter health table") p_metrics.add_argument("--publish", action="store_true", help="Show PUBLISH rate trend") p_metrics.add_argument("--scores", action="store_true", help="Show score distribution") p_metrics.add_argument("--alerts", action="store_true", help="Show current alerts") p_metrics.add_argument("--all", action="store_true", help="Show everything") p_metrics.add_argument("--json", action="store_true", help="Output as JSON") sub.add_parser("health", help="System health check") args = parser.parse_args() commands = { "ingest": cmd_ingest, "summarize": cmd_summarize, "query": cmd_query, "render": cmd_render, "archive": cmd_archive, "themes": cmd_themes, "dedup": cmd_dedup, "brief": cmd_brief, "weekly": cmd_weekly, "health": cmd_health, } if args.command and args.command in commands: commands[args.command](args) else: parser.print_help() if __name__ == "__main__": main()