spb/ai-atlas

1"""NVIDIA — corporate blog RSS + technical (developer) blog Atom feed → ANNOUNCEMENT events."""2from __future__ import annotations34from aiatlas.registry import org_ref5from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext6from aiatlas.sdk.facts import Facts, Target7from aiatlas.sdk.fetch import FetchResult89from ._common import announcement_events1011BLOG_RSS = "https://blogs.nvidia.com/feed/"12DEVELOPER_RSS = "https://developer.nvidia.com/blog/feed"131415class NvidiaConnector(BaseConnector):16    name = "nvidia"17    label = "NVIDIA — blog & technical blog feeds"18    description = "NVIDIA corporate blog (RSS) and NVIDIA Technical Blog (Atom): announcements, model and platform releases."19    source_key = "blogs.nvidia.com"20    version = "1"21    parser_version = "1"22    interval_seconds = 720023    min_interval_seconds = 360024    rate_per_min = 1025    tier = 126    priority = 127    expected_min_records = 128    concurrency = 22930    async def discover(self, ctx: RunContext) -> list[Target]:31        return [32            Target(url=BLOG_RSS, doc_type="feed", key="blog", min_bytes=1000),33            Target(url=DEVELOPER_RSS, doc_type="feed", key="developer", min_bytes=1000),34        ]3536    async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:37        facts = Facts()38        org = org_ref("nvidia")39        facts.entities.append(org)40        if parsed.kind != "feed":41            return facts42        source = "developer.nvidia.com/blog" if target.key == "developer" else "blogs.nvidia.com"43        announcement_events(facts, org, parsed.feed_items, source_name=source, max_follow=10)44        facts.document_title = "NVIDIA Technical Blog" if target.key == "developer" else "NVIDIA Blog"45        facts.document_entity = org46        return facts474849CONNECTORS = [NvidiaConnector]50