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.

Module Attributes

`KNOWN_FEEDER_ROLES`	Supported feeder roles.
`ROLE_SUBDIRS`	Role subdirectory names under the feeds data directory.
`ROLE_PRIORITY`	Role priority order for canonical feeder resolution.

Functions

`auto_disable_failed_linkedin_sources`(...)	Write `linkedin-rss-hub-disabled-at` to YAML for every all-503 LinkedIn failure.
`build_twitter_source_file_lookup`(twitter_sources)	Build a lower-case Twitter handle to feeder YAML file lookup.
`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_suspended_twitter_handle`(handle, ...)	Mark a suspended X account as unavailable in feeder YAML files.
`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.

FEEDS_DATA_DIR = PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/feeds'): Repository-resolved base directory for feed source YAML files.

POST_TRACKING_DATA_DIR = PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/feeds'): Backwards-compatible alias for the old constant name.

KNOWN_FEEDER_ROLES = {'curator', 'protocol', 'stablecoin', 'vault'}: Supported feeder roles.

ROLE_SUBDIRS = {'curator': 'curators', 'protocol': 'protocols', 'stablecoin': 'stablecoins', 'vault': 'vaults'}: Role subdirectory names under the feeds data directory.

ROLE_PRIORITY = ('stablecoin', 'protocol', 'curator', 'vault'): Role priority order for canonical feeder resolution. When a canonical-feeder-id matches files in multiple roles, the role listed first wins.

class TrackedPostSource

Bases: object

A single logical source mapping for feed collection.

feeder_id: str: Canonical slug for the feeder, matching the curator, protocol, stablecoin, or vault slug.

name: str: Human-readable feeder name shown in diagnostics and exports.

role: str: Feeder role such as protocol, curator, stablecoin, or vault.

website: Optional[str]: Company website for the feeder when configured in YAML.

source_type: str: Source transport type, currently rss, twitter, or linkedin.

source_key: str: Source-specific stable key, such as feed URL, Twitter username, or LinkedIn company id.

canonical_url: str: Canonical source URL used for fetching or bridge construction.

mapping_file: pathlib.Path: Path to the YAML file that defined this source.

get_logical_key()

Return the natural unique key for this source.

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

__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

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.

feeder_id: str: Feeder-id of the alias file.

canonical_feeder_id: str: Feeder-id of the canonical source-bearing feeder.

role: str: Role declared in the alias file (may differ from the canonical feeder’s role).

name: str: Human-readable name from the alias file.

mapping_file: pathlib.Path: Path to the alias YAML file.

__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

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_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

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

build_twitter_source_file_lookup(twitter_sources)

Build a lower-case Twitter handle to feeder YAML file lookup.

Twitter sources are already normalised by _normalise_twitter_source(), but this helper keeps callback users defensive and centralises the repeated mapping logic.

Parameters: twitter_sources (list[eth_defi.feed.sources.TrackedPostSource]) – Twitter sources loaded from feeder YAML files.
Returns: Mapping from lower-case handle to the YAML files that define it.
Return type: dict[str, list[pathlib.Path]]

mark_suspended_twitter_handle(handle, user_id, source_files_by_handle, today_str)

Mark a suspended X account as unavailable in feeder YAML files.

X can resolve suspended accounts to user IDs, but reject adding them to a list. We reuse the existing twitter-handle-resolved-unknown-at marker because it disables the source until an operator verifies that the account has been restored or repointed.

Parameters

handle (str) – Twitter/X handle that failed list membership.
user_id (str) – Numeric X user ID returned by handle resolution.
source_files_by_handle (dict[str, list[pathlib.Path]]) – Mapping from lower-case handle to feeder YAML files.
today_str (str) – ISO date string to write to the marker.

Return type

None

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

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_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]]

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

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, curatorwatch, 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, curatorwatch, short_description, long_description, twitter, linkedin, rss, etc.
Return type: dict