""" Fetches raw items from a single source. Two strategies (ADR-0015): - 'rss': feedparser, for the national/economic news sources that actually publish a feed. - 'html': a generic title + first-substantial-paragraph extractor, for government bulletins and funder/program pages that don't publish RSS. Expect a higher needs_review rate from these by design — the completeness bar in classify.py catches weak extractions rather than showing them to users. Network errors for one source must never take down a whole poll run — the caller (pipeline.py) treats a per-source failure as "zero items from this source this run", not a fatal error. """ from dataclasses import dataclass from datetime import datetime, timezone from time import mktime import feedparser import httpx from bs4 import BeautifulSoup REQUEST_TIMEOUT = 15.0 USER_AGENT = "VenThriveNewsBot/1.0 (+https://venthrive.com; business-news aggregator)" @dataclass class RawItem: url: str title: str text: str published_at: datetime | None def fetch_rss(feed_url: str) -> list[RawItem]: parsed = feedparser.parse( feed_url, agent=USER_AGENT, request_headers={"User-Agent": USER_AGENT}, ) items: list[RawItem] = [] for entry in parsed.entries: url = entry.get("link", "").strip() title = entry.get("title", "").strip() if not url or not title: continue text = entry.get("summary", "") or entry.get("description", "") text = BeautifulSoup(text, "html.parser").get_text(" ", strip=True) published_at = None if entry.get("published_parsed"): published_at = datetime.fromtimestamp(mktime(entry.published_parsed), tz=timezone.utc) items.append(RawItem(url=url, title=title, text=text, published_at=published_at)) return items def fetch_html(page_url: str) -> list[RawItem]: """One item: the page itself. Government/funder pages are usually a single announcement or a landing page we read as-is — there's no reliable, generic way to discover "articles" on an arbitrary site's markup, so this reads the page's own headline + body rather than trying to crawl links from it.""" resp = httpx.get(page_url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT, follow_redirects=True) resp.raise_for_status() soup = BeautifulSoup(resp.text, "html.parser") title_tag = soup.find("h1") or soup.find("title") title = title_tag.get_text(strip=True) if title_tag else "" paragraphs = [p.get_text(" ", strip=True) for p in soup.find_all("p")] substantial = [p for p in paragraphs if len(p) > 60] text = " ".join(substantial[:5]) if not title or not text: return [] return [RawItem(url=page_url, title=title, text=text, published_at=None)] def fetch_source(source: dict) -> list[RawItem]: if source["source_type"] == "rss": return fetch_rss(source["url"]) if source["source_type"] == "html": return fetch_html(source["url"]) raise ValueError(f"Unknown source_type: {source['source_type']!r}")