Spaces:
Running
Running
| """ | |
| 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)" | |
| 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}") | |