Spaces:
Running
Running
File size: 3,165 Bytes
1f5f31a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """
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}")
|