Epictetus
560 lines
21 KiB
Python
560 lines
21 KiB
Python
#!/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 math
|
|
import os
|
|
import random
|
|
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, browser_user_agent, jitter_sleep
|
|
from adapters._store import true_first_seen, upsert_entries
|
|
|
|
|
|
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=4, user_agent=None):
|
|
"""
|
|
Args:
|
|
subreddits: List of subreddit names.
|
|
rate_limit: Base seconds between subreddit requests (with jitter).
|
|
user_agent: Custom User-Agent header.
|
|
"""
|
|
self.subreddits = subreddits or self.DEFAULT_SUBREDDITS
|
|
self.rate_limit = rate_limit
|
|
self.user_agent = user_agent or browser_user_agent()
|
|
|
|
def name(self) -> str:
|
|
return "reddit"
|
|
|
|
def _sleep_with_jitter(self, base=None):
|
|
"""Sleep with ±30% jitter to avoid pattern detection."""
|
|
base = base or self.rate_limit
|
|
jitter = base * 0.3 * (2 * random.random() - 1) # ±30%
|
|
time.sleep(base + jitter)
|
|
|
|
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."""
|
|
url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
|
|
req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
|
|
|
|
backoff = [2, 5] # staggered backoff: 2s then 5s
|
|
for attempt in range(3): # max 3 attempts
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=8) 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 < len(backoff):
|
|
delay = backoff[attempt]
|
|
print(f" RSS 429 for r/{subreddit}, retrying in {delay}s")
|
|
jitter_sleep(delay)
|
|
continue
|
|
print(f" RSS rate-limited for r/{subreddit}, skip")
|
|
return []
|
|
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 after 3 attempts, 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=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:
|
|
jitter_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")
|
|
|
|
# Local-inference / on-device signal (suggested source: r/MachineLearning
|
|
# "I tried X on-device" posts — high builder signal → local-inference feed)
|
|
on_device_kw = [
|
|
"on-device", "on device", "local inference", "local llm",
|
|
"ran locally", "running locally", "in my pocket", "on my phone",
|
|
"edge device", "offline", "no gpu", "consumer gpu", "rtx",
|
|
"single gpu", "self-host", "self host",
|
|
]
|
|
if any(kw in title_lower or kw in content for kw in on_device_kw):
|
|
tags.append("local-inference")
|
|
|
|
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})
|
|
|
|
backoff = [2, 5] # staggered backoff: 2s then 5s
|
|
for attempt in range(3): # max 3 attempts
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=8) 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 < len(backoff):
|
|
delay = backoff[attempt]
|
|
print(f" RSS 429 for r/{subreddit}, retrying in {delay}s")
|
|
jitter_sleep(delay)
|
|
continue
|
|
print(f" RSS rate-limited for r/{subreddit}, skip")
|
|
return []
|
|
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 after 3 attempts, 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)
|
|
# Try JSON first — if it's blocked on the first subreddit, skip
|
|
# the test-RSS call (which would waste a request and risk rate-limiting)
|
|
# and go straight to the RSS loop
|
|
first_json = self._try_json(self.subreddits[0])
|
|
json_worked = bool(first_json)
|
|
|
|
all_entries = []
|
|
seen_ids = set()
|
|
|
|
# Add first JSON results if any
|
|
for p in first_json:
|
|
if p["id"] not in seen_ids:
|
|
seen_ids.add(p["id"])
|
|
all_entries.append(p)
|
|
|
|
# If JSON didn't work, fall back to RSS for all subreddits
|
|
if not json_worked:
|
|
print(" JSON endpoints blocked, using RSS fallback")
|
|
# Brief cooldown before RSS barrage
|
|
jitter_sleep(3)
|
|
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)
|
|
self._sleep_with_jitter()
|
|
|
|
# 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")
|
|
first_seen = true_first_seen(raw_meta, "reddit", now_str)
|
|
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": first_seen,
|
|
"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 = upsert_entries(conn, entries)
|
|
print(f"\n Stored {stored} entries")
|
|
|
|
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.")
|