saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
"""RSS/Atom feed collector."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time as time_module
|
|
from datetime import UTC, datetime
|
|
from urllib.parse import quote
|
|
|
|
import feedparser
|
|
import httpx
|
|
|
|
from app.collectors.base import (
|
|
CollectedDocument,
|
|
CollectionResult,
|
|
CompanyContext,
|
|
DiscoveredSource,
|
|
SourceConfig,
|
|
)
|
|
from app.collectors.extraction import canonicalize_url, compute_content_hash, normalize_whitespace
|
|
from app.core.config import get_settings
|
|
from app.core.http import FetchError, SsrfBlockedError, fetch_with_retries
|
|
from app.core.logging import get_logger
|
|
from app.models.enums import SourceStatus, SourceType
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class RssCollector:
|
|
source_type = SourceType.RSS
|
|
|
|
async def discover(self, company: CompanyContext) -> list[DiscoveredSource]:
|
|
# Google News' search RSS endpoint needs no API key and reliably
|
|
# exists for any query - unlike a company's own press-room feed
|
|
# (which would need a search provider to locate), this one URL
|
|
# formula works for every company and already aggregates wire-
|
|
# service releases (PRNewswire/BusinessWire/GlobeNewswire) as
|
|
# they're published, so a dedicated wire-specific collector isn't
|
|
# needed on top of it. Users can still add any other feed manually
|
|
# (see custom_url.py's sibling "add any public URL" path).
|
|
query_url = (
|
|
f"https://news.google.com/rss/search?q={quote(company.name)}&hl=en-US&gl=US&ceid=US:en"
|
|
)
|
|
return [
|
|
DiscoveredSource(
|
|
source_type=SourceType.RSS,
|
|
name=f"{company.name} — Google News",
|
|
base_url=query_url,
|
|
)
|
|
]
|
|
|
|
async def collect(self, source: SourceConfig, company: CompanyContext) -> CollectionResult:
|
|
if not source.base_url:
|
|
return CollectionResult(status=SourceStatus.FAILED, error="No feed URL configured")
|
|
|
|
settings = get_settings()
|
|
try:
|
|
result = await fetch_with_retries(source.base_url, settings=settings)
|
|
except SsrfBlockedError as exc:
|
|
return CollectionResult(status=SourceStatus.BLOCKED_BY_POLICY, error=str(exc))
|
|
except (FetchError, httpx.HTTPError) as exc:
|
|
return CollectionResult(status=SourceStatus.FAILED, error=str(exc))
|
|
|
|
if result.status_code >= 400:
|
|
return CollectionResult(
|
|
status=SourceStatus.FAILED, error=f"HTTP {result.status_code}", pages_attempted=1
|
|
)
|
|
|
|
parsed = feedparser.parse(result.content)
|
|
if parsed.bozo and not parsed.entries:
|
|
return CollectionResult(
|
|
status=SourceStatus.FAILED,
|
|
error=str(parsed.get("bozo_exception", "Unparseable feed")),
|
|
pages_attempted=1,
|
|
)
|
|
|
|
max_items = source.configuration_metadata.get("max_items", 20)
|
|
documents: list[CollectedDocument] = []
|
|
for entry in parsed.entries[:max_items]:
|
|
link = entry.get("link")
|
|
if not link:
|
|
continue
|
|
summary = entry.get("summary", "") or entry.get("description", "")
|
|
text = normalize_whitespace(f"{entry.get('title', '')}\n\n{summary}")
|
|
if not text:
|
|
continue
|
|
|
|
pub_date = None
|
|
if entry.get("published_parsed"):
|
|
pub_date = datetime.fromtimestamp(
|
|
time_module.mktime(entry.published_parsed), tz=UTC
|
|
)
|
|
|
|
documents.append(
|
|
CollectedDocument(
|
|
url=link,
|
|
canonical_url=canonicalize_url(link),
|
|
title=entry.get("title"),
|
|
author=entry.get("author"),
|
|
publication_date=pub_date,
|
|
retrieved_date=datetime.now(UTC),
|
|
content_text=text,
|
|
content_hash=compute_content_hash(text),
|
|
metadata={"feed_url": source.base_url},
|
|
extraction_method="feedparser",
|
|
http_status=result.status_code,
|
|
trust_score=0.6,
|
|
)
|
|
)
|
|
|
|
status = SourceStatus.ACTIVE if documents else SourceStatus.FAILED
|
|
return CollectionResult(status=status, documents=documents, pages_attempted=1)
|