curated_feed.fetch

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

TRACKING_PARAM_PREFIXES
TRACKING_PARAMS
SPACE_BEFORE_PUNCTUATION
DEFAULT_PORTS
ID_LENGTH
RETRY_STATUS_CODES
MAX_RETRY_AFTER_SECONDS

Classes

FeedStats	Per-feed accounting, aggregated into the run summary.
FetchResult	Outcome of fetching one feed.
RunSummary	Everything the end-of-run report needs.

Functions

is_tracking_param(→ bool)
canonicalize_url(→ str)	Normalise a URL for identity: lowercase host, no fragment, no tracking.
make_id(→ str)	Stable id for an item: a hash of its canonical URL.
html_to_text(→ str)	Strip markup and collapse whitespace.
truncate(→ str)	Truncate on a word boundary, marking the cut with an ellipsis.
entry_published(→ datetime.datetime | None)	Best available publication timestamp, or None if the feed gave none.
build_record(→ curated_feed.models.Candidate | None)	Normalise one feed entry into a corpus record, or None if unusable.
records_from_feed(...)	Parse feed bytes into records.
sort_records(→ list[curated_feed.models.Candidate])	Newest first, undated items last, ties broken by title for determinism.
fetch_one(→ FetchResult)	Fetch one feed, retrying transient failures with a short backoff.
retry_delay(→ float)	How long to wait before retrying, honouring Retry-After when it is sane.
build_client(→ httpx.AsyncClient)	The client one run makes its requests with.
fetch_all(→ list[FetchResult])	Fetch every feed with bounded concurrency.
collect(→ RunSummary)	Fetch, normalise, deduplicate and write one day of records.
format_summary(→ str)	Human-readable run report.
build_parser(→ argparse.ArgumentParser)
main(→ int)

Module Contents

curated_feed.fetch.TRACKING_PARAM_PREFIXES = ('utm_', 'at_', 'pk_', 'hsa_', 'vero_', '_hs', 'mc_')

curated_feed.fetch.TRACKING_PARAMS

curated_feed.fetch.SPACE_BEFORE_PUNCTUATION

curated_feed.fetch.DEFAULT_PORTS

curated_feed.fetch.ID_LENGTH = 16

curated_feed.fetch.RETRY_STATUS_CODES

curated_feed.fetch.MAX_RETRY_AFTER_SECONDS = 30.0

class curated_feed.fetch.FeedStats

Per-feed accounting, aggregated into the run summary.

items_seen: int = 0

kept: int = 0

skipped_outside_window: int = 0

skipped_no_url: int = 0

missing_date: int = 0

class curated_feed.fetch.FetchResult

Outcome of fetching one feed.

feed: curated_feed.opml.Feed

content: bytes | None = None

error: str | None = None

class curated_feed.fetch.RunSummary

Everything the end-of-run report needs.

output_path: pathlib.Path

attempted: int = 0

succeeded: int = 0

failures: list[tuple[curated_feed.opml.Feed, str]] = []

items_seen: int = 0

skipped_outside_window: int = 0

skipped_no_url: int = 0

skipped_duplicate: int = 0

missing_date: int = 0

written: int = 0

curated_feed.fetch.is_tracking_param(name: str) → bool

curated_feed.fetch.canonicalize_url(url: str) → str

Normalise a URL for identity: lowercase host, no fragment, no tracking.

curated_feed.fetch.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.

curated_feed.fetch.html_to_text(value: str) → str

Strip markup and collapse whitespace.

curated_feed.fetch.truncate(text: str, limit: int) → str

Truncate on a word boundary, marking the cut with an ellipsis.

curated_feed.fetch.entry_published(entry: dict) → datetime.datetime | None

Best available publication timestamp, or None if the feed gave none.

curated_feed.fetch.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.

curated_feed.fetch.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.

curated_feed.fetch.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.

async curated_feed.fetch.fetch_one(client: httpx.AsyncClient, feed: curated_feed.opml.Feed, *, max_retries: int) → FetchResult

Fetch one feed, retrying transient failures with a short backoff.

curated_feed.fetch.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.

curated_feed.fetch.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'

async curated_feed.fetch.fetch_all(feeds: collections.abc.Sequence[curated_feed.opml.Feed], config: curated_feed.config.Config) → list[FetchResult]

Fetch every feed with bounded concurrency.

curated_feed.fetch.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.

curated_feed.fetch.format_summary(summary: RunSummary) → str

Human-readable run report.

curated_feed.fetch.build_parser() → argparse.ArgumentParser

curated_feed.fetch.main(argv: collections.abc.Sequence[str] | None = None) → int