athena-oracle/adapters/hackernews.py

#!/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, browser_user_agent, jitter_sleep
from adapters._store import true_first_seen, upsert_entries


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 browser_user_agent()

    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": browser_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
                jitter_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")
            # first_seen = TRUE publish date (HN 'time'), not harvest time
            first_seen = true_first_seen(raw_meta, "hackernews", now_str)
            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": first_seen,
                "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 = upsert_entries(conn, entries)
        print(f"\n  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.")