Epictetus
a9fbe27178
- adapters/__init__.py: add load_queries() + source_config() (stdlib json,
safe fallback to {} on missing/corrupt config so pipeline never crashes).
- config/queries.json: per-adapter blocks (hackernews.keywords, arxiv.categories,
reddit.subreddits, rss.feeds+keywords, github.search_terms). JSON (not
yaml) to honor Athena's dependency-free runtime; PyYAML avoided.
- hackernews: drop class AI_KEYWORDS + the DUPLICATE inline list inside
_is_ai_relevant() (the internal drift Ty flagged). Now loads self.ai_keywords
from config. Simplified matching to single substring pass (boundary variants
'ai ',' ai','ai-','-ai' approximate word-boundary; dropped the niche
'compute+tech-context' guard as not worth centralizing).
- reddit: DEFAULT_SUBREDDITS kept as fallback; __init__ prefers config.
- arxiv: DEFAULT_CATEGORIES kept as fallback; __init__ prefers config.
- rss: FEEDS + AI_KEYWORDS kept as module fallbacks; __init__ prefers config.
Keywords stay regex form (re.search) as in original.
- github: trending queries moved to config search_terms; fallback retained.
- DELETE reddit_proof.py: standalone PoC v5 at repo root, own main()+init_db()
+ direct INSERT OR REPLACE, NOT in cron, NOT imported anywhere -> dead
code. Also removes its byte-duplicate SUBREDDITS.
NOTE: fallback class constants remain intentionally (issue #7 cut #5: safe
rollout). Curation VALUES now live in one file; the constants are inert
unless config/queries.json is missing.
Verified: all adapters compile; config loads (HN 46 kw, RSS 10 feeds);
full dry-run fetches all 6 sources; grep confirms HN internal dup list gone.
282 lines
10 KiB
Python
282 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Hacker News adapter for Athena.
|
|
Fetches best stories via the official Firebase API — no auth, no scraping.
|
|
|
|
API: https://hacker-news.firebaseio.com/v0/
|
|
Endpoints used:
|
|
/v0/beststories.json — IDs of current best stories (200 items)
|
|
/v0/item/{id}.json — story details (title, url, score, time, by, type)
|
|
|
|
Rate limits: HN is generous; 3s spacing between batch requests is polite.
|
|
Strategy: fetch best stories, filter for AI/ML relevance, score by points+comments.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone
|
|
|
|
from adapters import SourceAdapter, source_config
|
|
|
|
|
|
class HackerNewsAdapter(SourceAdapter):
|
|
"""Hacker News Firebase API adapter."""
|
|
|
|
BASE = "https://hacker-news.firebaseio.com/v0"
|
|
|
|
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."""
|
|
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}")
|
|
return None
|
|
except Exception as e:
|
|
print(f" Request error: {e}")
|
|
return None
|
|
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.
|
|
"""
|
|
title_lower = title.lower()
|
|
for kw in self.ai_keywords:
|
|
if kw in title_lower:
|
|
return True
|
|
return False
|
|
|
|
def _score(self, item: dict) -> float:
|
|
"""Score: points + engagement (comments), log scale."""
|
|
score = item.get("score", 0)
|
|
comments = item.get("descendants", 0)
|
|
|
|
import math
|
|
# Log scale on points (HN stories range 1-2000+)
|
|
point_score = min(math.log1p(score) / 1.8, 7.0)
|
|
# Engagement bonus (comments indicate discussion quality)
|
|
engagement = min(math.log1p(comments) / 2.5, 2.0)
|
|
# Type bonus: original content (no URL) is often higher signal
|
|
if item.get("type") == "story" and not item.get("url"):
|
|
engagement += 0.5 # self-post bonus (often original content)
|
|
|
|
return min(round(point_score + engagement, 2), 10.0)
|
|
|
|
def _tags(self, item: dict) -> list:
|
|
"""Generate category tags from HN metadata."""
|
|
tags = ["hackernews"]
|
|
|
|
title = (item.get("title", "") or "").lower()
|
|
url = (item.get("url", "") or "").lower()
|
|
combined = f"{title} {url}"
|
|
|
|
# Source type detection
|
|
if item.get("type") == "story" and not item.get("url"):
|
|
tags.append("self-post")
|
|
else:
|
|
tags.append("link-post")
|
|
|
|
# Domain tagging from URL
|
|
if "arxiv.org" in url:
|
|
tags.append("source:arxiv")
|
|
elif "github.com" in url:
|
|
tags.append("source:github")
|
|
elif "blog" in url or "medium.com" in url or "substack.com" in url:
|
|
tags.append("source:blog")
|
|
elif "twitter.com" in url or "x.com" in url:
|
|
tags.append("source:twitter")
|
|
elif "youtube.com" in url or "youtu.be" in url:
|
|
tags.append("source:video")
|
|
|
|
# AI subdomain tagging from title
|
|
if any(kw in combined for kw in ["agent", "agents"]):
|
|
tags.append("topic:agents")
|
|
if any(kw in combined for kw in ["llm", "language model", "gpt", "transformer"]):
|
|
tags.append("topic:llm")
|
|
if any(kw in combined for kw in ["inference", "compute", "training"]):
|
|
tags.append("topic:infrastructure")
|
|
if any(kw in combined for kw in ["alignment", "safety", "trust"]):
|
|
tags.append("topic:safety")
|
|
if any(kw in combined for kw in ["open-source", "open source", "oss"]):
|
|
tags.append("topic:open-source")
|
|
if any(kw in combined for kw in ["pricing", "cost", "margin", "business"]):
|
|
tags.append("topic:business")
|
|
|
|
# Engagement level
|
|
score = item.get("score", 0)
|
|
if score > 500:
|
|
tags.append("engagement:high")
|
|
elif score > 200:
|
|
tags.append("engagement:medium")
|
|
|
|
return tags
|
|
|
|
def fetch(self, query: str = "", limit: int = 20) -> list[dict]:
|
|
"""
|
|
Fetch AI/ML stories from Hacker News.
|
|
|
|
Fetches best stories (HN's internal ranking), filters for AI relevance,
|
|
returns top entries by score.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# Fetch best story IDs (200 items)
|
|
story_ids = self._request("/beststories.json")
|
|
if not story_ids or not isinstance(story_ids, list):
|
|
print(" Failed to fetch best stories")
|
|
return []
|
|
|
|
# Fetch story details (batch, with spacing)
|
|
stories = []
|
|
# Only need to check ~100 stories to find 20 AI ones
|
|
for idx, sid in enumerate(story_ids[:100]):
|
|
item = self._request(f"/item/{sid}.json")
|
|
if item and item.get("type") == "story" and item.get("title"):
|
|
stories.append(item)
|
|
if idx % 20 == 19: # polite spacing every 20 requests
|
|
time.sleep(1)
|
|
|
|
# Filter for AI relevance
|
|
ai_stories = [s for s in stories if self._is_ai_relevant(s.get("title", ""))]
|
|
|
|
# Score and sort
|
|
for s in ai_stories:
|
|
s["_score"] = self._score(s)
|
|
ai_stories.sort(key=lambda s: s.get("_score", 0), reverse=True)
|
|
ai_stories = ai_stories[:limit]
|
|
|
|
# Convert to DB format
|
|
entries = []
|
|
for item in ai_stories:
|
|
score = item.pop("_score", 0)
|
|
source_id = str(item.get("id", ""))
|
|
|
|
# URL: story URL or fallback
|
|
url = item.get("url", "") or ""
|
|
if not url:
|
|
url = f"https://news.ycombinator.com/item/{source_id}"
|
|
|
|
# Title (clean)
|
|
title = item.get("title", "")
|
|
|
|
# Extracted text: for self-posts, we don't have body text via this API.
|
|
# We store what we have (title + metadata) and leave extracted_text empty.
|
|
# A future enhancement could fetch comments via /v0/item/{id}.json children.
|
|
extracted_text = ""
|
|
|
|
tags = self._tags(item)
|
|
|
|
# Structured metadata
|
|
raw_meta = {
|
|
"id": item.get("id"),
|
|
"by": item.get("by", ""),
|
|
"score": item.get("score", 0),
|
|
"descendants": item.get("descendants", 0),
|
|
"time": item.get("time", 0),
|
|
"type": item.get("type", "story"),
|
|
"is_self_post": not bool(item.get("url")),
|
|
"score_type": "actual", # real HN points
|
|
}
|
|
|
|
now_str = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
entries.append({
|
|
"source": "hackernews",
|
|
"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
|
|
|
|
parser = argparse.ArgumentParser(description="Hacker News adapter for Athena")
|
|
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"=== Hacker News Adapter ===")
|
|
print(f" Limit: {args.limit}")
|
|
print()
|
|
|
|
adapter = HackerNewsAdapter()
|
|
entries = adapter.fetch(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} hn_points={meta.get('score', '?')}")
|
|
print(f" {e['title'][:90]}")
|
|
print(f" {e['url']}")
|
|
|
|
print(f"\n Done.")
|