curated_feed.fetch ================== .. py:module:: curated_feed.fetch .. autoapi-nested-parse:: Fetch subscribed feeds and append normalised records to the corpus. One JSONL file per collection day, one record per item. Records are deduplicated against every corpus file already on disk (by `id`), so the command can run daily and accumulate without repeating items. A single broken feed never fails the run. Attributes ---------- .. autoapisummary:: curated_feed.fetch.TRACKING_PARAM_PREFIXES curated_feed.fetch.TRACKING_PARAMS curated_feed.fetch.SPACE_BEFORE_PUNCTUATION curated_feed.fetch.DEFAULT_PORTS curated_feed.fetch.ID_LENGTH curated_feed.fetch.RETRY_STATUS_CODES curated_feed.fetch.MAX_RETRY_AFTER_SECONDS Classes ------- .. autoapisummary:: curated_feed.fetch.FeedStats curated_feed.fetch.FetchResult curated_feed.fetch.RunSummary Functions --------- .. autoapisummary:: curated_feed.fetch.is_tracking_param curated_feed.fetch.canonicalize_url curated_feed.fetch.make_id curated_feed.fetch.html_to_text curated_feed.fetch.truncate curated_feed.fetch.entry_published curated_feed.fetch.build_record curated_feed.fetch.records_from_feed curated_feed.fetch.sort_records curated_feed.fetch.fetch_one curated_feed.fetch.retry_delay curated_feed.fetch.build_client curated_feed.fetch.fetch_all curated_feed.fetch.collect curated_feed.fetch.format_summary curated_feed.fetch.build_parser curated_feed.fetch.main Module Contents --------------- .. py:data:: TRACKING_PARAM_PREFIXES :value: ('utm_', 'at_', 'pk_', 'hsa_', 'vero_', '_hs', 'mc_') .. py:data:: TRACKING_PARAMS .. py:data:: SPACE_BEFORE_PUNCTUATION .. py:data:: DEFAULT_PORTS .. py:data:: ID_LENGTH :value: 16 .. py:data:: RETRY_STATUS_CODES .. py:data:: MAX_RETRY_AFTER_SECONDS :value: 30.0 .. py:class:: FeedStats Per-feed accounting, aggregated into the run summary. .. py:attribute:: items_seen :type: int :value: 0 .. py:attribute:: kept :type: int :value: 0 .. py:attribute:: skipped_outside_window :type: int :value: 0 .. py:attribute:: skipped_no_url :type: int :value: 0 .. py:attribute:: missing_date :type: int :value: 0 .. py:class:: FetchResult Outcome of fetching one feed. .. py:attribute:: feed :type: curated_feed.opml.Feed .. py:attribute:: content :type: bytes | None :value: None .. py:attribute:: error :type: str | None :value: None .. py:class:: RunSummary Everything the end-of-run report needs. .. py:attribute:: output_path :type: pathlib.Path .. py:attribute:: attempted :type: int :value: 0 .. py:attribute:: succeeded :type: int :value: 0 .. py:attribute:: failures :type: list[tuple[curated_feed.opml.Feed, str]] :value: [] .. py:attribute:: items_seen :type: int :value: 0 .. py:attribute:: skipped_outside_window :type: int :value: 0 .. py:attribute:: skipped_no_url :type: int :value: 0 .. py:attribute:: skipped_duplicate :type: int :value: 0 .. py:attribute:: missing_date :type: int :value: 0 .. py:attribute:: written :type: int :value: 0 .. py:function:: is_tracking_param(name: str) -> bool .. py:function:: canonicalize_url(url: str) -> str Normalise a URL for identity: lowercase host, no fragment, no tracking. .. py:function:: make_id(canonical_url: str) -> str Stable id for an item: a hash of its canonical URL. Feed GUIDs are deliberately ignored — they are inconsistent across sources and sometimes change between fetches of the same article. .. py:function:: html_to_text(value: str) -> str Strip markup and collapse whitespace. .. py:function:: truncate(text: str, limit: int) -> str Truncate on a word boundary, marking the cut with an ellipsis. .. py:function:: entry_published(entry: dict) -> datetime.datetime | None Best available publication timestamp, or None if the feed gave none. .. py:function:: build_record(entry: dict, *, feed: curated_feed.opml.Feed, source_title: str, fetched_at: datetime.datetime, summary_max_chars: int) -> curated_feed.models.Candidate | None Normalise one feed entry into a corpus record, or None if unusable. .. py:function:: records_from_feed(content: bytes, feed: curated_feed.opml.Feed, *, fetched_at: datetime.datetime, cutoff: datetime.datetime | None, summary_max_chars: int) -> tuple[list[curated_feed.models.Candidate], FeedStats, str | None] Parse feed bytes into records. Returns the records, per-feed stats, and a parse error for feeds that yielded nothing usable. Items with no usable date are kept and flagged rather than dropped: feeds are inconsistent about dates and silence is not evidence. .. py:function:: sort_records(records: collections.abc.Iterable[curated_feed.models.Candidate]) -> list[curated_feed.models.Candidate] Newest first, undated items last, ties broken by title for determinism. .. py:function:: fetch_one(client: httpx.AsyncClient, feed: curated_feed.opml.Feed, *, max_retries: int) -> FetchResult :async: Fetch one feed, retrying transient failures with a short backoff. .. py:function:: retry_delay(response: httpx.Response, attempt: int) -> float How long to wait before retrying, honouring `Retry-After` when it is sane. Public because the paywall stage retries on the same terms, and two copies of a backoff policy is one more than anybody will remember to keep in step. .. py:function:: build_client(settings: curated_feed.config.FetchSettings) -> httpx.AsyncClient The client one run makes its requests with. A factory rather than a client built inline, so that this module's one external dependency can be substituted the way every other one in `src/` can — `deploy.py` does the same for the command it runs. A test serves the subscription list's feeds through a mock transport and the whole stage runs without a socket. >>> build_client(FetchSettings()).headers["user-agent"] 'curated-feed/0.1' .. py:function:: fetch_all(feeds: collections.abc.Sequence[curated_feed.opml.Feed], config: curated_feed.config.Config) -> list[FetchResult] :async: Fetch every feed with bounded concurrency. .. py:function:: collect(feeds: collections.abc.Sequence[curated_feed.opml.Feed], config: curated_feed.config.Config, *, now: datetime.datetime) -> RunSummary Fetch, normalise, deduplicate and write one day of records. .. py:function:: format_summary(summary: RunSummary) -> str Human-readable run report. .. py:function:: build_parser() -> argparse.ArgumentParser .. py:function:: main(argv: collections.abc.Sequence[str] | None = None) -> int