Epictetus
564 lines
20 KiB
Python
564 lines
20 KiB
Python
"""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 <command> [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=30)
|
|
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=30)
|
|
if t.is_alive():
|
|
error = f"timeout after 30s"
|
|
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 = sub.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()
|