Add Hacker News adapter + wire into pipeline
- adapters/hackernews.py: HN Firebase API adapter with AI keyword filtering - Word-boundary matching to avoid substring traps (Britain/Guinea) - Score: log(points) + log(comments), actual HN scores - Wired into ENABLED_SOURCES + verification in pipeline.py - Live test: 10 AI/ML stories fetched, all clean
This commit is contained in:
Epictetus
@@ -0,0 +1,321 @@
|
||||
#!/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
|
||||
|
||||
|
||||
class HackerNewsAdapter(SourceAdapter):
|
||||
"""Hacker News Firebase API adapter."""
|
||||
|
||||
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)"
|
||||
|
||||
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.
|
||||
|
||||
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()
|
||||
|
||||
# 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:
|
||||
"""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.")
|
||||
Reference in New Issue
Block a user