Build It: A Daily Data Pipeline With LLM Enrichment

# pyproject.toml: httpx, pandas, anthropic, sqlalchemy
import asyncio
import httpx
import pandas as pd
from sqlalchemy import create_engine, text
from anthropic import AsyncAnthropic

DB_URL = "sqlite:///pipeline.db"
engine = create_engine(DB_URL)

def init_db():
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT NOT NULL,
                url TEXT,
                fetched_at TEXT NOT NULL,
                sentiment TEXT,
                UNIQUE(id)
            )
        """))

async def fetch_stories() -> pd.DataFrame:
    url = "https://hacker-news.firebaseio.com/v0/topstories.json"
    async with httpx.AsyncClient(timeout=10) as client:
        ids = (await client.get(url)).json()[:20]
        async def get_one(sid):
            r = await client.get(f"https://hacker-news.firebaseio.com/v0/item/{sid}.json")
            return r.json()
        raw = await asyncio.gather(*(get_one(i) for i in ids))
    df = pd.DataFrame(raw)[["id", "title", "url"]].dropna(subset=["title"])
    df["fetched_at"] = pd.Timestamp.utcnow().isoformat()
    return df

client = AsyncAnthropic()
sem = asyncio.Semaphore(5)

async def classify(title: str) -> str:
    async with sem:
        try:
            r = await client.messages.create(
                model="claude-haiku-4-5",   # cheaper for bulk classification
                max_tokens=10,
                messages=[{
                    "role": "user",
                    "content": f"Classify sentiment of this headline as exactly one word: positive, negative, or neutral. No other text.\n\nHeadline: {title}"
                }],
            )
            word = r.content[0].text.strip().lower()
            return word if word in {"positive", "negative", "neutral"} else "neutral"
        except Exception as e:
            print(f"classify failed: {e}")
            return "unknown"

async def enrich(df: pd.DataFrame) -> pd.DataFrame:
    sentiments = await asyncio.gather(*(classify(t) for t in df["title"]))
    df = df.copy()
    df["sentiment"] = sentiments
    return df

def upsert(df: pd.DataFrame) -> int:
    sql = text("""
        INSERT INTO stories (id, title, url, fetched_at, sentiment)
        VALUES (:id, :title, :url, :fetched_at, :sentiment)
        ON CONFLICT(id) DO UPDATE SET
            sentiment = excluded.sentiment,
            fetched_at = excluded.fetched_at
    """)
    rows = df.to_dict(orient="records")
    with engine.begin() as conn:
        conn.execute(sql, rows)
    return len(rows)

async def main():
    init_db()
    df = await fetch_stories()
    df = await enrich(df)
    n = upsert(df)
    print(f"Inserted/updated {n} rows.")
    # Read back top negatives
    with engine.connect() as conn:
        neg = pd.read_sql("SELECT title FROM stories WHERE sentiment = 'negative' LIMIT 5", conn)
    print("\nRecent negative headlines:")
    print(neg.to_string(index=False))

asyncio.run(main())