feed.sources

Documentation for eth_defi.feed.sources Python module.

Feed source mappings.

Load and validate YAML-defined mappings for RSS feeds, Twitter/X usernames, LinkedIn company identifiers, and feeder websites that should be tracked for vault-related post collection.

Functions

`auto_disable_failed_linkedin_sources`(...)	Write `linkedin-rss-hub-disabled-at` to YAML for every all-503 LinkedIn failure.
`load_feeder_metadata`(yaml_path)	Load a single feeder YAML file and return its metadata as a plain dict.
`load_post_sources`([mappings_dir])	Load and validate all feed source mappings.
`mark_linkedin_source_disabled`(yaml_path, ...)	Append `linkedin-rss-hub-disabled-at` to a feeder YAML without rewriting it.
`mark_rss_source_dead`(yaml_path, dead_at)	Append `rss-dead-at` to a feeder YAML without rewriting it.
`mark_rss_source_failure`(yaml_path, failure_at)	Stamp RSS failure fields on a feeder YAML.
`mark_twitter_handle_unknown`(yaml_path, ...)	Append `twitter-handle-resolved-unknown-at` to a feeder YAML.
`mark_twitter_source_dead`(yaml_path, dead_at)	Append `twitter-dead-at` to a feeder YAML without rewriting it.
`resolve_canonical_feeder_yaml`(...)	Find the YAML file for a canonical feeder, searching by role priority.
`resolve_feeder_id`(feeder_id, aliases)	Resolve a feeder_id through the alias mapping.

Classes

`FeederAlias`	A feeder YAML that delegates feed sources to another feeder.
`TrackedPostSource`	A single logical source mapping for feed collection.

class FeederAlias

Bases: object

A feeder YAML that delegates feed sources to another feeder.

Alias files contain only identity metadata (feeder-id, name, role) and a canonical-feeder-id pointing to the feeder whose feed sources should be collected. They produce no TrackedPostSource entries.

__init__(feeder_id, canonical_feeder_id, role, name, mapping_file)

Parameters

feeder_id (str) –
canonical_feeder_id (str) –
role (str) –
name (str) –
mapping_file (pathlib.Path) –

Return type

None

class TrackedPostSource

Bases: object

A single logical source mapping for feed collection.

__init__(feeder_id, name, role, website, source_type, source_key, canonical_url, mapping_file)

Parameters

feeder_id (str) –
name (str) –
role (str) –
website (Optional[str]) –
source_type (str) –
source_key (str) –
canonical_url (str) –
mapping_file (pathlib.Path) –

Return type

None

get_logical_key()

Return the natural unique key for this source.

Return type: tuple[str, str, str, str]

auto_disable_failed_linkedin_sources(summary, sources, disabled_at)

Write linkedin-rss-hub-disabled-at to YAML for every all-503 LinkedIn failure.

When all configured RSSHub bridges return HTTP 503 for a LinkedIn source, LinkedIn is gating that company page behind authentication. This function stamps the feeder YAML so future scan runs skip the source entirely rather than retrying.

Parameters

summary (CollectorRunSummary) – Collector run result from collect_posts().
sources (collections.abc.Sequence[eth_defi.feed.sources.TrackedPostSource]) – Source list passed to the same collect_posts() call.
disabled_at (str) – ISO date string to stamp in the YAML, e.g. 2026-04-04.

Returns

Number of YAML files updated.

Return type

int

load_feeder_metadata(yaml_path)

Load a single feeder YAML file and return its metadata as a plain dict.

Uses the shared feeder schema for validation, including slug format validation via _validate_slug() and role validation via _validate_role(). Asserts that feeder-id matches the filename stem to catch slug/filename mismatches.

This function provides a public API for modules that need feeder metadata (slug, name, descriptions, website, twitter, linkedin, rss) without pulling in the full feed-collection machinery of load_post_sources().

Parameters: yaml_path (pathlib.Path) – Path to a feeder YAML file.
Returns: Dict with keys: feeder-id, name, role, website, short_description, long_description, twitter, linkedin, rss, etc.
Return type: dict

load_post_sources(mappings_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/feeds'))

Load and validate all feed source mappings.

Returns: Tuple of (sources, feeders_skipped, aliases) where feeders_skipped counts YAML files with all sources disabled (not aliases) and aliases lists feeders that delegate their sources to a canonical feeder via canonical-feeder-id.
Parameters: mappings_dir (pathlib.Path) –
Return type: tuple[list[eth_defi.feed.sources.TrackedPostSource], int, list[eth_defi.feed.sources.FeederAlias]]

mark_linkedin_source_disabled(yaml_path, disabled_at)

Append linkedin-rss-hub-disabled-at to a feeder YAML without rewriting it.

Preserves existing file content including comments. The new field is appended as a trailing line so that YAML structure and indentation are not disturbed.

Parameters

yaml_path (pathlib.Path) – Path to the feeder YAML file.
disabled_at (str) – ISO date string to stamp, e.g. 2026-04-04.

Returns

True when the file was updated, False when the field was already present.

Return type

bool

mark_rss_source_dead(yaml_path, dead_at)

Append rss-dead-at to a feeder YAML without rewriting it.

Used when an RSS feed is valid but has not published any new posts for a year or more.

Parameters

yaml_path (pathlib.Path) – Path to the feeder YAML file.
dead_at (str) – ISO date string, e.g. 2026-04-06.

Returns

True when the file was updated, False when already present.

Return type

bool

mark_rss_source_failure(yaml_path, failure_at, status_code=None, exception_message=None)

Stamp RSS failure fields on a feeder YAML.

Records the most recent RSS failure so operators can see which feeds are broken. Overwrites any previous failure fields.

Parameters

yaml_path (pathlib.Path) – Path to the feeder YAML file.
failure_at (str) – ISO date string, e.g. 2026-04-06.
status_code (Optional[int]) – HTTP status code, or None for non-HTTP failures.
exception_message (Optional[str]) – Exception message or HTTP status text.

Returns

True when the file was updated.

Return type

bool

mark_twitter_handle_unknown(yaml_path, unknown_at)

Append twitter-handle-resolved-unknown-at to a feeder YAML.

Used when the X API cannot resolve a Twitter handle to a user ID, typically because the account has been suspended, deleted, or renamed.

Parameters

yaml_path (pathlib.Path) – Path to the feeder YAML file.
unknown_at (str) – ISO date string, e.g. 2026-04-06.

Returns

True when the file was updated, False when already present.

Return type

bool

mark_twitter_source_dead(yaml_path, dead_at)

Append twitter-dead-at to a feeder YAML without rewriting it.

Used when a Twitter account has not posted for the configured death detection period.

Parameters

yaml_path (pathlib.Path) – Path to the feeder YAML file.
dead_at (str) – ISO date string to stamp, e.g. 2026-04-04.

Returns

True when the file was updated, False when the field was already present.

Return type

bool

resolve_canonical_feeder_yaml(canonical_feeder_id, mappings_dir)

Find the YAML file for a canonical feeder, searching by role priority.

Searches subdirectories in priority order: stablecoins/, protocols/, curators/, vaults/. Returns the path to the first matching YAML file whose filename stem equals canonical_feeder_id.

Parameters

canonical_feeder_id (str) – The feeder-id slug to look up.
mappings_dir (pathlib.Path) – Root directory containing role subdirectories (e.g. eth_defi/data/feeds).

Raises

FileNotFoundError – When no matching YAML file is found in any role subdirectory.

Return type

pathlib.Path

resolve_feeder_id(feeder_id, aliases)

Resolve a feeder_id through the alias mapping.

Returns the canonical feeder_id when the input is an alias, or the input unchanged when it is not an alias.

Parameters

feeder_id (str) – Feeder-id to resolve.
aliases (collections.abc.Sequence[eth_defi.feed.sources.FeederAlias]) – Alias list returned by load_post_sources().

Return type

str