Build It: Python Web Scraper With AI-Parsed Output

# pyproject.toml dependencies # httpx, beautifulsoup4, anthropic, pydantic import asyncio import httpx from bs4 import BeautifulSoup from pydantic import BaseModel from anthropic import AsyncAnthropic class Story(BaseModel): rank: int title: str url: str | None points: int comments: int async def fetch_html(url: str) -> str: headers = {"User-Agent": "tendril-scraper/1.0"} async with httpx.AsyncClient(headers=headers, timeout=10) as client: r = await client.get(url) r.raise_for_status() return r.text

def extract_text_blocks(html: str) -> list[str]: soup = BeautifulSoup(html, "html.parser") # HN wraps stories in <tr class="athing">; we'll just get the visible text of each row blocks = [] for row in soup.select("tr.athing"): sibling = row.find_next_sibling("tr") text = row.get_text(" ", strip=True) if sibling: text += " " + sibling.get_text(" ", strip=True) link_tag = row.select_one(".titleline > a") url = link_tag.get("href") if link_tag else None blocks.append(f"URL={url}\n{text}") return blocks

client = AsyncAnthropic() async def parse_story(block: str) -> Story | None: prompt = f"""Extract fields from this Hacker News row as JSON. Fields: rank (int), title (str), url (str or null), points (int), comments (int). Return ONLY valid JSON, no preface. <row> {block} </row>""" try: response = await client.messages.create( model="claude-opus-4-7", max_tokens=300, messages=[{"role": "user", "content": prompt}], ) raw = response.content[0].text.strip() # strip potential code-fence if raw.startswith("```"): raw = raw.strip("`").split("\n", 1)[1].rsplit("\n", 1)[0] return Story.model_validate_json(raw) except Exception as e: print(f"parse failed: {e}") return None async def main(): html = await fetch_html("https://news.ycombinator.com/") blocks = extract_text_blocks(html)[:10] stories = await asyncio.gather(*(parse_story(b) for b in blocks)) for s in filter(None, stories): print(f"{s.rank:2}. {s.title} ({s.points} pts, {s.comments} comments)") asyncio.run(main())

import hashlib import json from pathlib import Path CACHE = Path(".cache") CACHE.mkdir(exist_ok=True) async def parse_story_cached(block: str) -> Story | None: key = hashlib.sha256(block.encode()).hexdigest()[:16] cache_file = CACHE / f"{key}.json" if cache_file.exists(): return Story.model_validate_json(cache_file.read_text()) result = await parse_story(block) if result: cache_file.write_text(result.model_dump_json()) return result