12 Commits

Author SHA1 Message Date
Epictetus 8017ded3ba fix(adapters): shared retry helper + run_log failure_class + enable RSS (issues #1 #2 #9)
- adapters/__init__.py: add http_get() unified retry (429/5xx only, max 2
  attempts, capped exp backoff) + AdapterHTTPError carrying failure_class;
  SourceAdapter.last_failure_class set on failure for pipeline capture.
- arxiv/github/huggingface/hackernews/reddit: route HTTP through http_get.
  Preserves GitHub 403 rate-limit retry and Reddit 403/429 fast-bail.
- schema.sql + pipeline.py: add run_log.failure_class column; rollup most-
  severe class across sources (5xx>4xx>429>error>zero_fetch>ok).
- pipeline.py: ENABLE RSS in ENABLED_SOURCES (was registered, disabled).
- RSS smoke test surfaced 3 broken feeds (anthropic 404, googleai 404,
  metaai 301) — left as-is, captured in feed_failures; URL fix is separate
  discovery task, not guessed.

Verified: full dry-run fetches all 6 sources; github live fetch OK;
Reddit 429 fast-bail preserved; no import/syntax errors.
2026-07-10 16:34:05 +00:00
Leonard 13d2d1dd1d Fill out Prioritized Task List and Deployment Plan 2026-07-09 04:29:42 +00:00
Leonard 2e68de8451 Fill out Prioritized Task List and Deployment Plan 2026-07-09 04:29:40 +00:00
Ty f64240d6dc added ascii tree to reqs 2026-07-08 21:44:34 +00:00
Ty 3ed95be797 Update docs/MVP-PRD.md
updated requirements
2026-07-08 21:38:24 +00:00
Ty 04863f3dcf Update docs/MVP-PRD.md
adding requirements
2026-07-08 21:33:48 +00:00
Ty 8f4716f875 docs: add initial user stories 2026-07-08 20:40:19 +00:00
Ty 75479abffa docs: add Chapter 1 draft - Vision and Scope (chapter 1 draft complete) 2026-07-08 20:19:10 +00:00
Ty 11329d7c74 docs: add Personas and Archetypes derived from main 2026-07-08 20:07:21 +00:00
Ty c32a846f79 stub: initial MVP milestone docs - Deployment-Plan.md 2026-07-08 19:49:17 +00:00
Ty 6c246c6467 stub: initial MVP milestone docs - Prioritized-Task-List.md 2026-07-08 19:49:16 +00:00
Ty 315a04041e stub: initial MVP milestone docs - MVP-PRD.md 2026-07-08 19:49:12 +00:00
15 changed files with 895 additions and 248 deletions
+85 -28
View File
@@ -1,39 +1,18 @@
"""Source adapters for AI Research Oracle."""
import json
import os
import urllib.request
import urllib.error
import time
from abc import ABC, abstractmethod
# Centralized curation config (issue #7). One file, per-adapter blocks.
# Stdlib-only (JSON, not YAML) to honor Athena's dependency-free runtime.
_QUERIES_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"config", "queries.json")
def load_queries():
"""Load config/queries.json. Returns {'sources': {...}}.
Safe fallback: if the file is missing/corrupt, returns an empty
{'sources': {}} so adapters fall back to their class defaults
(constructor None-override) instead of crashing the pipeline.
"""
try:
with open(_QUERIES_PATH) as f:
data = json.load(f)
return data if isinstance(data, dict) else {"sources": {}}
except Exception:
return {"sources": {}}
def source_config(name: str) -> dict:
"""Return the per-adapter block for `name`, or {} if absent."""
return load_queries().get("sources", {}).get(name, {}) or {}
class SourceAdapter(ABC):
"""Base class for all ingestion adapters."""
def __init__(self):
# Set by http_get when a request fails (issue #2 classification)
self.last_failure_class = None
@abstractmethod
def name(self) -> str:
"""Source name: 'github', 'arxiv', 'reddit'."""
@@ -43,3 +22,81 @@ class SourceAdapter(ABC):
def fetch(self, query: str = "", limit: int = 20) -> list[dict]:
"""Return entries matching DB schema fields."""
pass
class AdapterHTTPError(Exception):
"""Raised by http_get on non-retryable or exhausted HTTP failures.
failure_class is one of: '429', '5xx', '4xx', 'error'.
Consumed by pipeline.py to populate run_log.failure_class (issue #2).
"""
def __init__(self, status, failure_class, message=""):
super().__init__(message)
self.status = status
self.failure_class = failure_class
def classify_http_status(code):
"""Map an HTTP status code to a run_log failure_class."""
if code == 429:
return "429"
if 500 <= code < 600:
return "5xx"
return "4xx" # 401/403/404 etc = client/config problem, not retried
def http_get(url, headers=None, timeout=15, max_retries=2,
backoff_base=2, retry_403_ratelimit=False, return_headers=False,
owner=None):
"""GET with a unified retry policy shared by all adapters (issue #1).
Retries ONLY on 429 and 5xx (transient). 4xx other than 429 are NOT
retried (config/client problems). Optional GitHub-style 403 rate-limit
handling via retry_403_ratelimit (waits for X-RateLimit-Reset header).
Returns response body bytes (or (body, headers) tuple if return_headers).
Raises AdapterHTTPError on non-retryable or exhausted failures, carrying
.failure_class for run_log classification. If owner is provided, sets
owner.last_failure_class so the pipeline can record it (issue #2).
"""
def _fail(code, fclass, msg):
if owner is not None:
owner.last_failure_class = fclass
raise AdapterHTTPError(code, fclass, msg)
req = urllib.request.Request(url, headers=headers or {})
last_err = None
for attempt in range(max_retries + 1):
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = resp.read()
if return_headers:
return body, resp.headers
return body
except urllib.error.HTTPError as e:
last_err = e
code = e.code
if code == 429 or (code == 403 and retry_403_ratelimit):
if attempt < max_retries:
if code == 403:
reset = int(e.headers.get("X-RateLimit-Reset", 0))
wait = (max(reset - int(time.time()), 0) + 1
if reset else backoff_base * (attempt + 1) * 15)
time.sleep(min(wait, 300))
else:
time.sleep(min(backoff_base * (attempt + 1), 30))
continue
fclass = "429" if code == 429 else "4xx"
_fail(code, fclass, f"HTTP {code} for {url} (exhausted)")
if 500 <= code < 600:
if attempt < max_retries:
time.sleep(min(backoff_base * (attempt + 1), 30))
continue
_fail(code, "5xx", f"HTTP {code} for {url} (exhausted)")
# other 4xx (401/403 w/o flag/404) — not retried
_fail(code, classify_http_status(code), f"HTTP {code} for {url}")
except Exception as e: # non-HTTP (timeout, DNS, conn reset)
_fail(None, "error", f"Request error: {e}")
# Defensive: loop should always raise or return
_fail(getattr(last_err, "code", None), "error", str(last_err))
+12 -10
View File
@@ -32,7 +32,7 @@ import xml.etree.ElementTree as ET
from datetime import datetime, timedelta, timezone
from html import unescape
from adapters import SourceAdapter, source_config
from adapters import SourceAdapter, http_get, AdapterHTTPError
# arXiv API
ARXIV_API = "http://export.arxiv.org/api/query"
@@ -45,8 +45,12 @@ class ArxivAdapter(SourceAdapter):
DEFAULT_CATEGORIES = ["cs.AI", "cs.LG", "cs.CL"]
def __init__(self, categories=None, rate_limit=3):
cfg = source_config("arxiv")
self.categories = categories or cfg.get("categories") or self.DEFAULT_CATEGORIES
"""
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:
@@ -147,14 +151,12 @@ class ArxivAdapter(SourceAdapter):
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")
raw = http_get(url, headers={"User-Agent": "ai-oracle/0.1"},
timeout=30, max_retries=2, owner=self)
return self._parse_atom(raw.decode("utf-8"))
except AdapterHTTPError as e:
print(f" {e.failure_class}: arXiv query ({e})")
return []
except Exception as e:
print(f" arXiv request error: {e}")
+24 -38
View File
@@ -17,7 +17,7 @@ import urllib.error
import urllib.parse
from datetime import datetime, timedelta, timezone
from adapters import SourceAdapter, source_config
from adapters import SourceAdapter, http_get, AdapterHTTPError
class GitHubAdapter(SourceAdapter):
@@ -29,11 +29,6 @@ class GitHubAdapter(SourceAdapter):
"""Initialize with optional read-only token (5000 req/hr vs 60)."""
self.token = token or os.environ.get("GITHUB_TOKEN", "")
self.cache = {}
# Curation centralized (issue #7): trending queries from config
cfg = source_config("github")
self.search_terms = cfg.get("search_terms") or [
"ai agent", "llm OR inference OR rag", "autonomous agent OR AI tool",
]
def name(self) -> str:
return "github"
@@ -48,39 +43,26 @@ class GitHubAdapter(SourceAdapter):
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):
"""GET via shared retry helper; 403 rate-limit handled as transient."""
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))
raw, headers = http_get(
url, headers=self._headers(), timeout=15,
max_retries=max_retries, retry_403_ratelimit=True,
return_headers=True, owner=self)
except AdapterHTTPError as e:
print(f" {e.failure_class}: GitHub {url}")
return None
# Informational: flag if we're close to the unauth rate ceiling
try:
remaining = int(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
print(f" ⚠ Rate limit low ({remaining} remaining)")
except Exception:
pass
try:
return json.loads(raw.decode("utf-8"))
except Exception as e:
print(f" Request error: {e}")
return None
print(f" GitHub decode error: {e}")
return None
def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list:
@@ -163,8 +145,12 @@ class GitHubAdapter(SourceAdapter):
cutoff = (now - timedelta(days=30)).strftime("%Y-%m-%d")
# Three queries for breadth: agents, LLM/infra, and security/tools
repos = []
for q in self.search_terms:
batch = self._search_repos(f"{q} created:>{cutoff}", sort="stars", per_page=30)
for q in [
f"ai agent created:>{cutoff}",
f"llm OR inference OR rag created:>{cutoff}",
f"autonomous agent OR AI tool created:>{cutoff}",
]:
batch = self._search_repos(q, sort="stars", per_page=30)
repos.extend(batch)
time.sleep(1) # polite spacing
+56 -21
View File
@@ -20,7 +20,7 @@ import urllib.request
import urllib.error
from datetime import datetime, timezone
from adapters import SourceAdapter, source_config
from adapters import SourceAdapter, http_get, AdapterHTTPError
class HackerNewsAdapter(SourceAdapter):
@@ -28,46 +28,81 @@ class HackerNewsAdapter(SourceAdapter):
BASE = "https://hacker-news.firebaseio.com/v0"
AI_KEYWORDS = [
# Multi-word phrases (unambiguous)
"language model", "deep learning", "foundation model", "retrieval augmented",
"code generation", "context length", "context window", "attention mechanism",
# Compound/abbreviations (unambiguous)
"llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
# Specific company/product names
"openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
# Topic-specific (with word boundary awareness in _is_ai_relevant)
"inference", "transformer", "diffusion", "alignment", "fine-tun",
"embedd", "pretrain", "post-train", "multimodal", "reasoning",
# Domain-specific (need boundary check)
"ai ", " ai", "ai-", "-ai", # "ai" as word, not substring
"agent", "agents", "neural", "autonomous",
"compute", "training run", "computer use", "coding agent",
# Community terms
"local-llm", "local llama", "llama ",
]
def __init__(self, user_agent=None):
self.user_agent = user_agent or "python:athena:v0.1 (by tony_tech)"
# Curation now centralized (issue #7): load from config/queries.json
cfg = source_config("hackernews")
self.ai_keywords = cfg.get("keywords") or []
def name(self) -> str:
return "hackernews"
def _request(self, path: str, max_retries: int = 2) -> dict | list | None:
"""Make a GET request to the HN Firebase API."""
"""GET via shared retry helper (retries 429/5xx)."""
url = f"{self.BASE}{path}"
req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
for attempt in range(max_retries + 1):
try:
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
except (urllib.error.HTTPError, urllib.error.URLError) as e:
if attempt < max_retries:
time.sleep(3 * (attempt + 1))
continue
print(f" HTTP error: {e}")
raw = http_get(url, headers={"User-Agent": self.user_agent},
timeout=15, max_retries=max_retries, owner=self)
except AdapterHTTPError as e:
print(f" {e.failure_class}: HN {path}")
return None
try:
return json.loads(raw.decode("utf-8"))
except Exception as e:
print(f" Request error: {e}")
return None
print(f" HN decode error: {e}")
return None
def _is_ai_relevant(self, title: str) -> bool:
"""Check if a story title is AI/ML relevant.
Keywords are loaded from config/queries.json (issue #7) into
self.ai_keywords — single source of truth, no inline duplicate.
Substring match; callers pass lowercased titles for boundary terms.
Uses multi-pass matching: first check unambiguous multi-word/phrases,
then check word-boundary matches for shorter keywords that could
false-positive (e.g. 'ai' matching 'Britain').
"""
title_lower = title.lower()
for kw in self.ai_keywords:
# Pass 1: unambiguous keywords (multi-word, compound, specific names)
unambiguous = [
"language model", "deep learning", "foundation model", "retrieval augmented",
"code generation", "context length", "context window", "attention mechanism",
"llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
"openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
"inference", "transformer", "diffusion", "alignment", "fine-tun",
"embedd", "pretrain", "post-train", "multimodal", "reasoning",
"agent", "agents", "neural", "autonomous",
"training run", "computer use", "coding agent",
"local-llm", "local llama", "llama ",
]
for kw in unambiguous:
if kw in title_lower:
return True
# Pass 2: word-boundary check for "ai" and "compute" (avoid 'Britain', 'Guinea', etc.)
import re
if re.search(r'\bai\b', title_lower):
return True
if re.search(r'\bcompute\b', title_lower):
# Only if combined with other tech context
tech_words = ["gpu", "tpu", "cluster", "datacenter", "data center", "server"]
if any(w in title_lower for w in tech_words):
return True
return False
def _score(self, item: dict) -> float:
+9 -14
View File
@@ -34,7 +34,7 @@ import urllib.request
import urllib.error
from datetime import datetime, timedelta, timezone
from adapters import SourceAdapter
from adapters import SourceAdapter, http_get, AdapterHTTPError
class HuggingFaceAdapter(SourceAdapter):
@@ -84,23 +84,18 @@ class HuggingFaceAdapter(SourceAdapter):
return headers
def _request(self, path: str, max_retries: int = 2) -> list | dict | None:
"""Make a GET request to the HF API."""
"""GET via shared retry helper (retries 429/5xx)."""
url = f"{self.BASE}{path}"
req = urllib.request.Request(url, headers=self._headers())
for attempt in range(max_retries + 1):
try:
with urllib.request.urlopen(req, timeout=20) as resp:
return json.loads(resp.read().decode("utf-8"))
except (urllib.error.HTTPError, urllib.error.URLError) as e:
if attempt < max_retries:
time.sleep(3 * (attempt + 1))
continue
print(f" HF API error: {e}")
raw = http_get(url, headers=self._headers(), timeout=20,
max_retries=max_retries, owner=self)
except AdapterHTTPError as e:
print(f" {e.failure_class}: HF {path}")
return None
try:
return json.loads(raw.decode("utf-8"))
except Exception as e:
print(f" Request error: {e}")
return None
print(f" HF decode error: {e}")
return None
def _is_ai_relevant(self, model: dict) -> bool:
+31 -44
View File
@@ -24,13 +24,13 @@ import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from html import unescape
from adapters import SourceAdapter, source_config
from adapters import SourceAdapter, http_get, AdapterHTTPError
class RedditAdapter(SourceAdapter):
"""Reddit RSS + JSON adapter."""
# Default subreddits for AI content (fallback if config missing)
# Default subreddits for AI content
DEFAULT_SUBREDDITS = [
"MachineLearning", "artificial", "LocalLLaMA", "Startups",
]
@@ -57,8 +57,7 @@ class RedditAdapter(SourceAdapter):
rate_limit: Seconds between subreddit requests.
user_agent: Custom User-Agent header.
"""
cfg = source_config("reddit")
self.subreddits = subreddits or cfg.get("subreddits") or self.DEFAULT_SUBREDDITS
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)"
@@ -102,33 +101,27 @@ class RedditAdapter(SourceAdapter):
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})
"""Fetch RSS feed for a subreddit (shared retry helper).
for attempt in range(2): # max 2 attempts, fail fast
Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
retry then []; 5xx -> helper retry then []. Not slower than before.
"""
url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
try:
with urllib.request.urlopen(req, timeout=10) as resp:
xml_data = resp.read().decode("utf-8")
break
except urllib.error.HTTPError as e:
if e.code in (403,):
print(f" RSS blocked (HTTP {e.code}) for r/{subreddit}")
return []
if e.code == 429:
if attempt == 0:
time.sleep(2) # single retry with short backoff
continue
raw = http_get(url, headers={"User-Agent": self.user_agent},
timeout=10, max_retries=1, backoff_base=2, owner=self)
except AdapterHTTPError as e:
if e.status == 403:
print(f" RSS blocked (HTTP 403) for r/{subreddit}")
elif e.status == 429:
print(f" RSS rate-limited for r/{subreddit}, skip")
return []
print(f" RSS HTTP {e.code} for r/{subreddit}")
else:
print(f" RSS {e.failure_class} 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 []
xml_data = raw.decode("utf-8")
# Parse Atom XML
entries = []
@@ -304,33 +297,27 @@ class RedditAdapter(SourceAdapter):
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})
"""Fetch RSS feed for a subreddit (shared retry helper).
for attempt in range(2): # max 2 attempts, fail fast
Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
retry then []; 5xx -> helper retry then []. Not slower than before.
"""
url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
try:
with urllib.request.urlopen(req, timeout=10) as resp:
xml_data = resp.read().decode("utf-8")
break
except urllib.error.HTTPError as e:
if e.code in (403,):
print(f" RSS blocked (HTTP {e.code}) for r/{subreddit}")
return []
if e.code == 429:
if attempt == 0:
time.sleep(2) # single retry with short backoff
continue
raw = http_get(url, headers={"User-Agent": self.user_agent},
timeout=10, max_retries=1, backoff_base=2, owner=self)
except AdapterHTTPError as e:
if e.status == 403:
print(f" RSS blocked (HTTP 403) for r/{subreddit}")
elif e.status == 429:
print(f" RSS rate-limited for r/{subreddit}, skip")
return []
print(f" RSS HTTP {e.code} for r/{subreddit}")
else:
print(f" RSS {e.failure_class} 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 []
xml_data = raw.decode("utf-8")
# Parse Atom XML
entries = []
+3 -15
View File
@@ -24,7 +24,7 @@ import feedparser
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from adapters import SourceAdapter, source_config
from adapters import SourceAdapter
# Curated feed list — AI-focused, reliable, diverse publishers.
@@ -83,18 +83,6 @@ AI_KEYWORDS = [
class RSSFeedsAdapter(SourceAdapter):
"""RSS feed aggregator for commercial AI news."""
# Module-level fallbacks (used only if config/queries.json is missing)
FEEDS = [
("rss:techcrunch", "TechCrunch AI", "https://techcrunch.com/category/artificial-intelligence/feed/"),
]
AI_KEYWORDS = [r"\bai\b"]
def __init__(self):
# Curation centralized (issue #7): config wins, fallbacks otherwise
cfg = source_config("rss")
self.feeds = cfg.get("feeds") or list(self.FEEDS)
self.ai_keywords = cfg.get("keywords") or list(self.AI_KEYWORDS)
def name(self) -> str:
return "rss"
@@ -104,7 +92,7 @@ class RSSFeedsAdapter(SourceAdapter):
tag_text = " ".join(tags).lower()
combined = text + " " + tag_text
for pattern in self.ai_keywords:
for pattern in AI_KEYWORDS:
if re.search(pattern, combined):
return True
return False
@@ -159,7 +147,7 @@ class RSSFeedsAdapter(SourceAdapter):
all_entries = []
feed_failures = []
for source_key, label, url in self.feeds:
for source_key, label, url in FEEDS:
try:
d = feedparser.parse(url)
if d.status not in (200, 301, 302, 307, 308) or not d.entries:
-49
View File
@@ -1,49 +0,0 @@
{
"sources": {
"hackernews": {
"keywords": [
"language model", "deep learning", "foundation model", "retrieval augmented",
"code generation", "context length", "context window", "attention mechanism",
"llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
"openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
"inference", "transformer", "diffusion", "alignment", "fine-tun",
"embed", "pretrain", "post-train", "multimodal", "reasoning",
"ai ", " ai", "ai-", "-ai",
"agent", "agents", "neural", "autonomous",
"compute", "training run", "computer use", "coding agent",
"local-llm", "local llama", "llama "
]
},
"arxiv": {
"categories": ["cs.AI", "cs.LG", "cs.CL"]
},
"reddit": {
"subreddits": ["MachineLearning", "artificial", "LocalLLaMA", "Startups"]
},
"rss": {
"feeds": [
["rss:techcrunch", "TechCrunch AI", "https://techcrunch.com/category/artificial-intelligence/feed/"],
["rss:venturebeat", "VentureBeat AI", "https://venturebeat.com/category/ai/feed/"],
["rss:theverge", "The Verge AI", "https://www.theverge.com/rss/ai-artificial-intelligence/index.xml"],
["rss:ainews", "AI News", "https://www.artificialintelligence-news.com/feed/"],
["rss:decoder", "The Decoder", "https://www.the-decoder.com/feed/"],
["rss:mittr", "MIT Tech Review AI", "https://www.technologyreview.com/topic/artificial-intelligence/feed/"],
["rss:openai", "OpenAI Blog", "https://openai.com/blog/rss.xml"],
["rss:anthropic", "Anthropic News", "https://www.anthropic.com/rss/news.xml"],
["rss:googleai", "Google AI Blog", "https://blog.google/technology/rss.xml"],
["rss:metaai", "Meta AI Blog", "https://ai.meta.com/blog/rss.xml"]
],
"keywords": [
"\\bai\\b", "\\bmachine learning\\b", "\\bdeep learning\\b", "\\bneural\\b",
"\\bgenerative ai\\b", "\\bgenerative\\b", "\\bllm\\b", "\\blarge language\\b",
"\\bfoundation model\\b", "\\btransformer\\b", "\\baugmented\\b",
"\\bagent\\b", "\\bautonomous\\b", "\\bmcp\\b", "\\bfunction call\\b",
"\\btool use\\b", "\\brai\\b", "\\bretrieval\\b",
"\\binference\\b", "\\bmodel\\b", "\\bembedding\\b", "\\btoken\\b"
]
},
"github": {
"search_terms": ["machine-learning", "deep-learning", "llm", "ai-agent", "transformer"]
}
}
}
+42
View File
@@ -0,0 +1,42 @@
# Deployment Plan — Athena MVP
## Prerequisites
- Docker installed on the target host
- Network access to the 6 source APIs/feeds (arxiv, github, huggingface, hackernews, reddit, rss_feeds)
- `GITHUB_TOKEN` available as an environment variable (optional, but raises the GitHub rate limit from 60/hr to 5000/hr)
- `HUGGINGFACE_TOKEN` available as an environment variable
- A local Ollama instance running `llama3.2:1b`, or an alternate reachable inference backend if swapping — per the model-agnostic `summarize(text) -> (summary, model)` contract
- Hermes cron infrastructure available and able to invoke `oracle-pipeline.sh`
## Setup
1. Clone `main` (not a milestone branch) onto the target host
2. `pip install -r requirements.txt` if present; otherwise confirm stdlib + `requests` are available
3. Run `schema.sql` against a fresh `oracle.db` — this file is git-ignored and created locally, never committed
4. Export required environment variables (`GITHUB_TOKEN`, `HUGGINGFACE_TOKEN`) — never hardcode these
5. Build and run inside Docker with the 150MB memory cap and non-root user enforced, per the PRD platform requirements
6. Manual smoke test: run `python3 pipeline.py` once and confirm ingest → store → summarize → score completes without errors before handing off to cron
## Cron / Scheduling
1. Confirm `oracle-pipeline.sh` is the entry point Hermes cron calls
2. Schedule for 13:00 UTC daily
3. Add a lock file or PID check so overlapping runs can't happen if a prior run is still in progress
4. Confirm the cron environment actually carries the exported tokens — cron environments are frequently minimal and won't inherit an interactive shell's exports
## Hermes Integration
1. Confirm the wrapper script's exit codes are meaningful (0 = success, non-zero = failure) so Hermes can act on them
2. Define where Hermes should look for pipeline output/logs
3. Decide on a failure-notification path (Hermes alert, log flag, etc.) — **not yet specified, needs a decision**
## Monitoring
1. Aggregate logs from each pipeline stage (ingest, store, summarize, score)
2. Track theme-scan new-arrival counts per cycle per theme — this is the core signal the falsification logic depends on, so it deserves visibility beyond raw logs
3. Add a heartbeat/dead-man's-switch alert if a scheduled run doesn't fire, rather than relying on someone noticing missing data days later
4. Watch memory usage against the 150MB cap under real production load, not just dev conditions
## Rollback
1. Back up `oracle.db` before any schema change — it's git-ignored and not recoverable from the repo itself
2. If a bad deploy breaks the pipeline, revert to the last known-good commit on `main` and redeploy the Docker image
3. Check falsification state (new-arrivals counters) after any rollback — rolling back mid-window could distort the 7-day dead-thesis calculation if not handled carefully
---
*Draft prepared by Claude from the README, whitepaper falsification logic, and MVP-PRD platform requirements on `main`. The Hermes failure-notification path is the one open decision blocking this from being final.*
+161
View File
@@ -0,0 +1,161 @@
## Chapter 1: Vision and Scope
### Elevator Pitch
Athena is an autonomous research intelligence engine that cuts through high-volume, fragmented signals by ingesting from multiple sources, surfacing cross-source convergence, and using falsification to distinguish real momentum from noise. While the initial focus is on AI signals, the system is designed to work with any class of signals. It delivers actionable insight into emerging trends and capability gaps while remaining model-agnostic and lightweight enough to run autonomously.
### 1.1 Vision
#### Why are we building it?
The AI space produces an overwhelming volume of new research, tools, discussions, and model releases every day. Individual sources only provide partial views, making it difficult to distinguish genuine, sustained trends from one-day spikes. Without a system that can detect convergence across sources and validate momentum over time, real opportunities tied to emerging capability gaps are missed.
#### What happens if we dont build it?
Without this capability, builders and researchers will continue to operate with fragmented, noisy signals. Early indicators of meaningful trends will remain hidden, decisions will stay reactive, and the ability to spot validated cross-source momentum before it becomes obvious will be lost.
#### When must it be done?
The foundational ability to reliably ingest, score, and validate signals through falsification must be established before meaningful trend detection and opportunity mapping can occur. This forms the core of the MVP and must be in place to enable the system to deliver on its intended value.
### 1.2 Personas and Archetypes
See committed document:
**`docs/Personas-and-Archetypes.md`** (on `MVP-milestone` branch)
**Summary of scoped personas and archetypes for MVP:**
**Personas**
- Pers-1 (Bob) Sector Trend Tracker (New to AI)
- Pers-2 (Alice) Content Creator
- Pers-3 (Sam) Hermes Research Agent
**Archetypes**
- Arch-1 (Small Scrappy VPS)
- Arch-2 (Research Consumption Layer)
All user stories in this PRD are scoped to combinations of the above.
### 1.3 Use Case Priority Taxonomy
This PRD focuses on defining the core functionality required for MVP. It also catalogs use cases and requirements across V1.0 V1.5 to maintain context. The primary goal is to deliver a working MVP, with future PRDs derived from the remaining prioritized content.
We will use the following prioritization model:
- **MVP**: The short list of P1 use cases required to prove the concept with a working prototype.
- **P1**: Use cases that are fundamental to successfully implementing the product vision.
- **P2**: Use cases that add strength, convenience, and quality to the product vision.
- **P3**: Use cases that bring additional value but can be cut if time or resource constrained.
## Chapter 2: User Stories (Bob)
These user stories are based on the personas and archetypes document contained in this repo.
Chapter 2.1 - Bob's user stories
**As Bob, I want to…**
**Bob-1.** Automatically receive daily updates on new AI innovations without having to manually check multiple sources.
**Bob-5.** See emerging trends and differentiate durable signal from temporary or artificial hype.
**Bob-10.** See when the same idea or pattern is appearing across multiple independent sources (GitHub, arXiv, Reddit, HN, HF).
**Bob-15.** Identify emerging capability gaps or opportunities early, before they become widely obvious.
**Bob-20.** Have research that gives me confidence it is exhaustive and vetted.
**Bob-25.** Adjust or alter the underlying data feeds and weights so I can tune the accuracy and relevance of the output.
**Bob-30.** Understand why a particular signal is considered strong or weak (e.g., cross-source convergence or falsification results).
Chapter 2.2 - Alice's user stories
As Alice, I want to…
Alice-1. Integrate deep, vetted research directly into my existing content production pipeline so I can reduce manual research time.
Alice-5. Query the research system with follow-up questions to explore specific angles or topics on demand.
Alice-10. Have my tools automatically receive curated, high-signal research so I can focus on content creation instead of information filtering.
Alice-15. Get research outputs in a structured format that my existing AI tools and workflows can consume without manual reformatting.
Alice-20. Quickly surface non-obvious insights and patterns from research data to develop more compelling content angles.
Alice-25. Control which research sources and signals are prioritized so the output stays aligned with my content focus and audience.
Alice-30. Understand the reasoning and supporting evidence behind key research findings so I can speak to them confidently in my content.
## Chapter 3: Requirements
Requirements defined as what the product / system must do, differentiated from what the persona can accomplish. Requirements are defined to meet the needs of use cases as well as the architectural system design.
High level design (refer to ***TBD_Design.MD for full design details)
High-Level Design
.
├── Runtime Environment
│ ├── Linux
│ └── Docker (containerized)
├── Core Components
│ ├── Database: SQLite
│ ├── Scheduling: Cron
│ └── Runtime: Python
├── Connectivity
│ ├── Outbound (Internet)
│ │ ├── HTTP client for data feeds (RSS, cURL, optional Playwright)
│ │ └── OpenAI-compatible inference endpoints
│ ├── Inbound (Internet)
│ │ └── HTTP server endpoint (MCP + external consumers)
│ └── Internal (Intranet)
│ └── HTTP client for local inference (e.g. Hermes)
├── Storage
│ ├── File system (daily digest artifacts stored outside container)
│ └── Temporary working storage during pipeline execution
├── Configuration & Secrets
│ ├── Research topic manifest (feeds, URLs, declarations)
│ ├── System settings (YAML)
│ └── Secrets (.env)
├── Business Logic / Pipeline Flow
│ ├── Starting trigger
│ ├── Preflight checks
│ ├── Query feeds → Temporary result storage
│ ├── Vet and promote final results to database
│ ├── Optional daily digest generation
│ └── Cleanup and sleep
└── Observability
├── Structured logging
├── Diagnostics and instrumentation (inside Docker)
└── Health/status reporting
3.1 Setup and configuration (REQ-SNC-XX)
Requirements for initial setup, deployment configs, updating, and uninstall
REQ-SNC-05 -, with outbound access to the internet and in/outbound access to the underlying OS network
REQ-SNC-10 - The installation process shall be a single command which can be run interactively or silently
REQ-SNC-15 - The insallation shall utilize best-practice settings and secrets storage
REQ-SNC-20 -
3.2 Platform requirements (REQ-PLT-XX)
REQ-PLT-05 - All processes will run as standard user (no admin / sudo elevation necessary)
REQ-PLT-10 - ...
REQ-PLT-15 - The system shall be Docker based limited to 150MB of memory
Requirements addressing what OS and hardware support is in scope
3.3 Performance and scalability (REQ-PERF-XX)
3.4 Instrumenation and diagnostics (REQ-DIAG-XX)
3.5
### 3.1 Reliability
REQ-REL-05: Once setup and configured, the system will reliably operate without interaction from the user.
REQ-REL-10: The system shall automatically retry failed source fetches with exponential backoff.
REQ-REL-15: The system shall not lose previously stored data on restart or failure.
REQ-REL-20: The daily pipeline shall complete successfully even if one or more sources are unavailable.
3.3 Observability and Diagnostics
REQ-DIAG-05: The system shall produce structured logs with timestamps and severity levels.
REQ-DIAG-10: A health check endpoint or command shall report overall system status and last successful run.
REQ-DIAG-15: Run logs shall capture per-source success/failure and basic metrics (items fetched, stored, failed).
3.4 Security
REQ-SEC-05: All external HTTP calls shall use TLS.
REQ-SEC-10: No secrets shall be hardcoded or stored in plaintext.
REQ-SEC-15: The system shall support least-privilege access for outbound API calls.
3.5 Integration and extensibility
REQ-INT-05: The system shall expose research output in a structured, machine-readable format (e.g., JSON files or API) consumable by external tools.
REQ-INT-10: The adapter layer shall support adding new sources without modifying core pipeline logic.
+59
View File
@@ -0,0 +1,59 @@
# Personas and Archetypes (Derived from /main)
**Status**: First Draft Derived from existing docs on `main`
**Source**: README.md + whitepaper.md (branch: main)
**Date**: 2026-07-08
## Overview
This document extracts the implied users and operating contexts directly from the current documentation on the `main` branch. It serves as the baseline before we expand or refine.
---
## Personas
### Pers-1 (Bob) Sector Trend Tracker (New to AI)
- Is relatively new to AI and the broader space.
- Wants to stay current with trends in AI (and potentially other sectors) without getting overwhelmed.
- Needs a way to keep up with the high volume of new research, tools, and discussions with minimal ongoing effort.
- Benefits from a system that filters noise and surfaces what actually matters.
### Pers-2 (Alice) Content Creator
- Runs a YouTube channel and an X account with 25k followers.
- Goal is to grow her audience significantly (targeting 1M followers).
- Needs help doing research across AI and related topics.
- Wants to convert research signals into interesting, timely, and compelling content for her audience.
### Pers-3 (Sam) Hermes Research Agent
- Is a Hermes agent profile with its own memory and endpoint connection.
- Acts as the dedicated research team member for an AI-first development team.
- Needs to stay current on a defined market segment (AI for the MVP; extensible to other segments later).
- Consumes structured signals from Athena to support ongoing research and decision-making within the team.
---
## Archetypes (Operating Environments)
### Arch-1 (Small Scrappy VPS)
- Small, low-budget, and scrappy VPS environment.
- Used primarily for learning and early prototyping.
- Requires a small-footprint workload that can be memory-constrained so it doesnt destabilize the host system.
### Arch-2 (Research Consumption Layer)
- Functions as a consumption layer for Athenas research output.
- Designed to support downstream AI systems (examples: MCP tools, LoRA adapters, or other agent profiles).
- Focuses on making Athenas signals and summaries easily consumable by other systems rather than direct human use.
---
## Notes & Limitations (from /main)
- The current documentation does **not** describe team or multi-user usage.
- Emphasis is on autonomous operation and signal integrity.
- Polished human-facing interfaces (e.g., daily digest) are not yet built.
---
## Next Steps
This version incorporates the updated personas and archetypes.
+42
View File
@@ -0,0 +1,42 @@
# Prioritized Task List — Athena MVP
Tied to the personas and requirements in `MVP-PRD.md`. Ordered by phase; within each phase, roughly in the order they should be tackled.
## Phase 1: Get Running Daily
- [ ] Verify the cron entry (`oracle-pipeline.sh`) fires reliably at 13:00 UTC under Hermes
- [ ] Confirm `pipeline.py` runs the full ingest → store → summarize → score cycle without manual intervention
- [ ] Add a lock/guard so a slow run can't overlap with the next day's cron trigger
- [ ] Validate all 6 adapters (arxiv, github, huggingface, hackernews, reddit, rss_feeds) independently — one adapter failing shouldn't kill the whole run
- [ ] Confirm environment-only secrets (`GITHUB_TOKEN`, `HUGGINGFACE_TOKEN`) resolve correctly in the cron context (cron environments are often stripped down compared to an interactive shell)
## Phase 2: Core Functionality
- [ ] Confirm `schema.sql` initializes `oracle.db` cleanly and stays idempotent across repeated runs
- [ ] Verify `theme_scan.py`'s 4-theme tagging (tool-call, context, compute, trust) against a few real days of data
- [ ] Confirm the falsification counter (new arrivals per cycle) is genuinely idempotent — re-running against unchanged data must yield 0 new
- [ ] Wire `summarize.py` to degrade gracefully when the Ollama endpoint (`llama3.2:1b`) isn't reachable — ingestion, scoring, and theme-scan must keep running without it
- [ ] Confirm `archive.py`'s cold-storage rotation doesn't delete data still needed inside the 7-day falsification window
## Phase 3: Observability & Reliability
- [ ] Add structured logging per pipeline stage (ingest, store, summarize, score) with pass/fail per adapter
- [ ] Surface theme-scan counts (new arrivals per theme per cycle) somewhere inspectable, not just buried in log files
- [ ] Add a daily heartbeat/health check so a silent failure (e.g. cron didn't fire at all) is detectable rather than just showing up as missing data later
- [ ] Decide and implement retry/backoff behavior for adapters that hit rate limits (especially GitHub without a token: 60/hr)
## Phase 4: Human Consumption Layer
- [ ] Extend `query.py` to support Bob's cross-source convergence lookups and Alice's curated-research pulls
- [ ] Define the output format(s) for a "trend confirmed" vs. "trend killed" verdict (per the 7-day dead-thesis rule)
- [ ] Decide how Alice's content pipeline actually consumes Athena's output — file drop, API, direct DB read — this is currently undefined
## Phase 5: Validation & UAT
- [ ] Run the pipeline unattended for at least one full 7-day falsification window
- [ ] Manually verify at least one theme through to a real "confirmed" or "killed" verdict
- [ ] Walk Bob's and Alice's user stories from `MVP-PRD.md` end-to-end against real output, not synthetic data
- [ ] Confirm memory stays under the 150MB cap under real daily load, not just in a light dev test
## Phase 6: Hermes Integration
- [ ] Confirm `oracle-pipeline.sh`'s contract matches what Hermes cron expects (exit codes, output location)
- [ ] Decide how Hermes is notified on pipeline failure vs. success — not yet specified
- [ ] Confirm the non-root execution requirement is actually satisfied inside the Hermes-invoked environment, not just in local Docker testing
---
*Draft prepared by Claude from the MVP-PRD, Personas doc, and README/whitepaper on `main`. Open items flagged "not yet specified" need a decision before Phase 46 can be considered done.*
+28 -7
View File
@@ -37,7 +37,7 @@ ADAPTERS = {
}
# Default enabled sources
ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface"]
ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"]
def init_db(db_path: str, schema_path: str) -> sqlite3.Connection:
@@ -252,9 +252,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
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)}
source_stats[source_name] = {"fetched": 0, "stored": 0,
"error": str(e),
"failure_class": "error"}
continue
# Capture classification from the adapter (set by http_get on failure)
fc = getattr(adapter, "last_failure_class", None)
# Add adapter_version to metadata
for entry in entries:
meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"]
@@ -262,7 +267,8 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
entry["raw_metadata"] = json.dumps(meta)
all_entries.extend(entries)
source_stats[source_name] = {"fetched": len(entries), "stored": 0}
source_stats[source_name] = {"fetched": len(entries), "stored": 0,
"failure_class": fc or "ok"}
print(f" Fetched: {len(entries)} entries")
# Small spacing between sources
@@ -294,16 +300,31 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
# Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but
# is still a degraded run — record it so run_log can tell
# "intermittent vs consistently-broken" apart over time.
zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")]
zero = [s for s, st in source_stats.items()
if st.get("fetched", 0) == 0 and not st.get("error")]
notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")]
if zero:
notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}")
notes = "; ".join(notes_parts) or "all sources ok"
# Rollup failure_class (issue #2): most severe across sources.
# Priority: 5xx > 4xx > 429 > error > zero_fetch > ok
rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0}
classes = [st.get("failure_class", "ok") for st in source_stats.values()]
if any(c in ("5xx", "4xx", "429", "error") for c in classes):
run_fc = max((c for c in classes if c in rank),
key=lambda c: rank[c])
elif zero:
run_fc = "zero_fetch"
else:
run_fc = "ok"
try:
conn.execute("""
INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
VALUES (?, ?, ?, ?, ?)
""", (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes))
INSERT INTO run_log (total_fetched, total_stored, sources_ok,
sources_failed, failure_class, notes)
VALUES (?, ?, ?, ?, ?, ?)
""", (len(all_entries), stored, json.dumps(ok), json.dumps(failed),
run_fc, notes))
conn.commit()
except Exception as e:
print(f" ⚠ run_log write failed: {e}")
+318
View File
@@ -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 <entry> 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"<div[^>]*>", "\n", text)
text = re.sub(r"</div>", "\n", text)
text = re.sub(r"<br\s*/?>", "\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()
+3
View File
@@ -22,6 +22,8 @@ CREATE INDEX IF NOT EXISTS idx_entries_category ON entries(category_tags);
-- 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).
-- failure_class (issue #2): one of 4xx / 5xx / 429 / zero_fetch / ok, derived
-- from the real HTTP response via adapters.http_get, not guessed after the fact.
CREATE TABLE IF NOT EXISTS run_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_time TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
@@ -29,6 +31,7 @@ CREATE TABLE IF NOT EXISTS run_log (
total_stored INTEGER DEFAULT 0,
sources_ok TEXT, -- JSON list of sources that succeeded
sources_failed TEXT, -- JSON list of sources that errored/skipped
failure_class TEXT, -- 4xx / 5xx / 429 / zero_fetch / ok
notes TEXT
);