Tony_tech/athena-oracle
3
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases Wiki Activity
Files
fix/adapter-health-1-2-9
athena-oracle/adapters/reddit.py
T
Epictetus 23cce4d609 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:38:57 +00:00

541 lines
20 KiB
Python
Raw Permalink Blame History

#!/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, http_get, AdapterHTTPError
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=1, 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"<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"<p[^>]*>", "\n", text)
text = re.sub(r"</p>", "\n", text)
text = re.sub(r"<a[^>]*href=\"([^\"]+)\"[^>]*>([^<]*)</a>", 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 (shared retry helper).
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:
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")
else:
print(f" RSS {e.failure_class} for r/{subreddit}")
return []
except Exception as e:
print(f" RSS error r/{subreddit}: {e}")
return []
xml_data = raw.decode("utf-8")
# 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=10) 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 (shared retry helper).
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:
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")
else:
print(f" RSS {e.failure_class} for r/{subreddit}")
return []
except Exception as e:
print(f" RSS error r/{subreddit}: {e}")
return []
xml_data = raw.decode("utf-8")
# 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)
# Try JSON first — if it's blocked on the first subreddit, bail fast
# rather than wasting time on all subreddits
first_json = self._try_json(self.subreddits[0])
if not first_json:
# JSON is blocked site-wide, try one RSS to confirm
test_rss = self._fetch_rss(self.subreddits[0])
if not test_rss:
print(" Reddit blocked (403/429), returning empty")
return []
# RSS works — fall through to full fetch below
all_entries = []
seen_ids = set()
json_worked = bool(first_json)
# Add first JSON results
for p in first_json:
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.")
Reference in New Issue View Git Blame Copy Permalink