Epictetus
8017ded3ba
- adapters/__init__.py: add http_get() unified retry (429/5xx only, max 2 attempts, capped exp backoff) + AdapterHTTPError carrying failure_class; SourceAdapter.last_failure_class set on failure for pipeline capture. - arxiv/github/huggingface/hackernews/reddit: route HTTP through http_get. Preserves GitHub 403 rate-limit retry and Reddit 403/429 fast-bail. - schema.sql + pipeline.py: add run_log.failure_class column; rollup most- severe class across sources (5xx>4xx>429>error>zero_fetch>ok). - pipeline.py: ENABLE RSS in ENABLED_SOURCES (was registered, disabled). - RSS smoke test surfaced 3 broken feeds (anthropic 404, googleai 404, metaai 301) — left as-is, captured in feed_failures; URL fix is separate discovery task, not guessed. Verified: full dry-run fetches all 6 sources; github live fetch OK; Reddit 429 fast-bail preserved; no import/syntax errors.
372 lines
16 KiB
Python
372 lines
16 KiB
Python
#!/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),
|
|
"failure_class": "error"}
|
|
continue
|
|
|
|
# Capture classification from the adapter (set by http_get on failure)
|
|
fc = getattr(adapter, "last_failure_class", None)
|
|
|
|
# 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,
|
|
"failure_class": fc or "ok"}
|
|
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"
|
|
|
|
# Rollup failure_class (issue #2): most severe across sources.
|
|
# Priority: 5xx > 4xx > 429 > error > zero_fetch > ok
|
|
rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0}
|
|
classes = [st.get("failure_class", "ok") for st in source_stats.values()]
|
|
if any(c in ("5xx", "4xx", "429", "error") for c in classes):
|
|
run_fc = max((c for c in classes if c in rank),
|
|
key=lambda c: rank[c])
|
|
elif zero:
|
|
run_fc = "zero_fetch"
|
|
else:
|
|
run_fc = "ok"
|
|
try:
|
|
conn.execute("""
|
|
INSERT INTO run_log (total_fetched, total_stored, sources_ok,
|
|
sources_failed, failure_class, notes)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (len(all_entries), stored, json.dumps(ok), json.dumps(failed),
|
|
run_fc, 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()
|