spb/company-atlas

1"""RSS / Atom / JSON Feed connector → news items (title, url, published_at, summary, language). One block per entry keyed by the2entry's canonical URL / id so the block diff equals the entry delta. Uses feedparser (bounded input)."""3from __future__ import annotations45import hashlib6from collections.abc import Mapping7from typing import Any89import feedparser1011from companyatlas.connectors._util import parse_date, strip_html12from companyatlas.fetch import FetchResult13from companyatlas.sdk.connector import Connector, ConnectorMeta, register14from companyatlas.sdk.models import Block, ExtractedNewsItem, Extraction15from companyatlas.sdk.normalize import language_guess, normalized_text, simhash16from companyatlas.taxonomy import FetchMode, Surface17from companyatlas.urls import canonicalize_url1819MAX_ENTRIES = 20020CATEGORY_BY_SURFACE = {Surface.NEWSROOM: "press", Surface.BLOG: "blog", Surface.CHANGELOG: "changelog", Surface.RESEARCH: "research",21                       Surface.INVESTOR_RELATIONS: "ir", Surface.FEED: "blog"}222324def _json_feed_items(data: dict[str, Any]) -> list[ExtractedNewsItem]:25    out: list[ExtractedNewsItem] = []26    for it in (data.get("items") or [])[:MAX_ENTRIES]:27        if not isinstance(it, dict) or not it.get("title") or not it.get("url"):28            continue29        summary = strip_html(it.get("summary") or it.get("content_text") or it.get("content_html") or "")[:600] or None30        out.append(ExtractedNewsItem(title=str(it["title"])[:300], url=str(it["url"]), published_at=parse_date(it.get("date_published")),31                                     summary=summary, language=(it.get("language") or data.get("language") or None)))32    return out333435@register36class FeedConnector(Connector):37    meta = ConnectorMeta(connector_id="feed-v1", name="RSS / Atom / JSON feed", version="1", category=Surface.FEED, fetch_mode=FetchMode.FEED,38                         default_interval_s=2 * 3600, surfaces=(),      # newsroom/blog/changelog HTML pages stay with generic-html; feeds win by URL or surface39                         url_pattern=r"(/(feed|rss|atom|feeds)(\.xml|\.json|/|$)|\.(rss|atom)$|/rss\.xml|/feed\.xml|/atom\.xml|/index\.xml$|feed\.json$)",40                         priority=45, accept="application/rss+xml,application/atom+xml,application/feed+json,application/xml,text/xml;q=0.9,*/*;q=0.5",41                         description="Syndication feeds → news items")4243    def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:44        surface = str(sensor.get("surface") or Surface.FEED)45        category = (sensor.get("config") or {}).get("category") or CATEGORY_BY_SURFACE.get(surface, "other")  # type: ignore[arg-type]46        items: list[ExtractedNewsItem] = []47        title: str | None = None48        lang: str | None = None49        if result.is_json:50            data = result.json()51            if isinstance(data, dict):52                items = _json_feed_items(data)53                title = data.get("title")54                lang = data.get("language")55        else:56            parsed = feedparser.parse(result.content[: 8 * 1024 * 1024])57            if parsed.bozo and not parsed.entries:58                raise ValueError(f"unparseable feed: {getattr(parsed, 'bozo_exception', 'unknown error')}")59            title = parsed.feed.get("title")60            lang = parsed.feed.get("language")61            for e in parsed.entries[:MAX_ENTRIES]:62                link = e.get("link") or next((ln.get("href") for ln in e.get("links", []) if ln.get("href")), None)63                etitle = (e.get("title") or "").strip()64                if not link or not etitle:65                    continue66                published = e.get("published") or e.get("updated") or e.get("created")67                summary = strip_html(e.get("summary") or (e.get("content") or [{}])[0].get("value", ""))[:600] or None68                items.append(ExtractedNewsItem(title=etitle[:300], url=link, published_at=parse_date(published), summary=summary,69                                               language=(e.get("language") or lang or None)))70        for it in items:71            it.category = category72            if not it.language:73                it.language = language_guess(f"{it.title} {it.summary or ''}")74        blocks: list[Block] = []75        seen: set[str] = set()76        lines: list[str] = []77        for i, it in enumerate(items):78            canon = canonicalize_url(it.url)79            if canon in seen:80                continue81            seen.add(canon)82            text = f"{it.title}\n{it.summary or ''}".strip()83            lines.append(it.title)84            blocks.append(Block(key=f"entry:{hashlib.blake2b(canon.encode('utf-8'), digest_size=8).hexdigest()}", kind="news_item", text=text, path=title or "Feed",85                                hash=hashlib.sha256(normalized_text(text).encode()).hexdigest()[:16], simhash=simhash(text), weight=1.2, order=i,86                                attrs={"url": canon, "published_at": it.published_at.isoformat() if it.published_at else None}))87        meta = {"feed_title": title, "entry_count": len(items), "category": category, "structured": True}88        return Extraction(text="\n".join(lines), blocks=blocks, title=title, language=(lang or "").split("-")[0] or None, meta=meta, news=items)899091__all__ = ["FeedConnector"]92