From 6396b66b45901e5981b6cf1980001c8ef828daaf Mon Sep 17 00:00:00 2001 From: Epictetus Date: Wed, 8 Jul 2026 05:32:28 +0000 Subject: [PATCH] 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 --- adapters/hackernews.py | 321 +++++++++++++++++++++++++++++++++++++++++ pipeline.py | 24 ++- 2 files changed, 344 insertions(+), 1 deletion(-) create mode 100644 adapters/hackernews.py diff --git a/adapters/hackernews.py b/adapters/hackernews.py new file mode 100644 index 0000000..6680c6b --- /dev/null +++ b/adapters/hackernews.py @@ -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.") diff --git a/pipeline.py b/pipeline.py index 8ac7432..1311b41 100644 --- a/pipeline.py +++ b/pipeline.py @@ -31,10 +31,11 @@ ADAPTERS = { "github": lambda: __import__("adapters.github", fromlist=["GitHubAdapter"]).GitHubAdapter(), "arxiv": lambda: __import__("adapters.arxiv", fromlist=["ArxivAdapter"]).ArxivAdapter(), "reddit": lambda: __import__("adapters.reddit", fromlist=["RedditAdapter"]).RedditAdapter(), + "hackernews": lambda: __import__("adapters.hackernews", fromlist=["HackerNewsAdapter"]).HackerNewsAdapter(), } # Default enabled sources -ENABLED_SOURCES = ["github", "arxiv", "reddit"] +ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews"] def init_db(db_path: str, schema_path: str) -> sqlite3.Connection: @@ -144,6 +145,27 @@ def verify_entries(conn: sqlite3.Connection, source: str, sample_size: int = 3): except Exception as e: print(f" ? [{eid}] {title[:60]}... verify failed: {e}") + elif src == "hackernews": + hn_id = meta.get("id", "") + if hn_id: + try: + req = urllib.request.Request( + f"https://hacker-news.firebaseio.com/v0/item/{hn_id}.json", + headers={"User-Agent": "ai-oracle/0.1"}, + ) + with urllib.request.urlopen(req, timeout=10) as resp: + live = json.loads(resp.read().decode("utf-8")) + live_score = live.get("score", 0) + db_score = meta.get("score", 0) + drift = abs(live_score - db_score) + if drift <= 20: + print(f" ✓ [{eid}] {title[:60]}... hn_score={db_score} drift={drift}") + passed += 1 + else: + print(f" ? [{eid}] {title[:60]}... hn_score={db_score} vs live={live_score} DRIFT={drift}") + except Exception as e: + print(f" ? [{eid}] {title[:60]}... verify failed: {e}") + elif src == "reddit": # For Reddit, just check the URL is reachable (no easy verification of scores via RSS) if url: