commit 67c002b665444bc81e540846755a3cb3054aa2ef Author: Epictetus Date: Wed Jul 8 04:03:36 2026 +0000 Initial commit: Oracle AI research pipeline (adapters, pipeline, summarize, query) Source-controlled baseline before Phase 5 cron. Excludes oracle.db, logs/, and __pycache__ via .gitignore. Pipeline verified running clean end-to-end (run_log write confirmed before conn.close()). diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..945d696 --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Oracle generated data — not source, grows over time +oracle.db +oracle.db-journal +oracle.db-wal +oracle.db-shm + +# Logs +logs/ + +# Python bytecache +__pycache__/ +*.pyc +*.pyo + +# Env / virtualenv +.venv/ +.env diff --git a/adapters/__init__.py b/adapters/__init__.py new file mode 100644 index 0000000..317cdec --- /dev/null +++ b/adapters/__init__.py @@ -0,0 +1,17 @@ +"""Source adapters for AI Research Oracle.""" + +from abc import ABC, abstractmethod + + +class SourceAdapter(ABC): + """Base class for all ingestion adapters.""" + + @abstractmethod + def name(self) -> str: + """Source name: 'github', 'arxiv', 'reddit'.""" + pass + + @abstractmethod + def fetch(self, query: str = "", limit: int = 20) -> list[dict]: + """Return entries matching DB schema fields.""" + pass diff --git a/adapters/arxiv.py b/adapters/arxiv.py new file mode 100644 index 0000000..2002d7a --- /dev/null +++ b/adapters/arxiv.py @@ -0,0 +1,527 @@ +#!/usr/bin/env python3 +""" +arXiv adapter for AI Research Oracle. +Fetches recent AI/ML papers via arXiv API (Atom XML). +No scraping — uses the official API endpoint. + +Rate limits: 1 req/3s (be polite). arXiv enforces this aggressively. + +============================================================================ +SCORING DESIGN PRINCIPLE (do not violate when copying this to other adapters) +============================================================================ +Structural metadata (recency, author count, abstract length, category +diversity) is a WEAK signal present in EVERY paper regardless of topic. It +must NEVER dominate the final score. Content-relevance signals (AI keyword +density in the abstract, methodology-contribution detection) MUST carry the +weight. A paper that merely USES an existing AI tool as incidental methodology +scores lower than one that CONTRIBUTES a new method — detection is by verb +(creative "we propose/introduce" vs evaluative "we evaluate/using"), not by +domain. Applied-domain papers are TAGGED for filtering, NEVER penalized: for a +startup-idea oracle, novel AI applied to a vertical is desired signal. +(See _score() and _methodology_claim_score() for the implementation.) +============================================================================ +""" + +import json +import os +import re +import time +import urllib.request +import urllib.error +import xml.etree.ElementTree as ET +from datetime import datetime, timedelta, timezone +from html import unescape + +from adapters import SourceAdapter + +# arXiv API +ARXIV_API = "http://export.arxiv.org/api/query" + + +class ArxivAdapter(SourceAdapter): + """arXiv API adapter.""" + + # Default categories to scan + DEFAULT_CATEGORIES = ["cs.AI", "cs.LG", "cs.CL"] + + def __init__(self, categories=None, rate_limit=3): + """ + Args: + categories: List of arXiv categories. Default: cs.AI, cs.LG, cs.CL + rate_limit: Seconds between API calls (default 3). + """ + self.categories = categories or self.DEFAULT_CATEGORIES + self.rate_limit = rate_limit + + def name(self) -> str: + return "arxiv" + + def _parse_atom(self, xml_data: str) -> list[dict]: + """Parse arXiv Atom XML into raw paper dicts.""" + root = ET.fromstring(xml_data) + + # Handle namespaces — arXiv uses multiple + # Walk all elements and extract what we need + papers = [] + + for entry in root.iter(): + tag = entry.tag.split("}")[-1] if "}" in entry.tag else entry.tag + if tag == "entry": + paper = { + "id": "", + "title": "", + "summary": "", + "published": "", + "updated": "", + "authors": [], + "categories": [], + "comment": "", + "journal_ref": "", + "doi": "", + "link": "", + } + + for child in entry: + ctag = child.tag.split("}")[-1] + + if ctag == "id": + # arXiv ID like http://arxiv.org/abs/cs.AI/2607.00123 + paper["id"] = child.text or "" + # Also extract clean ID + if "abs/" in paper["id"]: + paper["arxiv_id"] = paper["id"].split("abs/")[-1] + + elif ctag == "title": + paper["title"] = unescape((child.text or "").strip()) + + elif ctag == "summary": + paper["summary"] = unescape((child.text or "").strip()) + + elif ctag == "published": + paper["published"] = child.text or "" + + elif ctag == "updated": + paper["updated"] = child.text or "" + + elif ctag == "author": + for ac in child: + a_tag = ac.tag.split("}")[-1] + if a_tag == "name": + paper["authors"].append(unescape((ac.text or "").strip())) + + elif ctag == "category": + term = child.get("term", "") + if term: + paper["categories"].append(term) + + elif ctag == "arxiv": + # arXiv-specific: comment, journal_ref, doi + sub_tag = child.tag.split("}")[-1] + if sub_tag == "comment": + paper["comment"] = unescape((child.text or "").strip()) + elif sub_tag == "journal_ref": + paper["journal_ref"] = unescape((child.text or "").strip()) + elif sub_tag == "doi": + paper["doi"] = unescape((child.text or "").strip()) + + elif ctag == "link": + href = child.get("href", "") + # Prefer the abstract page link + if "abs/" in href and not paper["link"]: + paper["link"] = href + elif "pdf/" in href: + paper["pdf_link"] = href + + # Clean up arxiv_id if not extracted from ID field + if "arxiv_id" not in paper and "abs/" in paper.get("id", ""): + paper["arxiv_id"] = paper["id"].split("abs/")[-1] + + if paper["title"] and paper["summary"]: + papers.append(paper) + + return papers + + def _request(self, query: str, max_results: int = 20, sort_by="submittedDate") -> list[dict]: + """Make an arXiv API request.""" + url = ( + f"{ARXIV_API}" + f"?search_query={urllib.parse.quote(query)}" + f"&sortBy={sort_by}" + f"&sortOrder=descending" + f"&max_results={max_results}" + ) + + req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"}) + + try: + with urllib.request.urlopen(req, timeout=30) as resp: + xml_data = resp.read().decode("utf-8") + return self._parse_atom(xml_data) + except urllib.error.HTTPError as e: + print(f" HTTP {e.code} for arXiv query") + return [] + except Exception as e: + print(f" arXiv request error: {e}") + return [] + + def _score(self, paper: dict, age_days: float) -> float: + """Score based on AI-methodology relevance, not structural metadata. + + arXiv has no upvotes/stars. The key signal is whether the paper is a + genuine AI/ML *contribution* (new architecture, framework, method) vs. + merely *using* an existing AI tool as incidental methodology. + + Structural signals (recency, author count, category diversity) are + weak and capped low so they cannot dominate the score — any paper has + them regardless of topic. The relevance signals (methodology claim + + AI keyword density) carry the weight. + + Applied-domain papers (healthcare, finance, biology) are TAGGED for + filtering but NOT penalized — for a startup-idea oracle, "novel AI + technique applied to a vertical" is exactly the signal we want. + """ + # --- Structural signals (weak, capped low) --- + # Recency: newer = slightly higher, but max 2.0 (was 5.0) + try: + pub_str = paper.get("published", "") + pub_dt = datetime.fromisoformat(pub_str.replace("Z", "+00:00")) + now = datetime.now(timezone.utc) + age_hours = (now - pub_dt).total_seconds() / 3600 + recency = max(0, 2.0 - age_hours / 48.0) + except (ValueError, TypeError): + recency = max(0, 2.0 - age_days * 0.1) + + # Author count: collaborative work is a weak signal, max 0.5 (was 2.0) + author_count = len(paper.get("authors", [])) + author_bonus = min(author_count * 0.05, 0.5) + + # Category diversity: cross-domain is mildly interesting, max 0.3 + cats = paper.get("categories", []) + diversity_bonus = min(len(cats) * 0.1, 0.3) + + # Journal/DOI: published = validated, small bonus + journal_bonus = 0.5 if paper.get("journal_ref") else 0.0 + doi_bonus = 0.2 if paper.get("doi") else 0.0 + + # Comment field (page count etc.): tiny bonus + comment = paper.get("comment", "") + comment_bonus = min(len(comment) / 300.0, 0.5) if comment else 0.0 + + # --- Relevance signals (carry the weight) --- + title_lower = paper.get("title", "").lower() + abstract = paper.get("summary", "") + abstract_lower = abstract.lower() + + # AI keyword density in abstract (not just title) + ai_keywords = [ + "agent", "llm", "large language", "gpt", "transformer", + "neural", "deep learning", "machine learning", "reinforcement", + "diffusion", "multimodal", "embedding", "attention", "rag", + "retrieval", "fine-tun", "pretrain", "self-supervised", + "foundation model", "reasoning", "alignment", "policy gradient", + "graph neural", "vision-language", "vla", "vlm", + ] + ai_kw_hits = sum(1 for kw in ai_keywords if kw in abstract_lower) + # Density matters more than single mention; saturate at ~6 hits + ai_keyword_bonus = min(ai_kw_hits * 0.35, 2.0) + + # Title keyword bonus (small) + title_kw_list = ["agent", "llm", "reasoning", "verif", "multimodal", + "embodied", "robot", "alignment", "autonomous"] + title_kw_hits = sum(1 for kw in title_kw_list if kw in title_lower) + title_kw_bonus = min(title_kw_hits * 0.2, 0.5) + + # AI methodology claim: does the abstract propose/invent something? + methodology_bonus = self._methodology_claim_score(abstract_lower) + + # --- Domain tag (no penalty) --- + combined_text = f"{title_lower} {abstract_lower}" + applied_domains = { + "applied:healthcare": ["gastric", "biopsy", "pathology", "clinical", + "patient", "diagnosis", "medical imaging", "h. pylori", + "cancer screening", "oncology", "neurology", "cardiology"], + "applied:finance": ["stock market", "trading", "portfolio", "forex", + "cryptocurrency", "fintech", "credit scoring", "fraud detection"], + "applied:biology": ["protein folding", "gene expression", "genome", + "molecular", "bioinformatics", "cell type", "organism"], + } + primary_domain = None + for domain, domain_keywords in applied_domains.items(): + hits = sum(1 for kw in domain_keywords if kw in combined_text) + if hits >= 2: # need at least 2 domain keywords to trigger + primary_domain = domain + break + if primary_domain: + paper["_applied_domain"] = primary_domain + + # Total: structural (capped ~3.5) + relevance (capped ~5.0) + structural = recency + author_bonus + diversity_bonus + journal_bonus + doi_bonus + comment_bonus + relevance = ai_keyword_bonus + title_kw_bonus + methodology_bonus + score = structural + relevance + + return min(round(score, 2), 10.0) + + def _methodology_claim_score(self, abstract_lower: str) -> float: + """Detect whether the abstract proposes a NEW AI method vs. using an existing one. + + The core distinction is the VERB: creative verbs (propose/introduce/ + develop/design) signal a contribution; evaluative verbs (evaluate/ + conduct/retrospective/using/based on) signal applying an existing tool. + + Check weak (evaluative) patterns FIRST — if the paper is an evaluation + or application of an existing system, it gets 0.5 even if it also says + 'we present'. Only if no weak pattern matches do we look for a genuine + contribution claim. + + Returns 0.0 (no claim / unclear), 0.5 (uses existing tool), + 2.0 (novel method), or 2.5 (explicit proposal of new AI system). + """ + # Weak FIRST: evaluative / application language + weak_markers = ["we evaluate", "we conducted", "we conduct", "retrospective", + "case study", "pilot study", "using", "we leverage", + "we applied", "based on", "we report a", "we present a retrospective", + "an evaluation of", "we benchmark", "empirical study"] + for wm in weak_markers: + if wm in abstract_lower: + return 0.5 + + # Strong: genuinely creative verbs + AI noun within 60 chars + proposal_verbs = ["we propose", "we introduce", "we develop", "we design", + "we formulate", "we construct", "we build", "we present a novel", + "we present a new"] + ai_nouns = ["model", "architecture", "framework", "method", "approach", + "agent", "system", "network", "algorithm", "pipeline", + "llm", "transformer", "policy", "graph", "orchestration", + "harness", "memory", "solver", "planner"] + for verb in proposal_verbs: + idx = abstract_lower.find(verb) + if idx != -1: + window = abstract_lower[idx:idx + 60] + if any(noun in window for noun in ai_nouns): + return 2.5 + + # Strong-ish: 'novel'/'new' + AI architecture/framework/model within 40 chars + for marker in ["novel", "new"]: + idx = abstract_lower.find(marker) + scan = 0 + while idx != -1 and scan < 5: + window = abstract_lower[idx:idx + 40] + if any(noun in window for noun in + ["architecture", "framework", "model", "method", + "approach", "agent", "system", "harness"]): + return 2.0 + idx = abstract_lower.find(marker, idx + 1) + scan += 1 + + return 0.0 + + def _tags(self, paper: dict) -> list: + """Generate category tags from arXiv metadata.""" + tags = ["arxiv"] + + # Add arXiv categories + for cat in paper.get("categories", [])[:5]: + tags.append(f"cat:{cat}") + + # Primary signal categories + cats_lower = [c.lower() for c in paper.get("categories", [])] + if "cs.ai" in cats_lower: + tags.append("ai-general") + if "cs.lg" in cats_lower: + tags.append("machine-learning") + if "cs.cl" in cats_lower: + tags.append("nlp") + if "cs.cv" in cats_lower: + tags.append("computer-vision") + if "cs.ro" in cats_lower: + tags.append("robotics") + if "cs.se" in cats_lower: + tags.append("software-engineering") + + # Check if it's a survey/tutorial + title_lower = paper.get("title", "").lower() + summary_lower = paper.get("summary", "").lower() + if any(kw in summary_lower for kw in ["survey", "tutorial", "overview of", "review of"]): + tags.append("survey") + + # Check for preprint vs published + if paper.get("journal_ref"): + tags.append("published") + else: + tags.append("preprint") + + # Applied-domain tag + if paper.get("_applied_domain"): + tags.append(paper["_applied_domain"]) + + return tags + + def fetch(self, query: str = "", limit: int = 20) -> list[dict]: + """ + Fetch papers from arXiv. + + If query is empty, fetch recent papers from configured categories. + If query is provided, search for it. + """ + now = datetime.now(timezone.utc) + all_papers = [] + + if query: + print(f" Searching arXiv: '{query}'") + papers = self._request(query, max_results=limit, sort_by="submittedDate") + all_papers.extend(papers) + else: + # Fetch from each configured category + for cat in self.categories: + q = f"cat:{cat}" + cat_papers = self._request(q, max_results=limit, sort_by="submittedDate") + all_papers.extend(cat_papers) + time.sleep(self.rate_limit) + + # Deduplicate by arxiv_id + seen = set() + unique = [] + for p in all_papers: + pid = p.get("arxiv_id", p.get("id", "")) + if pid and pid not in seen: + seen.add(pid) + unique.append(p) + all_papers = unique + + # Sort by published date (newest first), take top limit + all_papers.sort( + key=lambda p: p.get("published", ""), + reverse=True, + ) + all_papers = all_papers[:limit] + + entries = [] + for paper in all_papers: + # Calculate age + published = paper.get("published", "") + try: + pub_dt = datetime.fromisoformat(published.replace("Z", "+00:00")) + age_days = (now - pub_dt).days + except (ValueError, TypeError): + age_days = 0 + + score = self._score(paper, age_days) + tags = self._tags(paper) + + # Source ID: arxiv_id + source_id = paper.get("arxiv_id", paper.get("id", "")).split("/")[-1] + + # URL: abstract page + url = paper.get("link", "") + if not url and paper.get("arxiv_id"): + url = f"https://arxiv.org/abs/{paper['arxiv_id']}" + + # Title (clean — remove trailing category markers) + title = re.sub(r"\s*\([A-Za-z0-9., ]*\)\s*$", "", paper.get("title", "")).strip() + + # Extracted text: summary (abstract) — already clean + extracted_text = paper.get("summary", "") + + # Structured metadata + raw_meta = { + "arxiv_id": paper.get("arxiv_id", ""), + "authors": paper.get("authors", []), + "author_count": len(paper.get("authors", [])), + "categories": paper.get("categories", []), + "published": paper.get("published", ""), + "updated": paper.get("updated", ""), + "comment": paper.get("comment", ""), + "journal_ref": paper.get("journal_ref", ""), + "doi": paper.get("doi", ""), + "pdf_link": paper.get("pdf_link", ""), + "age_days": age_days, + "abstract_length": len(extracted_text), + "score_type": "estimated", # arXiv has no upvotes + } + # Store applied-domain info if detected + if paper.get("_applied_domain"): + raw_meta["applied_domain"] = paper["_applied_domain"] + + now_str = now.strftime("%Y-%m-%dT%H:%M:%SZ") + entries.append({ + "source": "arxiv", + "source_id": source_id, + "url": url, + "title": title, + "extracted_text": extracted_text, + "summary": None, + "category_tags": json.dumps(tags), + "signal_score": score, + "raw_metadata": json.dumps(raw_meta), + "first_seen": now_str, + "last_updated": now_str, + }) + + return entries + + +if __name__ == "__main__": + import argparse + import sqlite3 + import urllib.parse + + parser = argparse.ArgumentParser(description="arXiv adapter for AI Research Oracle") + parser.add_argument("--query", default="", help="Search query (empty = recent categories)") + parser.add_argument("--limit", type=int, default=20, help="Max entries") + parser.add_argument("--db", default=os.path.join(os.path.dirname(__file__), "..", "oracle.db"), help="SQLite DB") + parser.add_argument("--schema", default=os.path.join(os.path.dirname(__file__), "..", "schema.sql"), help="Schema file") + parser.add_argument("--dry-run", action="store_true", help="Don't store in DB") + args = parser.parse_args() + + print(f"=== arXiv Adapter ===") + print(f" Query: {args.query or '(recent AI categories)'}") + print(f" Limit: {args.limit}") + print() + + adapter = ArxivAdapter() + entries = adapter.fetch(query=args.query, limit=args.limit) + + print(f" Fetched {len(entries)} entries") + + if not args.dry_run: + conn = sqlite3.connect(args.db) + if os.path.exists(args.schema): + with open(args.schema) as f: + conn.executescript(f.read()) + conn.commit() + + 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"], entry["summary"], + 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: {e}") + + conn.commit() + conn.close() + print(f" Stored {stored} entries") + + # Print top 5 + print(f"\n Top entries:") + for i, e in enumerate(entries[:5]): + meta = json.loads(e["raw_metadata"]) if isinstance(e["raw_metadata"], str) else e["raw_metadata"] + authors = meta.get("authors", []) + author_str = f"{authors[0]} et al." if len(authors) > 2 else ", ".join(authors[:2]) + print(f" [{i+1}] score={e['signal_score']:.2f} authors={author_str}") + print(f" {e['title'][:90]}") + print(f" {e['url']}") + print(f" abstract={meta.get('abstract_length', 0)}ch") + + print(f"\n Done.") diff --git a/adapters/github.py b/adapters/github.py new file mode 100644 index 0000000..a3cda52 --- /dev/null +++ b/adapters/github.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +""" +GitHub adapter for AI Research Oracle. +Fetches repos via GitHub REST API — no scraping, no trending page. +Definition of "trending": created or pushed in last N days, sorted by stars. + +Rate limits: 60 req/hr unauthenticated. +Strategy: cache results, deduplicate, budget calls. +""" + +import json +import os +import re +import time +import urllib.request +import urllib.error +import urllib.parse +from datetime import datetime, timedelta, timezone + +from adapters import SourceAdapter + + +class GitHubAdapter(SourceAdapter): + """GitHub REST API adapter.""" + + BASE = "https://api.github.com" + + def __init__(self, token: str = None): + """Initialize with optional read-only token (5000 req/hr vs 60).""" + self.token = token or os.environ.get("GITHUB_TOKEN", "") + self.cache = {} + + def name(self) -> str: + return "github" + + def _headers(self): + headers = { + "Accept": "application/vnd.github.v3+json", + "User-Agent": "ai-oracle/0.1", + } + if self.token: + headers["Authorization"] = f"token {self.token}" + return headers + + def _request(self, url: str, max_retries: int = 2) -> dict | list | None: + """Make a GET request with retry on 403 (rate limit).""" + req = urllib.request.Request(url, headers=self._headers()) + + for attempt in range(max_retries + 1): + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read().decode("utf-8")) + + # Check rate limit headers + remaining = int(resp.headers.get("X-RateLimit-Remaining", 0)) + if remaining <= 5: + print(f" ⚠ Rate limit low ({remaining} remaining), stopping") + break + + return data + except urllib.error.HTTPError as e: + if e.code == 403: + # Rate limited — reset time is in headers + reset = int(e.headers.get("X-RateLimit-Reset", 0)) + if reset: + wait = max(reset - int(time.time()), 0) + 1 + print(f" ⚠ Rate limited, wait {wait}s") + else: + wait = 30 * (attempt + 1) + print(f" 403 on attempt {attempt + 1}, retry in {wait}s") + time.sleep(min(wait, 300)) # cap at 5 min + continue + print(f" HTTP {e.code} for {url}") + return None + except Exception as e: + print(f" Request error: {e}") + return None + + return None + + def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list: + """Search repositories via GitHub API.""" + url = f"{self.BASE}/search/repositories?q={urllib.parse.quote(query)}&sort={sort}&order={order}&per_page={per_page}" + result = self._request(url) + if isinstance(result, dict) and "items" in result: + return result["items"] + return [] + + def _get_readme(self, repo_url: str) -> str: + """Fetch README content for a repo.""" + if repo_url in self.cache: + return self.cache[repo_url] + + result = self._request(repo_url, max_retries=1) + if result and isinstance(result, dict): + import base64 + content = result.get("content", "") + encoding = result.get("encoding", "base64") + if encoding == "base64" and content: + try: + decoded = base64.b64decode(content).decode("utf-8", errors="replace") + # Truncate to 8000 chars to keep context manageable + self.cache[repo_url] = decoded[:8000] + return decoded[:8000] + except Exception: + pass + return "" + + def _score(self, repo: dict, age_days: float) -> float: + """Score: stars weighted by recency. Higher = better signal.""" + stars = repo.get("stargazers_count", 0) + # Normalize: log scale on stars, decay by age + import math + star_score = min(math.log1p(stars) / 2.0, 10.0) # log(1000) ≈ 6.9 → ~3.5 + # Recency bonus: newer repos get a slight boost + recency_bonus = max(0, 1.0 - age_days / 30.0) * 1.5 # up to +1.5 for very recent + return min(star_score + recency_bonus, 10.0) + + def _tags(self, repo: dict) -> list: + """Generate category tags from repo metadata.""" + tags = ["github"] + lang = repo.get("language", "") + if lang: + tags.append(f"lang:{lang.lower()}") + topics = repo.get("topics", []) + for t in topics[:5]: + tags.append(f"topic:{t}") + # Check description for AI/ML signals + desc = (repo.get("description") or "").lower() + if any(kw in desc for kw in ["agent", "agents"]): + tags.append("agents") + if any(kw in desc for kw in ["llm", "large language"]): + tags.append("llm") + if any(kw in desc for kw in ["rag", "retrieval"]): + tags.append("rag") + return tags + + def fetch(self, query: str = "", limit: int = 20) -> list[dict]: + """ + Fetch repos from GitHub. + + If query is empty, fetch "trending" = recent high-star AI repos. + If query is provided, search for it. + """ + now = datetime.now(timezone.utc) + + if query: + print(f" Searching: '{query}'") + repos = self._search_repos(query, sort="stars", per_page=min(limit * 2, 100)) + else: + # "Trending" = AI repos created in last 14 days, sorted by stars + cutoff = (now - timedelta(days=14)).strftime("%Y-%m-%d") + # Two queries to get breadth + repos = [] + for q in ["ai agents created:>=2026-06-01", "llm inference created:>=2026-06-01"]: + batch = self._search_repos(q, sort="stars", per_page=50) + repos.extend(batch) + time.sleep(1) # polite spacing + + # Deduplicate by full_name + seen = set() + unique = [] + for r in repos: + fn = r.get("full_name", "") + if fn not in seen: + seen.add(fn) + unique.append(r) + repos = unique + + # Sort by stars descending, take top limit + repos.sort(key=lambda r: r.get("stargazers_count", 0), reverse=True) + repos = repos[:limit] + + entries = [] + readme_budget = min(10, limit) # Only fetch README for top 10 to stay under rate limit + for idx, repo in enumerate(repos): + # Calculate age + created = repo.get("created_at") + if created: + try: + created_dt = datetime.fromisoformat(created.replace("Z", "+00:00")) + age_days = (now - created_dt).days + except (ValueError, TypeError): + age_days = 0 + else: + age_days = 0 + + score = self._score(repo, age_days) + tags = self._tags(repo) + + # Source ID: repo full_name + source_id = repo.get("full_name", "").replace("/", "__") + + # URL + url = repo.get("html_url", "") + + # Title: repo name with description + name = repo.get("name", "") + desc = repo.get("description", "") or "" + if desc: + title = f"{name}: {desc[:100]}" + else: + title = name + + # README as extracted_text + # Search API doesn't return readme_url — construct it from owner/repo + owner = repo.get("owner", {}).get("login", "") + repo_name = repo.get("name", "") + readme_text = "" + if owner and repo_name and idx < readme_budget: + readme_text = self._get_readme(f"https://api.github.com/repos/{owner}/{repo_name}/readme") + time.sleep(0.5) # polite spacing between README fetches + + # Structured metadata + raw_meta = { + "full_name": repo.get("full_name", ""), + "owner": repo.get("owner", {}).get("login", ""), + "stars": repo.get("stargazers_count", 0), + "forks": repo.get("forks_count", 0), + "open_issues": repo.get("open_issues_count", 0), + "language": repo.get("language", ""), + "topics": repo.get("topics", []), + "created_at": repo.get("created_at", ""), + "updated_at": repo.get("updated_at", ""), + "pushed_at": repo.get("pushed_at", ""), + "age_days": age_days, + "readme_length": len(readme_text), + "score_type": "actual", # based on real star counts + } + + now_str = now.strftime("%Y-%m-%dT%H:%M:%SZ") + entries.append({ + "source": "github", + "source_id": source_id, + "url": url, + "title": title, + "extracted_text": readme_text, + "summary": None, # LLM later + "category_tags": json.dumps(tags), + "signal_score": round(score, 2), + "raw_metadata": json.dumps(raw_meta), + "first_seen": now_str, + "last_updated": now_str, + }) + + return entries + + +if __name__ == "__main__": + import argparse + import sqlite3 + import os + + parser = argparse.ArgumentParser(description="GitHub adapter for AI Research Oracle") + parser.add_argument("--query", default="", help="Search query (empty = trending AI)") + parser.add_argument("--limit", type=int, default=20, help="Max entries") + parser.add_argument("--db", default=os.path.join(os.path.dirname(__file__), "..", "oracle.db"), help="SQLite DB") + parser.add_argument("--schema", default=os.path.join(os.path.dirname(__file__), "..", "schema.sql"), help="Schema file") + parser.add_argument("--dry-run", action="store_true", help="Don't store in DB") + args = parser.parse_args() + + print(f"=== GitHub Adapter ===") + print(f" Query: {args.query or '(trending AI)'}") + print(f" Limit: {args.limit}") + print() + + adapter = GitHubAdapter() + entries = adapter.fetch(query=args.query, limit=args.limit) + + print(f" Fetched {len(entries)} entries") + + if not args.dry_run: + conn = sqlite3.connect(args.db) + if os.path.exists(args.schema): + with open(args.schema) as f: + conn.executescript(f.read()) + conn.commit() + + 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"], entry["summary"], + 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: {e}") + + conn.commit() + conn.close() + print(f" Stored {stored} entries") + + # Print top 5 + print(f"\n Top entries:") + for i, e in enumerate(entries[:5]): + meta = json.loads(e["raw_metadata"]) if isinstance(e["raw_metadata"], str) else e["raw_metadata"] + print(f" [{i+1}] score={e['signal_score']:.2f} stars={meta.get('stars', '?')}") + print(f" {e['title'][:90]}") + print(f" {e['url']}") + print(f" text={len(e.get('extracted_text', ''))}ch") + + print(f"\n Done.") diff --git a/adapters/reddit.py b/adapters/reddit.py new file mode 100644 index 0000000..d40f82b --- /dev/null +++ b/adapters/reddit.py @@ -0,0 +1,535 @@ +#!/usr/bin/env python3 +""" +Reddit adapter v2 for AI Research Oracle. +Uses Reddit RSS feeds (Atom XML) with proper filtering and scoring. + +Improvements over v1 (proof of concept): +- Filters AutoModerator posts and sticky threads +- Proper User-Agent header +- Attempts old.reddit.com .json for real upvote/comment data +- Falls back to RSS with estimated scoring if JSON blocked +- Rate limit awareness with retry logic + +Rate limits: Reddit aggressively 429s unauthenticated requests. +Strategy: 3s spacing between subreddits, retry with backoff. +""" + +import json +import os +import re +import time +import urllib.request +import urllib.error +import xml.etree.ElementTree as ET +from datetime import datetime, timezone +from html import unescape + +from adapters import SourceAdapter + + +class RedditAdapter(SourceAdapter): + """Reddit RSS + JSON adapter.""" + + # Default subreddits for AI content + DEFAULT_SUBREDDITS = [ + "MachineLearning", "artificial", "LocalLLaMA", "Startups", + ] + + # Posts to filter out (sticky threads, recurring mod posts) + FILTER_TITLES = { + "self-promotion thread", + "monthly who's hiring", + "weekly self-promotion", + "discussion thread", + "show and tell", + "weekly show", + } + + FILTER_AUTHORS = { + "automod", + "automoderator", + } + + def __init__(self, subreddits=None, rate_limit=3, user_agent=None): + """ + Args: + subreddits: List of subreddit names. + rate_limit: Seconds between subreddit requests. + user_agent: Custom User-Agent header. + """ + self.subreddits = subreddits or self.DEFAULT_SUBREDDITS + self.rate_limit = rate_limit + self.user_agent = user_agent or "python:ai-oracle:v0.1 (by tony_tech)" + + def name(self) -> str: + return "reddit" + + def _clean_html(self, html: str) -> str: + """Extract readable text from Reddit's HTML content.""" + if not html: + return "" + text = re.sub(r"", "", html, flags=re.DOTALL) + text = re.sub(r"]*>", "\n", text) + text = re.sub(r"", "\n", text) + text = re.sub(r"", "\n", text, flags=re.I) + text = re.sub(r"]*>", "\n", text) + text = re.sub(r"

", "\n", text) + text = re.sub(r"]*href=\"([^\"]+)\"[^>]*>([^<]*)", r"\2", text) + text = re.sub(r"<[^>]+>", "", text) + text = unescape(text) + text = re.sub(r"\n\s*\n+", "\n\n", text) + return text.strip() + + def _is_sticky(self, entry: dict) -> bool: + """Check if a post is a sticky/mod post that should be filtered.""" + title_lower = entry.get("title", "").lower().strip() + author_lower = entry.get("author", "").lower().strip() + + # Filter by author + if author_lower in self.FILTER_AUTHORS: + return True + + # Filter by title patterns + for pattern in self.FILTER_TITLES: + if pattern in title_lower: + return True + + # Filter very short posts (likely link-only with no content) + if len(entry.get("content", "")) < 20: + return True + + return False + + def _fetch_rss(self, subreddit: str) -> list[dict]: + """Fetch RSS feed for a subreddit.""" + url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" + req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) + + for attempt in range(3): + try: + with urllib.request.urlopen(req, timeout=15) as resp: + xml_data = resp.read().decode("utf-8") + break + except urllib.error.HTTPError as e: + if e.code == 429: + wait = 5 * (attempt + 1) + time.sleep(wait) + continue + print(f" RSS HTTP {e.code} for r/{subreddit}") + return [] + except Exception as e: + print(f" RSS error r/{subreddit}: {e}") + return [] + else: + print(f" r/{subreddit}: still rate limited, skip") + return [] + + # Parse Atom XML + entries = [] + root = ET.fromstring(xml_data) + + for entry_el in root.iter(): + tag = entry_el.tag.split("}")[-1] if "}" in entry_el.tag else entry_el.tag + if tag != "entry": + continue + + data = {"title": "", "url": "", "author": "", "content": "", "published": "", "id": ""} + + for child in entry_el: + ctag = child.tag.split("}")[-1] + if ctag == "title": + data["title"] = unescape((child.text or "").strip()) + elif ctag == "link": + data["url"] = child.get("href", "") + elif ctag == "author": + for ac in child: + if ac.tag.split("}")[-1] == "name": + data["author"] = unescape((ac.text or "").strip()) + elif ctag == "content": + data["content"] = child.text or "" + elif ctag == "published": + data["published"] = child.text or "" + elif ctag == "id": + data["id"] = child.text or "" + + if data["title"] and data["url"]: + entries.append(data) + + return entries + + def _try_json(self, subreddit: str, sort: str = "hot") -> list[dict]: + """Try to fetch JSON data from old.reddit.com for real scores.""" + url = f"https://old.reddit.com/r/{subreddit}/{sort}.json?limit=25" + req = urllib.request.Request(url, headers={ + "User-Agent": self.user_agent, + "Accept": "application/json", + }) + + for attempt in range(2): + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read().decode("utf-8")) + + posts = [] + if isinstance(data, dict) and "data" in data: + for child in data["data"].get("children", []): + d = child.get("data", {}) + if d.get("title"): + posts.append({ + "title": d.get("title", ""), + "url": d.get("url", ""), + "author": d.get("author", ""), + "selftext": d.get("selftext", ""), + "score": d.get("score", 0), + "num_comments": d.get("num_comments", 0), + "ups": d.get("ups", 0), + "id": d.get("id", ""), + "created_utc": d.get("created_utc", 0), + "permalink": d.get("permalink", ""), + "link_flair_text": d.get("link_flair_text", ""), + "subreddit": subreddit, + "source": "json", + }) + return posts + except urllib.error.HTTPError as e: + if e.code in (429, 403, 404): + return [] # JSON endpoint blocked, fall back to RSS + if attempt < 1: + time.sleep(5) + continue + return [] + except Exception: + return [] + return [] + + def _score(self, entry: dict, source: str) -> float: + """Score based on available signals. + + If source is 'json', use real upvote/comment counts. + If source is 'rss', use heuristics (content length, flair, etc.). + """ + if source == "json" and entry.get("score", 0) > 0: + import math + # Real upvote-based scoring + score = entry.get("score", 0) + comments = entry.get("num_comments", 0) + + # Log scale on upvotes, bonus for engagement + star_score = min(math.log1p(score) / 2.0, 8.0) + engagement = min(comments / 50.0, 2.0) + + return min(round(star_score + engagement, 2), 10.0) + else: + # RSS heuristic — no real scores available + content = entry.get("content", "") + title = entry.get("title", "") + + # Base score + base = 3.0 + + # Content length bonus (substantive posts get higher scores) + content_len = len(self._clean_html(content)) + if content_len > 5000: + base += 2.0 + elif content_len > 2000: + base += 1.5 + elif content_len > 1000: + base += 1.0 + elif content_len > 500: + base += 0.6 + elif content_len > 200: + base += 0.3 + elif content_len > 50: + base += 0.1 + + # Post type from title markers + if " [R]" in title or " [r]" in title: # Research + base += 1.5 + elif " [P]" in title or " [p]" in title: # Project + base += 1.2 + elif " [N]" in title or " [n]" in title: # News + base += 0.8 + elif " [D]" in title or " [d]" in title: # Discussion + base += 0.5 + + # Flair bonus + flair = entry.get("link_flair_text", "").lower() + if "research" in flair: + base += 1.0 + elif "project" in flair: + base += 0.8 + + return min(round(base, 2), 10.0) + + def _tags(self, entry: dict, source: str) -> list: + """Generate category tags from Reddit metadata.""" + tags = ["reddit"] + + subreddit = entry.get("subreddit", "").lower() + if "machinelearning" in subreddit: + tags.append("machine-learning") + elif "artificial" in subreddit: + tags.append("ai-general") + elif "localllama" in subreddit: + tags.append("local-llm") + elif "startups" in subreddit: + tags.append("startups") + + title = entry.get("title", "") + if " [P]" in title or " [p]" in title: + tags.append("project") + elif " [R]" in title or " [r]" in title: + tags.append("research") + elif " [D]" in title or " [d]" in title: + tags.append("discussion") + elif " [N]" in title or " [n]" in title: + tags.append("news") + else: + tags.append("general") + + # Check for self-referential virality signal + content = entry.get("content", "").lower() + title_lower = title.lower() + if any(kw in title_lower or kw in content for kw in [ + "stars", "viral", "trending", "blow up", "trending github", + ]): + tags.append("meta:virality") + + return tags + + def _fetch_rss(self, subreddit: str) -> list[dict]: + """Fetch RSS feed for a subreddit.""" + url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" + req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) + + for attempt in range(3): + try: + with urllib.request.urlopen(req, timeout=15) as resp: + xml_data = resp.read().decode("utf-8") + break + except urllib.error.HTTPError as e: + if e.code == 429: + wait = 5 * (attempt + 1) + time.sleep(wait) + continue + print(f" RSS HTTP {e.code} for r/{subreddit}") + return [] + except Exception as e: + print(f" RSS error r/{subreddit}: {e}") + return [] + else: + print(f" r/{subreddit}: still rate limited, skip") + return [] + + # Parse Atom XML + entries = [] + root = ET.fromstring(xml_data) + + for entry_el in root.iter(): + tag = entry_el.tag.split("}")[-1] if "}" in entry_el.tag else entry_el.tag + if tag != "entry": + continue + + data = {"title": "", "url": "", "author": "", "content": "", "published": "", "id": ""} + + for child in entry_el: + ctag = child.tag.split("}")[-1] + if ctag == "title": + data["title"] = unescape((child.text or "").strip()) + elif ctag == "link": + data["url"] = child.get("href", "") + elif ctag == "author": + for ac in child: + if ac.tag.split("}")[-1] == "name": + data["author"] = unescape((ac.text or "").strip()) + elif ctag == "content": + data["content"] = child.text or "" + elif ctag == "published": + data["published"] = child.text or "" + elif ctag == "id": + data["id"] = child.text or "" + + if data["title"] and data["url"]: + entries.append(data) + + return entries + + def fetch(self, query: str = "", limit: int = 20) -> list[dict]: + """ + Fetch Reddit posts. + + Tries JSON first (real scores), falls back to RSS (heuristic scoring). + Filters AutoModerator and sticky posts. + """ + now = datetime.now(timezone.utc) + all_entries = [] + seen_ids = set() + json_worked = False + + # Try JSON first + for sub in self.subreddits[:2]: # Try JSON on first 2 subreddits + posts = self._try_json(sub) + if posts: + json_worked = True + for p in posts: + if p["id"] not in seen_ids: + seen_ids.add(p["id"]) + all_entries.append(p) + time.sleep(self.rate_limit) + + # If JSON didn't work, fall back to RSS for all subreddits + if not json_worked: + print(" JSON endpoints blocked, using RSS fallback") + for sub in self.subreddits: + entries = self._fetch_rss(sub) + for e in entries: + # Convert RSS format to unified format + entry = { + "title": e["title"], + "url": e["url"], + "author": e["author"], + "content": e["content"], + "published": e["published"], + "id": e["id"].replace("t3_", ""), + "score": 0, + "num_comments": 0, + "ups": 0, + "selftext": "", + "created_utc": 0, + "permalink": "", + "link_flair_text": "", + "subreddit": sub, + "source": "rss", + } + if entry["id"] not in seen_ids: + seen_ids.add(entry["id"]) + all_entries.append(entry) + time.sleep(self.rate_limit) + + # Filter sticky/mod posts + filtered = [] + for entry in all_entries: + if not self._is_sticky(entry): + filtered.append(entry) + + print(f" Filtered: {len(all_entries)} → {len(filtered)} (removed {len(all_entries) - len(filtered)} sticky/mod)") + + # Sort by score, take top limit + for entry in filtered: + entry["_score"] = self._score(entry, entry.get("source", "rss")) + filtered.sort(key=lambda e: e.get("_score", 0), reverse=True) + filtered = filtered[:limit] + + # Convert to DB format + entries = [] + for entry in filtered: + source_id = entry.get("id", "") or entry.get("url", "").split("/")[-1] + score = entry.pop("_score", 0) + + # Clean title + title = entry.get("title", "") + + # Extracted text: selftext (JSON) or content (RSS) + if entry.get("source") == "json": + extracted_text = entry.get("selftext", "") + else: + extracted_text = self._clean_html(entry.get("content", "")) + + tags = self._tags(entry, entry.get("source", "rss")) + + # Structured metadata + raw_meta = { + "subreddit": entry.get("subreddit", ""), + "author": entry.get("author", ""), + "published": entry.get("published", ""), + "score": entry.get("score", 0), + "num_comments": entry.get("num_comments", 0), + "ups": entry.get("ups", 0), + "link_flair_text": entry.get("link_flair_text", ""), + "source_type": entry.get("source", "rss"), # json or rss + "content_length": len(extracted_text), + "score_type": "actual" if entry.get("source") == "json" and entry.get("score", 0) > 0 else "estimated", + } + + now_str = now.strftime("%Y-%m-%dT%H:%M:%SZ") + entries.append({ + "source": "reddit", + "source_id": source_id, + "url": entry.get("url", ""), + "title": title, + "extracted_text": extracted_text, + "summary": None, + "category_tags": json.dumps(tags), + "signal_score": score, + "raw_metadata": json.dumps(raw_meta), + "first_seen": now_str, + "last_updated": now_str, + }) + + return entries + + +if __name__ == "__main__": + import argparse + import sqlite3 + + parser = argparse.ArgumentParser(description="Reddit adapter v2 for AI Research Oracle") + parser.add_argument("--limit", type=int, default=20, help="Max entries") + parser.add_argument("--db", default=os.path.join(os.path.dirname(__file__), "..", "oracle.db"), help="SQLite DB") + parser.add_argument("--schema", default=os.path.join(os.path.dirname(__file__), "..", "schema.sql"), help="Schema file") + parser.add_argument("--dry-run", action="store_true", help="Don't store in DB") + args = parser.parse_args() + + print(f"=== Reddit Adapter v2 ===") + print(f" Limit: {args.limit}") + print() + + adapter = RedditAdapter() + entries = adapter.fetch(limit=args.limit) + + print(f"\n Fetched {len(entries)} entries") + + if not args.dry_run: + conn = sqlite3.connect(args.db) + if os.path.exists(args.schema): + with open(args.schema) as f: + conn.executescript(f.read()) + conn.commit() + + 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"], entry["summary"], + 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: {e}") + + conn.commit() + conn.close() + print(f" Stored {stored} entries") + + # Print top 5 + print(f"\n Top entries:") + for i, e in enumerate(entries[:5]): + meta = json.loads(e["raw_metadata"]) if isinstance(e["raw_metadata"], str) else e["raw_metadata"] + src = meta.get("source_type", "?") + ups = meta.get("score", "?") + comments = meta.get("num_comments", "?") + score_type = meta.get("score_type", "?") + print(f" [{i+1}] score={e['signal_score']:.2f} ({score_type}) ups={ups} comments={comments} src={src}") + print(f" {e['title'][:90]}") + print(f" r/{meta.get('subreddit', '?')} by {meta.get('author', '?')}") + print(f" text={meta.get('content_length', 0)}ch") + + print(f"\n Done.") diff --git a/pipeline.py b/pipeline.py new file mode 100644 index 0000000..05ecbaa --- /dev/null +++ b/pipeline.py @@ -0,0 +1,283 @@ +#!/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(), +} + +# Default enabled sources +ENABLED_SOURCES = ["github", "arxiv", "reddit"] + + +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 == "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}") + + 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")] + notes = "; ".join(f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")) 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() diff --git a/query.py b/query.py new file mode 100644 index 0000000..fb18230 --- /dev/null +++ b/query.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python3 +""" +AI Research Oracle — Query & Snapshot CLI. + +Query the unified database and produce Claude-ready snapshots. +Supports filtering by source, score, confidence, tag, and date range. + +Usage: + python3 query.py top 10 # top 10 across all sources + python3 query.py by-source github 5 # top 5 from GitHub + python3 query.py by-tag "agent" # entries tagged with "agent" + python3 query.py snapshot # full snapshot for Claude relay + python3 query.py search "world model" # keyword search in titles/summaries + python3 query.py recent --hours 24 # entries from last 24h + python3 query.py stats # database statistics +""" + +import argparse +import json +import os +import sqlite3 +import sys +from datetime import datetime, timezone, timedelta +from urllib.parse import quote_plus + +sys.path.insert(0, os.path.dirname(__file__)) + +DB_PATH = os.path.join(os.path.dirname(__file__), "oracle.db") + + +def get_db(): + """Open database connection.""" + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + return conn + + +def format_entry(row: sqlite3.Row, rank: int = 0) -> dict: + """Format a DB row into a clean dict for output.""" + summary = json.loads(row["summary"]) if row["summary"] else {} + meta = json.loads(row["raw_metadata"]) if row["raw_metadata"] else {} + tags = json.loads(row["category_tags"]) if row["category_tags"] else [] + + score_type = meta.get("score_type", "?") + source_detail = "" + if row["source"] == "github": + source_detail = f"⭐ {meta.get('stars', '?')} stars" + elif row["source"] == "arxiv": + source_detail = f"arXiv:{meta.get('arxiv_id', '?')}" + elif row["source"] == "reddit": + source_detail = f"r/{meta.get('subreddit', '?')}" + + return { + "rank": rank, + "source": row["source"], + "title": row["title"], + "url": row["url"], + "score": row["signal_score"], + "score_type": score_type, + "confidence": summary.get("confidence", "?"), + "source_detail": source_detail, + "tags": tags, + "one_liner": summary.get("one_liner", ""), + "key_technical_point": summary.get("key_technical_point", ""), + "potential_use_case": summary.get("potential_use_case", ""), + "first_seen": row["first_seen"], + } + + +def cmd_top(args): + """Top N entries across all sources (per-source ranking).""" + conn = get_db() + cur = conn.cursor() + + # Score filter + min_score = getattr(args, 'min_score', 0) or 0 + # Confidence filter + min_confidence = getattr(args, 'min_confidence', None) or None + # Source filter + source_filter = getattr(args, 'source', None) or None + + where = [] + params = [] + if min_score > 0: + where.append("signal_score >= ?") + params.append(min_score) + if min_confidence: + where.append("json_extract(summary,'$.confidence') = ?") + params.append(min_confidence) + if source_filter: + where.append("source = ?") + params.append(source_filter) + + where_str = (" AND " if where else "") + " AND ".join(where) if where else "" + limit = args.n or 10 + + cur.execute(f""" + SELECT * FROM entries {where_str} + ORDER BY signal_score DESC + LIMIT ? + """, params + [limit]) + + rows = cur.fetchall() + entries = [format_entry(r, i+1) for i, r in enumerate(rows)] + + print(f"Top {len(entries)} entries{' by score' if not source_filter else f' from {source_filter}'}:\n") + _print_entries(entries) + conn.close() + + +def cmd_by_source(args): + """Top N from a specific source.""" + conn = get_db() + cur = conn.cursor() + cur.execute(""" + SELECT * FROM entries WHERE source = ? + ORDER BY signal_score DESC + LIMIT ? + """, (args.source, args.n or 10)) + + rows = cur.fetchall() + entries = [format_entry(r, i+1) for i, r in enumerate(rows)] + + print(f"Top {len(entries)} from {args.source}:\n") + _print_entries(entries) + conn.close() + + +def cmd_by_tag(args): + """Entries matching a tag.""" + conn = get_db() + cur = conn.cursor() + cur.execute(""" + SELECT * FROM entries + WHERE json_extract(category_tags,'$') LIKE ? + ORDER BY signal_score DESC + LIMIT 20 + """, (f'%"{args.tag}"%',)) + + rows = cur.fetchall() + entries = [format_entry(r, i+1) for i, r in enumerate(rows)] + + print(f"Entries tagged '{args.tag}':\n") + _print_entries(entries) + conn.close() + + +def cmd_search(args): + """Keyword search in titles and summaries.""" + conn = get_db() + cur = conn.cursor() + q = f"%{args.query}%" + cur.execute(""" + SELECT * FROM entries + WHERE title LIKE ? + OR json_extract(summary,'$.one_liner') LIKE ? + OR json_extract(summary,'$.key_technical_point') LIKE ? + ORDER BY signal_score DESC + LIMIT 20 + """, (q, q, q)) + + rows = cur.fetchall() + entries = [format_entry(r, i+1) for i, r in enumerate(rows)] + + print(f"Search results for '{args.query}':\n") + _print_entries(entries) + conn.close() + + +def cmd_recent(args): + """Entries from the last N hours.""" + hours = args.hours or 24 + cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%SZ") + conn = get_db() + cur = conn.cursor() + cur.execute(""" + SELECT * FROM entries WHERE first_seen >= ? + ORDER BY first_seen DESC + """, (cutoff,)) + + rows = cur.fetchall() + entries = [format_entry(r, i+1) for i, r in enumerate(rows)] + + print(f"Entries from last {hours}h:\n") + _print_entries(entries) + conn.close() + + +def cmd_snapshot(args): + """Full snapshot for Claude relay. + + Produces a structured summary of the current DB state, + formatted for Claude to reason over. + """ + conn = get_db() + cur = conn.cursor() + + # DB stats + cur.execute("SELECT COUNT(*) FROM entries") + total = cur.fetchone()[0] + + cur.execute("SELECT source, COUNT(*) as cnt, AVG(signal_score) as avg_score, MIN(first_seen) as oldest, MAX(last_updated) as newest FROM entries GROUP BY source") + source_stats = {r["source"]: dict(r) for r in cur.fetchall()} + + # Security-tagged entries + cur.execute(""" + SELECT COUNT(*) FROM entries + WHERE summary IS NOT NULL AND summary != '' AND json_extract(summary,'$.potential_use_case') LIKE '%security%' + """) + security_count = cur.fetchone()[0] + + # Top 10 overall + cur.execute("SELECT * FROM entries WHERE summary IS NOT NULL AND summary != '' ORDER BY signal_score DESC LIMIT 10") + top_entries = [format_entry(r, i+1) for i, r in enumerate(cur.fetchall())] + + # Confidence distribution + cur.execute(""" + SELECT json_extract(summary,'$.confidence') as conf, COUNT(*) as cnt + FROM entries WHERE summary IS NOT NULL AND summary != '' + GROUP BY conf + """) + conf_dist = {r["conf"]: r["cnt"] for r in cur.fetchall()} + + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + snapshot = { + "snapshot_time": now, + "total_entries": total, + "sources": source_stats, + "security_flagged": security_count, + "confidence_distribution": conf_dist, + "top_10": top_entries, + } + + # Output as formatted text for relay + print("=" * 70) + print("AI RESEARCH ORACLE — SNAPSHOT") + print("=" * 70) + print(f"Time: {now}") + print(f"Total entries: {total}") + print() + + print("Source breakdown:") + for src, stats in source_stats.items(): + print(f" {src}: {stats['cnt']} entries, avg score {stats['avg_score']:.2f}") + print() + + if conf_dist: + print(f"Confidence distribution: {conf_dist}") + print() + + if security_count: + print(f"⚠ {security_count} entries flagged as security:dual-use") + print() + + print("Top 10 by signal score (per-source ranking, NOT cross-source comparable):") + print("-" * 70) + for e in top_entries: + score_label = f"{e['score']:.2f} ({e['score_type']})" + conf = e["confidence"] + print(f"\n [{e['rank']}] {e['source'].upper()} | {score_label} | confidence={conf}") + print(f" {e['title']}") + print(f" {e['source_detail']}") + if e["one_liner"]: + print(f" → {e['one_liner'][:120]}") + print(f" Tags: {', '.join(e['tags'][:4])}") + + print("\n" + "=" * 70) + print("NOTE: Scores are NOT comparable across sources. GitHub uses") + print("actual star counts (log-scaled), arXiv/reddit use estimated") + print("heuristics. Rank within-source, not cross-source.") + print("=" * 70) + + conn.close() + + +def cmd_explain(args): + """Explain why an entry scored the way it did.""" + conn = get_db() + cur = conn.cursor() + + if args.entry_id.isdigit(): + cur.execute("SELECT * FROM entries WHERE id = ?", (args.entry_id,)) + else: + q = f"%{args.entry_id}%" + cur.execute("SELECT * FROM entries WHERE title LIKE ?", (q,)) + + row = cur.fetchone() + if not row: + print(f"Entry not found: {args.entry_id}") + conn.close() + return + + meta = json.loads(row["raw_metadata"]) if row["raw_metadata"] else {} + summary = json.loads(row["summary"]) if row["summary"] else {} + tags = json.loads(row["category_tags"]) if row["category_tags"] else [] + + print(f"=== Score Explanation ===\n") + print(f"Title: {row['title']}") + print(f"Source: {row['source']}") + print(f"Score: {row['signal_score']:.2f} ({meta.get('score_type', '?')})") + print(f"Confidence: {summary.get('confidence', '?')}") + print() + + if row["source"] == "arxiv": + print(f"Authors: {meta.get('author_count', '?')}") + print(f"Categories: {', '.join(meta.get('categories', []))}") + print(f"Published: {meta.get('published', '?')}") + print(f"Abstract length: {meta.get('abstract_length', '?')} chars") + print(f"Applied domain: {meta.get('applied_domain', 'none (core AI)')}") + print() + print("Scoring (arXiv — relevance-weighted, structural capped):") + print(f" - Structural (recency+authors+diversity, capped ~3.5): weak signals") + print(f" - AI keyword density in abstract (capped 2.0)") + print(f" - AI methodology claim (propose/novel = up to 2.5)") + print(f" - Applied-domain is TAGGED, not penalized") + if meta.get('applied_domain'): + print(f" Note: tagged '{meta['applied_domain']}' for filtering — no score penalty") + elif row["source"] == "github": + print(f"Stars: {meta.get('stars', '?')}") + print(f"Language: {meta.get('language', '?')}") + print() + print("Scoring (GitHub actual star count, log-scaled)") + + print() + print(f"Tags: {', '.join(tags)}") + if summary.get('one_liner'): + print(f"One-liner: {summary['one_liner'][:120]}") + + conn.close() + + +def cmd_stats(args): + """Database statistics.""" + conn = get_db() + cur = conn.cursor() + + cur.execute("SELECT COUNT(*) FROM entries") + total = cur.fetchone()[0] + + cur.execute("SELECT source, COUNT(*) as cnt, ROUND(AVG(signal_score),2) as avg_score, MIN(signal_score) as min_score, MAX(signal_score) as max_score FROM entries GROUP BY source") + rows = cur.fetchall() + + print(f"Database: {DB_PATH}") + print(f"Total entries: {total}\n") + + print(f"{'Source':<12} {'Count':<8} {'Avg':<8} {'Min':<8} {'Max':<8}") + print("-" * 44) + for r in rows: + print(f"{r['source']:<12} {r['cnt']:<8} {r['avg_score']:<8} {r['min_score']:<8} {r['max_score']:<8}") + + # Summarization status + cur.execute("SELECT COUNT(*) FROM entries WHERE summary IS NOT NULL") + summarized = cur.fetchone()[0] + cur.execute("SELECT COUNT(*) FROM entries WHERE summary IS NULL") + pending = cur.fetchone()[0] + print(f"\nSummarization: {summarized} done, {pending} pending") + + # Confidence distribution + cur.execute("SELECT json_extract(summary,'$.confidence') as c, COUNT(*) as n FROM entries WHERE summary IS NOT NULL AND summary != '' GROUP BY c") + if cur.fetchall(): + print(f"Confidence: {' | '.join(f'{r[0]}={r[1]}' for r in cur.fetchall())}") + + # Recent run history (failure visibility) + cur.execute("SELECT run_time, total_fetched, total_stored, sources_ok, sources_failed FROM run_log ORDER BY id DESC LIMIT 5") + runs = cur.fetchall() + if runs: + print(f"\nRecent runs (last {len(runs)}):") + for r in runs: + failed = json.loads(r["sources_failed"]) if r["sources_failed"] else [] + status = "✓ all ok" if not failed else f"⚠ partial: {', '.join(failed)}" + print(f" {r['run_time']} fetched={r['total_fetched']} stored={r['total_stored']} {status}") + + conn.close() + + +def _print_entries(entries: list[dict]): + """Pretty-print a list of entries.""" + if not entries: + print(" (no results)") + return + + for e in entries: + score_label = f"{e['score']:.2f} ({e['score_type']})" + conf = e["confidence"] + print(f" [{e['rank']}] {e['source'].upper():6} | {score_label} | confidence={conf}") + print(f" {e['title']}") + if e["source_detail"]: + print(f" {e['source_detail']}") + if e["one_liner"]: + print(f" → {e['one_liner'][:120]}") + # Show applied-domain and security tags prominently + shown_tags = [t for t in e["tags"] if t.startswith("applied:") or t.startswith("security:")] + other_tags = [t for t in e["tags"] if not t.startswith("applied:") and not t.startswith("security:")] + display_tags = shown_tags + other_tags[:4] + if display_tags: + print(f" Tags: {', '.join(display_tags)}") + print() + + +def main(): + parser = argparse.ArgumentParser(description="AI Research Oracle — Query") + sub = parser.add_subparsers(dest="command") + + # top + p_top = 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) + + # by-source + p_src = sub.add_parser("by-source", help="Top N from a source") + p_src.add_argument("source") + p_src.add_argument("n", type=int, nargs="?", default=10) + + # by-tag + p_tag = sub.add_parser("by-tag", help="Entries by tag") + p_tag.add_argument("tag") + + # search + p_search = sub.add_parser("search", help="Keyword search") + p_search.add_argument("query") + + # recent + p_recent = sub.add_parser("recent", help="Recent entries") + p_recent.add_argument("--hours", type=int, default=24) + + # snapshot + sub.add_parser("snapshot", help="Full snapshot for Claude") + + # explain + p_explain = sub.add_parser("explain", help="Explain why an entry scored high") + p_explain.add_argument("entry_id", help="Entry ID or partial title") + + # stats + sub.add_parser("stats", help="Database statistics") + + args = parser.parse_args() + + commands = { + "top": cmd_top, + "by-source": cmd_by_source, + "by-tag": cmd_by_tag, + "search": cmd_search, + "recent": cmd_recent, + "snapshot": cmd_snapshot, + "explain": cmd_explain, + "stats": cmd_stats, + } + + cmd = commands.get(args.command) + if cmd: + cmd(args) + else: + parser.print_help() + + +if __name__ == "__main__": + main() diff --git a/reddit_proof.py b/reddit_proof.py new file mode 100644 index 0000000..dbbe236 --- /dev/null +++ b/reddit_proof.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +""" +Reddit Idea Generator — Proof of Concept v5 +Uses Reddit RSS feeds (Atom XML). No browser needed. +Trafilatura for clean text extraction. SQLite for storage. + +Usage: python3 reddit_proof.py [count] +Example: python3 reddit_proof.py 20 +""" + +import sys +import json +import re +import xml.etree.ElementTree as ET +import sqlite3 +import os +import time +import urllib.request +import urllib.error +from datetime import datetime, timezone +from html import unescape + +import trafilatura + +DB_PATH = os.path.join(os.path.dirname(__file__), "oracle.db") +SCHEMA_PATH = os.path.join(os.path.dirname(__file__), "schema.sql") + +SUBREDDITS = [ + "MachineLearning", "artificial", "LocalLLaMA", "Startups", +] + + +def init_db(): + conn = sqlite3.connect(DB_PATH) + with open(SCHEMA_PATH) as f: + conn.executescript(f.read()) + conn.commit() + return conn + + +def fetch_rss(subreddit, sort="hot"): + """Fetch RSS feed for a subreddit. Returns parsed entries.""" + url = f"https://www.reddit.com/r/{subreddit}/{sort}/.rss?limit=50" + req = urllib.request.Request(url, headers={"User-Agent": "oracle-reddit-proof/1.0"}) + + for attempt in range(3): + try: + with urllib.request.urlopen(req, timeout=15) as resp: + xml_data = resp.read().decode("utf-8") + break + except urllib.error.HTTPError as e: + if e.code == 429: + wait = 5 * (attempt + 1) + print(f" 429 on r/{subreddit}, retry in {wait}s") + time.sleep(wait) + continue + print(f" RSS error r/{subreddit}: {e}") + return [] + except Exception as e: + print(f" RSS error r/{subreddit}: {e}") + return [] + else: + print(f" r/{subreddit}: still rate limited, skip") + return [] + + # Parse Atom XML — find all elements + root = ET.fromstring(xml_data) + entries = [] + + # Handle namespace: Atom uses http://www.w3.org/2005/Atom + # But ET.findall with ns prefix requires registering the namespace + # Simpler approach: strip namespace from tags and search directly + for entry in root.iter(): + # Get local name (strip namespace) + tag = entry.tag.split("}")[-1] if "}" in entry.tag else entry.tag + + if tag == "entry": + title = None + link = None + author = "" + content = "" + pub = "" + eid = "" + + for child in entry: + ctag = child.tag.split("}")[-1] + if ctag == "title": + title = child.text + elif ctag == "link": + link = child.get("href", "") + elif ctag == "author": + name_el = child[0] if child else None + if name_el: + name_tag = name_el.tag.split("}")[-1] + if name_tag == "name": + author = name_el.text or "" + elif ctag == "content": + content = child.text or "" + elif ctag == "published": + pub = child.text or "" + elif ctag == "id": + eid = child.text or "" + + if title and link: + entries.append({ + "title": unescape(title.strip()), + "url": link, + "author": unescape(author.strip()), + "content": content, + "published": pub, + "id": eid, + "subreddit": subreddit, + }) + + return entries + + +def clean_html_content(html): + """Extract readable text from Reddit's HTML content.""" + if not html: + return "" + text = re.sub(r"", "", html, flags=re.DOTALL) + text = re.sub(r"]*>", "\n", text) + text = re.sub(r"", "\n", text) + text = re.sub(r"", "\n", text, flags=re.I) + text = re.sub(r"<[^>]+>", "", text) + text = unescape(text) + text = re.sub(r"\n\s*\n+", "\n\n", text) + return text.strip() + + +def main(): + if len(sys.argv) > 1: + count = int(sys.argv[1]) + else: + count = 20 + + print(f"=== Reddit Idea Generator — Proof of Concept v5 ===") + print(f" count: {count}") + print() + + conn = init_db() + cursor = conn.cursor() + + # Step 1: Fetch RSS + print(f"[1/3] Fetching RSS feeds...") + all_entries = [] + seen_ids = set() + + for i, sub in enumerate(SUBREDDITS): + entries = fetch_rss(sub) + new = [e for e in entries if e["id"] not in seen_ids] + seen_ids.update(e["id"] for e in new) + all_entries.extend(new) + if new: + print(f" r/{sub}: {len(new)} entries") + # Rate limit between subreddits + if i < len(SUBREDDITS) - 1: + time.sleep(3) + + print(f" Total: {len(all_entries)} entries") + + if not all_entries: + print("\n No entries fetched. Reddit may be rate-limiting this IP.") + print(" Try again later or use fewer subreddits.") + sys.exit(1) + + # Limit to count + entries_to_store = all_entries[:count] + print(f" Storing {len(entries_to_store)} entries") + + # Step 2: Store + stored = 0 + for entry in entries_to_store: + post_id = entry["id"].replace("t3_", "") + content_text = clean_html_content(entry["content"]) + + # Signal score — RSS hot feed already sorted by relevance + # Use position-based scoring (higher rank = higher score) + idx = entries_to_store.index(entry) + score = max(10.0 - idx * 0.5, 1.0) + + # Category tags + category_tags = ["reddit"] + sub = entry.get("subreddit", "").lower() + if "machinelearning" in sub: + category_tags.append("machine-learning") + elif "artificial" in sub: + category_tags.append("ai-general") + elif "localllama" in sub: + category_tags.append("local-llm") + elif "startups" in sub: + category_tags.append("startups") + + # Post type from title markers + title = entry.get("title", "") + if " [P]" in title or " [p]" in title: + category_tags.append("project") + elif " [R]" in title or " [r]" in title: + category_tags.append("research") + elif " [D]" in title or " [d]" in title: + category_tags.append("discussion") + elif " [N]" in title or " [n]" in title: + category_tags.append("news") + else: + category_tags.append("general") + + # Clean title (remove [X] markers) + clean_title = re.sub(r"\s*\[[A-Z]\]\s*$", "", title) + + raw_meta = { + "subreddit": entry["subreddit"], + "author": entry["author"], + "published": entry["published"], + "text_length": len(content_text), + } + + source_id = post_id or entry["url"].split("/")[-1] or f"rss_{stored}" + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + try: + cursor.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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + "reddit", source_id, entry["url"], clean_title, + content_text, + None, # summary — LLM later + json.dumps(category_tags), + score, + json.dumps(raw_meta), + now, now, + )) + stored += 1 + except Exception as e: + print(f" DB ERROR: {e}") + + conn.commit() + print(f" Stored {stored} entries") + + # Step 3: Summary + print(f"\n[3/3] Summary") + cursor.execute("SELECT COUNT(*) FROM entries") + total = cursor.fetchone()[0] + print(f" Total entries in DB: {total}") + cursor.execute("SELECT COUNT(*) FROM entries WHERE source='reddit'") + reddit_count = cursor.fetchone()[0] + print(f" Reddit entries: {reddit_count}") + cursor.execute("SELECT AVG(signal_score) FROM entries WHERE source='reddit'") + avg_score = cursor.fetchone()[0] or 0 + print(f" Avg signal score: {avg_score:.2f}") + + # Subreddit distribution + cursor.execute(""" + SELECT raw_metadata, COUNT(*) FROM entries + WHERE source='reddit' + GROUP BY raw_metadata + ORDER BY COUNT(*) DESC + """) + print(f"\n Subreddit distribution:") + for meta, cnt in cursor.fetchall(): + d = json.loads(meta) + print(f" r/{d.get('subreddit', '?')}: {cnt}") + + # Top 5 + print(f"\n Top 5 by signal score:") + cursor.execute(""" + SELECT id, title, signal_score, raw_metadata, category_tags, + LENGTH(extracted_text) as text_len + FROM entries WHERE source='reddit' + ORDER BY signal_score DESC + LIMIT 5 + """) + for row in cursor.fetchall(): + eid, title, score, meta, tags, txt_len = row + meta_dict = json.loads(meta) if meta else {} + print(f" [{eid}] score={score:.1f} text={txt_len}ch") + print(f" {title[:90]}") + print(f" r/{meta_dict.get('subreddit', '?')} " + f"by {meta_dict.get('author', '?')}") + + # Extraction quality + print(f"\n Extraction quality (top entry):") + cursor.execute(""" + SELECT title, extracted_text + FROM entries WHERE source='reddit' + ORDER BY signal_score DESC + LIMIT 1 + """) + row = cursor.fetchone() + if row: + title, excerpt = row + print(f" Title: {title[:80]}") + print(f" Length: {len(excerpt) if excerpt else 0} chars") + if excerpt: + print(f" Preview:\n {excerpt[:400]}...") + else: + print(" (empty)") + + # Check for garbled extractions + cursor.execute(""" + SELECT COUNT(*) FROM entries + WHERE source='reddit' AND LENGTH(extracted_text) < 100 + """) + short_count = cursor.fetchone()[0] + if short_count > 0: + print(f"\n ⚠ {short_count}/{stored} entries have very short extractions (<100 chars)") + print(" These are likely link-only posts or external links") + + conn.close() + print(f"\n Database: {DB_PATH}") + print(" Done.") + + +if __name__ == "__main__": + main() diff --git a/schema.sql b/schema.sql new file mode 100644 index 0000000..9e48fab --- /dev/null +++ b/schema.sql @@ -0,0 +1,33 @@ +CREATE TABLE IF NOT EXISTS entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + source_id TEXT NOT NULL, + url TEXT, + title TEXT, + extracted_text TEXT, + summary TEXT, + category_tags TEXT, + signal_score REAL, + raw_metadata TEXT, + first_seen TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + last_updated TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + UNIQUE(source, source_id) +); + +CREATE INDEX IF NOT EXISTS idx_entries_source ON entries(source); +CREATE INDEX IF NOT EXISTS idx_entries_signal ON entries(signal_score DESC); +CREATE INDEX IF NOT EXISTS idx_entries_category ON entries(category_tags); + +-- Run log: records each pipeline invocation for failure visibility + growth control. +-- Partial failures (e.g. Reddit rate-limited) are detectable here, not hidden +-- as a "complete" run. Also enables future pruning decisions (entries older +-- than N days with no re-fetch can be archived). +CREATE TABLE IF NOT EXISTS run_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_time TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + total_fetched INTEGER DEFAULT 0, + total_stored INTEGER DEFAULT 0, + sources_ok TEXT, -- JSON list of sources that succeeded + sources_failed TEXT, -- JSON list of sources that errored/skipped + notes TEXT +); diff --git a/summarize.py b/summarize.py new file mode 100644 index 0000000..af9a6df --- /dev/null +++ b/summarize.py @@ -0,0 +1,552 @@ +#!/usr/bin/env python3 +""" +AI Research Oracle — Summarization Engine (v1). + +Generates structured summaries for entries where summary IS NULL. +Uses source-specific extraction logic (no LLM required — eliminates hallucination). + +Output schema: {one_liner, key_technical_point, potential_use_case, confidence} + +Architecture note: This v1 uses deterministic extraction rules to avoid +hallucination. When a local LLM becomes available (Ollama GPU, Hermes API), +swap in LLM mode via --llm flag. The DB schema is identical. + +Usage: + python3 summarize.py # summarize all pending + python3 summarize.py --source github # specific source + python3 summarize.py --limit 10 # max entries + python3 summarize.py --verify # spot-check 2-3 summaries +""" + +import argparse +import json +import os +import re +import sqlite3 +import sys +import time +from datetime import datetime, timezone + +sys.path.insert(0, os.path.dirname(__file__)) + + +def extract_github_summary(title: str, content: str) -> dict: + """Extract summary from GitHub README content. + + Strategy: Clean HTML, find the first substantive paragraph that + describes the project (usually below the badges), extract the + "what it does" sentence. + """ + # Aggressive HTML cleaning + text = re.sub(r']*>', '\n', content) + text = re.sub(r'

', '\n', content) + text = re.sub(r']*>', '\n## ', text) + text = re.sub(r'', '\n', text) + text = re.sub(r'<[^>]+>', '', text) + text = re.sub(r'&', '&', text) + text = re.sub(r'—', '—', text) + text = re.sub(r''', "'", text) + text = re.sub(r'·', '·', text) + # Remove code blocks (``` ... ```) — often ASCII art + text = re.sub(r'```[\s\S]*?```', '', text) + text = re.sub(r'\n\s*\n+', '\n\n', text) + text = text.strip() + + # Confidence starts from source content quality + source_confidence = "low" + if len(text) > 2000: + source_confidence = "high" + elif len(text) > 500: + source_confidence = "medium" + + # Find the one-liner: look for project description paragraph + one_liner = _find_project_description(text, title) + if not one_liner: + one_liner = title[:200] + + # Key technical point + key_tech = _extract_technical_point(text, source_confidence) + + # Use case + use_case = _extract_use_case(text, title) + + # Quality-gate confidence on extraction signals, not raw length + confidence = _assess_extraction_quality(one_liner, key_tech, use_case, source_confidence) + + # Tag security tooling if detected + if _is_security_tooling(title, one_liner, key_tech): + use_case = use_case + " [security:dual-use]" + + return { + "one_liner": one_liner[:200], + "key_technical_point": key_tech[:200], + "potential_use_case": use_case[:200], + "confidence": confidence, + } + + +def extract_arxiv_summary(title: str, content: str) -> dict: + """Extract summary from arXiv abstract. + + Strategy: arXiv abstracts have a predictable structure: + 1. Background/motivation + 2. "In this paper we propose..." + 3. Results + 4. Implications + + We extract the contribution statement and key finding. + """ + text = re.sub(r'<[^>]+>', ' ', content) + text = re.sub(r'\s+', ' ', text).strip() + + # Confidence based on abstract clarity + confidence = "high" if len(text) > 300 else "medium" + + # One-liner: find the contribution statement + one_liner = _find_contribution(text) + if not one_liner: + # Fallback: use title as base + one_liner = f"This paper presents {title.lower()}" + + # Key technical point: look for method description + key_tech = _extract_method(text) + + # Use case: look for application statements + use_case = _extract_application(text) + + return { + "one_liner": one_liner[:200], + "key_technical_point": key_tech[:200], + "potential_use_case": use_case[:200], + "confidence": confidence, + } + + +def extract_reddit_summary(title: str, content: str) -> dict: + """Extract summary from Reddit post. + + Strategy: Reddit posts vary wildly in quality. Extract the core + question or claim, note if it's discussion vs announcement. + """ + text = re.sub(r'<[^>]+>', ' ', content) + text = re.sub(r'\s+', ' ', text).strip() + + # Confidence based on content length + if len(text) > 500: + confidence = "high" + elif len(text) > 100: + confidence = "medium" + else: + confidence = "low" + + # One-liner from title (Reddit titles are usually the summary) + one_liner = title[:200] if title else text[:150] + + # Key technical point from content + key_tech = text[:200] if text else "No additional content in post" + + # Use case: community relevance + use_case = "AI community discussion" + + return { + "one_liner": one_liner, + "key_technical_point": key_tech, + "potential_use_case": use_case, + "confidence": confidence, + } + + +# --- Extraction helpers --- + +def _assess_extraction_quality(one_liner: str, key_tech: str, use_case: str, source_confidence: str) -> str: + """Assess extraction quality based on output signals, not source length. + + A short-but-complete Reddit title should score higher confidence + than a long README that yielded a fragment. + """ + score = 0 + penalties = 0 + + # One-liner quality + ol = one_liner.strip() + ol_len = len(ol) + + # Length window: 40-200 chars is a reasonable sentence + if 40 <= ol_len <= 200: + score += 2 + elif 20 <= ol_len < 40: + score += 1 + elif ol_len > 200: + penalties += 1 # too long, likely grabbed too much + + # Ends with terminal punctuation + if ol.endswith(('.', '!', '?', '…')): + score += 1 + else: + penalties += 1 + + # Contains subject-verb pattern (basic heuristic) + if re.search(r'\b(?:is|are|provides|enables|implements|makes|allows|builds|creates|runs|uses)\b', ol, re.I): + score += 1 + # Or starts with a proper noun/capitalized phrase + elif re.match(r'^[A-Z]\w+', ol) and ol_len > 30: + score += 0.5 + + # No unmatched brackets (artifact from markdown/HTML) + open_brackets = ol.count('[') + ol.count('(') + close_brackets = ol.count(']') + ol.count(')') + if abs(open_brackets - close_brackets) > 0: + penalties += 1 + if open_brackets > 2: + penalties += 1 # likely grabbed markdown link syntax + + # Key technical point quality + kt = key_tech.strip() + if kt and len(kt) > 20 and not kt.startswith('See '): + score += 1 + else: + penalties += 0.5 + + # Use case quality + uc = use_case.strip() + if uc and len(uc) > 10 and not uc.startswith('Relevant for'): + score += 1 + else: + penalties += 0.5 + + # Final confidence based on score - penalties + net = score - penalties + if net >= 3: + return source_confidence # extraction is good, trust source quality + elif net >= 1: + return "medium" + else: + return "low" + + +def _is_security_tooling(title: str, one_liner: str, key_tech: str) -> bool: + """Detect if a project is security/offensive tooling.""" + combined = f"{title} {one_liner} {key_tech}".lower() + security_signals = [ + "offensive", "pentest", "red team", "exploit", "kill chain", + "attack surface", "vulnerability scan", "zero-day", + "reverse engineer", "c2", "command and control", + ] + return any(sig in combined for sig in security_signals) + + +def _find_project_description(text: str, title: str) -> str | None: + """Find the project description paragraph in a README.""" + paras = text.split('\n\n') + proj_name = title.split(':')[0].split('/')[0].strip().lower() + + for para in paras: + para = para.strip() + if not para or para.startswith('##') or len(para) < 20: + continue + # Skip badges, stats lines, separator lines + if 'img' in para.lower() or 'badge' in para.lower() or 'shields' in para.lower(): + continue + # Skip lines that start with stats (~54%, etc.) + if re.match(r'^[~$#€£¥*»\d]', para): + continue + # Skip ASCII art (high ratio of special chars) + special_chars = sum(1 for c in para if not c.isalnum() and not c.isspace() and c not in ',.!?;:\'"-()[]') + if special_chars / max(len(para), 1) > 0.4: + continue + if len(para) < 40: + continue + # Good paragraph — extract first sentence + sentence = re.split(r'[.!?]', para)[0].strip() + if len(sentence) > 30: + return sentence + '.' + + # Fallback: look for "is a" pattern anywhere + patterns = [ + rf'{re.escape(proj_name[:20])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?', + r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides)\s+[^.]+\.?', + ] + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + return None + + +def _find_what_sentence(text: str, title: str) -> str | None: + """Find the 'X is a...' sentence that describes what the project does.""" + patterns = [ + rf'{re.escape(title[:30])}\s+(?:is|enables|provides|implements)\s+[^.]+\.?', + r'(?:This\s+)?(?:project|library|framework|tool|package)\s+(?:is|enables|provides|implements)\s+[^.]+\.?', + r'(?:makes|allows)\s+[^\s]+\s+(?:to|can)\s+[^.]+\.?', + r'(?:\w+\s+(?:is|provides|enables|implements|delivers))\s+[a-z].{10,100}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + # Fallback: first meaningful paragraph + for para in text.split('\n\n'): + para = para.strip() + if len(para) > 30 and not para.startswith('#'): + return para[:200] + return None + + +def _find_contribution(text: str) -> str | None: + """Find the 'we propose/introduce/present' statement in an abstract.""" + patterns = [ + r'(?:we|this\s+paper)\s+(?:propose|introduce|present|propose and evaluate)\s+[^.]{10,150}\.', + r'(?:we\s+(?:show|demonstrate|find|discover|observe))\s+[^.]{10,150}\.', + r'(?:we\s+(?:introduce|present|propose))\s+(?:a|an|our)\s+\w+\s+[^.]{5,150}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + # Fallback: first sentence + first = re.split(r'[.!?]', text)[0].strip() + return first if first else None + + +def _extract_technical_point(text: str, confidence: str) -> str: + """Extract the main technical approach or innovation.""" + patterns = [ + r'architecture(?:\s+designed)?\s+(?:for|to|that)\s+[^.]+\.?', + r'(?:using|via|based\s+on|through)\s+[a-z][^.]{10,100}\.', + r'(?:novel|new|unique|innovative)\s+\w+\s+[^.]{5,80}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + # Fallback: confidence-based + if confidence == "low": + return "Technical details not available in extracted content" + return "See README for technical details" + + +def _extract_method(text: str) -> str: + """Extract the method/approach from an arXiv abstract.""" + patterns = [ + r'(?:method|approach|framework|technique|model|system)\s+(?:based|using|via|through|with)\s+[a-z][^.]{10,120}\.', + r'(?:combining|leveraging|exploiting)\s+[a-z][^.]{10,120}\.', + r'(?:learn|train|optimize|generate)\s+[a-z][^.]{10,120}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + # Fallback: core contribution + for pattern in [ + r'(?:propose|introduce)\s+(?:a|an)\s+[^.]{10,100}\.', + ]: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + return "See full paper for methodology" + + +def _extract_use_case(text: str, title: str) -> str: + """Extract potential use case from README content.""" + patterns = [ + r'(?:for|to)\s+(?:developers|engineers|researchers|teams)\s+who?\s+[^.]{5,80}\.', + r'(?:enables|allows|helps)\s+[^\s]+\s+to\s+[^.]{10,80}\.', + r'(?:use\s+case|application|target\s+user)\s*:\s*[^.]{10,80}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + return f"Relevant for {title.lower()[:50]} developers and users" + + +def _extract_application(text: str) -> str: + """Extract application/use case from arXiv abstract.""" + patterns = [ + r'(?:application|use\s+case|can\s+be\s+used|could\s+be\s+applied)\s+(?:for|in|to)\s+[a-z][^.]{10,80}\.', + r'(?:improve|enhance|advance)\s+[a-z][^.]{10,80}\.', + ] + + for pattern in patterns: + match = re.search(pattern, text, re.I) + if match: + return match.group(0)[:200] + + # Generic fallback based on title keywords + title_lower = text[:200].lower() + if any(k in title_lower for k in ["agent", "agentic"]): + return "Building AI agent systems" + elif any(k in title_lower for k in ["verification", "verify"]): + return "LLM output verification and reliability" + elif any(k in title_lower for k in ["embodied", "robot"]): + return "Embodied AI and robotics applications" + elif any(k in title_lower for k in ["distill"]): + return "Model distillation and knowledge transfer" + return "See paper for specific applications" + + +def summarize_entry(entry: dict, conn: sqlite3.Connection) -> bool: + """Summarize a single entry using rule-based extraction.""" + source = entry["source"] + title = entry["title"] + content = entry.get("extracted_text", "") + eid = entry["id"] + + if not content or len(content) < 50: + return False + + # Source-specific extraction + if source == "github": + summary = extract_github_summary(title, content) + elif source == "arxiv": + summary = extract_arxiv_summary(title, content) + elif source == "reddit": + summary = extract_reddit_summary(title, content) + else: + summary = extract_reddit_summary(title, content) # fallback + + # Store + cur = conn.cursor() + cur.execute("UPDATE entries SET summary = ? WHERE id = ?", + (json.dumps(summary), eid)) + conn.commit() + return True + + +def verify_summaries(conn: sqlite3.Connection, source: str, sample_size: int = 3): + """Spot-check summaries against source text. + + Look for hallucinated specifics: numbers, claims, features not + present in the original extracted_text. + + NOTE: Rule-based extraction v1 is inherently lower-risk for + hallucination since it extracts actual text, not generates new claims. + But we still verify the extraction logic is working correctly. + """ + cur = conn.cursor() + cur.execute(""" + SELECT id, title, extracted_text, summary + FROM entries WHERE source = ? AND summary IS NOT NULL + ORDER BY RANDOM() + LIMIT ? + """, (source, sample_size)) + + rows = cur.fetchall() + if not rows: + print(f" No summaries to verify for {source}") + return + + for eid, title, source_text, summary_json in rows: + summary = json.loads(summary_json) + one_liner = summary.get("one_liner", "") + confidence = summary.get("confidence", "?") + + issues = [] + + # Check: does the one-liner contain text actually present in source? + # (For rule-based extraction, this should always be true) + words = one_liner.split()[:5] + found = sum(1 for w in words if w.lower() in source_text.lower()) + if found < 3: + issues.append(f"Low overlap: {found}/5 words from source") + + # Check: confidence matches content length + if confidence == "high" and len(source_text) < 500: + issues.append("High confidence on short source") + elif confidence == "low" and len(source_text) > 2000: + issues.append("Low confidence on long source") + + if issues: + print(f" ⚠ [{eid}] {title[:50]}... issues: {'; '.join(issues)}") + print(f" Summary: {one_liner[:80]}...") + else: + print(f" ✓ [{eid}] {title[:50]}... confidence={confidence}") + + time.sleep(0.3) + + +def main(): + parser = argparse.ArgumentParser(description="AI Research Oracle — Summarization") + parser.add_argument("--source", default=None, help="Filter by source (github/arxiv/reddit)") + parser.add_argument("--limit", type=int, default=0, help="Max entries (0=all)") + parser.add_argument("--verify", action="store_true", help="Spot-check summaries") + args = parser.parse_args() + + db_path = os.path.join(os.path.dirname(__file__), "oracle.db") + conn = sqlite3.connect(db_path) + cur = conn.cursor() + + # Find pending entries + where = "summary IS NULL" + params = [] + if args.source: + where += " AND source = ?" + params.append(args.source) + + cur.execute(f"SELECT COUNT(*) FROM entries WHERE {where}", params) + total_pending = cur.fetchone()[0] + print(f"=== Summarization Engine (Rule-based v1) ===") + print(f" Pending entries: {total_pending}") + + if total_pending == 0: + print(" Nothing to summarize.") + conn.close() + return + + # Fetch entries + limit_clause = " LIMIT ?" if args.limit > 0 else "" + limit_params = params + [args.limit] if args.limit > 0 else params + + cur.execute(f""" + SELECT id, source, title, extracted_text + FROM entries WHERE {where} + ORDER BY signal_score DESC + {limit_clause} + """, limit_params) + + entries = [{"id": r[0], "source": r[1], "title": r[2], "extracted_text": r[3]} for r in cur.fetchall()] + print(f" Processing: {len(entries)} entries") + print() + + success = 0 + failed = 0 + for entry in entries: + try: + if summarize_entry(entry, conn): + success += 1 + print(f" ✓ [{entry['id']}] {entry['title'][:60]}... ({entry['source']})") + else: + failed += 1 + print(f" ⚠ [{entry['id']}] Skipped: {entry['title'][:40]}... (too short)") + except Exception as e: + print(f" ✗ [{entry['id']}] Error: {e}") + failed += 1 + + if args.verify: + print(f"\n [Verification]") + sources = [args.source] if args.source else ["github", "arxiv", "reddit"] + for src in sources: + print(f" Checking {src}...") + verify_summaries(conn, src) + print() + + print(f" Results: {success} summarized, {failed} failed") + conn.close() + print(f"\n Done.") + + +if __name__ == "__main__": + main()