capcat.core.source_system.source_registry

File: Application/capcat/core/source_system/source_registry.py

Description

Source registry for automatic discovery and management of news sources. Handles both config-driven and custom source types.

Classes

SourceRegistry

Registry for managing and discovering news sources.

Supports both config-driven sources (YAML/JSON) and custom source implementations. Provides auto-discovery and validation of sources.

Methods

init
def __init__(self, sources_dir: str = None, project_root: Path = None)

Initialize the source registry.

Args: sources_dir: Deprecated. Direct path to sources directory. project_root: Project root for user source discovery. Builtin sources come from the installed package; user sources come from /Config/sources/active/.

Parameters:

  • self
  • sources_dir (str) optional
  • project_root (Path) optional
discover_sources
def discover_sources(self) -> Dict[str, SourceConfig]

Discover all available sources (both config-driven and custom).

Scans builtin sources first, then user sources. User sources with the same name/ID as a builtin source take precedence.

Returns: Dictionary mapping source names to their configurations

Raises: SourceError: If source discovery fails

Parameters:

  • self

Returns: Dict[str, SourceConfig]

_discover_config_driven_sources
def _discover_config_driven_sources(self)

Discover sources defined by configuration files.

Parameters:

  • self
_discover_custom_sources
def _discover_custom_sources(self)

Discover custom source implementations.

Parameters:

  • self
_load_custom_source
def _load_custom_source(self, source_dir: Path)

Load a custom source implementation.

Parameters:

  • self
  • source_dir (Path)

⚠️ High complexity: 12

_load_config_file
def _load_config_file(self, config_file: Path) -> Dict

Load a configuration file (YAML or JSON).

Parameters:

  • self
  • config_file (Path)

Returns: Dict

_validate_config_driven_config
def _validate_config_driven_config(self, config: SourceConfig) -> List[str]

Validate configuration for config-driven sources.

Parameters:

  • self
  • config (SourceConfig)

Returns: List[str]

get_source
def get_source(self, source_name: str, session = None) -> BaseSource

Get a source instance by name.

Args: source_name: Name of the source session: Optional HTTP session

Returns: BaseSource instance

Raises: SourceError: If source is not found or cannot be instantiated

Parameters:

  • self
  • source_name (str)
  • session optional

Returns: BaseSource

get_source_for_url
def get_source_for_url(self, url: str) -> Optional[Tuple[BaseSource, str]]

Return (source_instance, source_id) for the first registered custom source whose can_handle_url() classmethod matches the given URL.

Returns None if no source claims the URL.

Parameters:

  • self
  • url (str)

Returns: Optional[Tuple[BaseSource, str]]

can_handle_url
def can_handle_url(self, url: str) -> bool

Return True if any registered custom source can handle the URL.

Parameters:

  • self
  • url (str)

Returns: bool

get_available_sources
def get_available_sources(self) -> List[str]

Get list of available source names.

Parameters:

  • self

Returns: List[str]

is_builtin_source
def is_builtin_source(self, source_id: str) -> bool

Return True if source_id was discovered from the builtin package path.

Builtin sources cannot be removed via the TUI - they are shipped with the application. User-added sources (or user overrides of builtins) are not in this set and are safe to remove.

Parameters:

  • self
  • source_id (str)

Returns: bool

get_sources_by_category
def get_sources_by_category(self, category: str) -> List[str]

Get non-hidden sources by category.

Parameters:

  • self
  • category (str)

Returns: List[str]

get_source_config
def get_source_config(self, source_name: str) -> Optional[SourceConfig]

Get source configuration by name.

Parameters:

  • self
  • source_name (str)

Returns: Optional[SourceConfig]

validate_all_sources
def validate_all_sources(self, deep_validation: bool = False) -> Dict[str, List[str]]

Validate all registered sources using enhanced validation engine.

Args: deep_validation: Whether to perform deep validation (network tests)

Returns: Dictionary mapping source names to validation errors (empty list if valid)

Parameters:

  • self
  • deep_validation (bool) optional

Returns: Dict[str, List[str]]

enhanced_validate_all_sources
def enhanced_validate_all_sources(self, deep_validation: bool = False)

Enhanced validation with detailed results.

Args: deep_validation: Whether to perform deep validation

Returns: Dictionary mapping source names to ValidationResult objects

Parameters:

  • self
  • deep_validation (bool) optional
generate_validation_report
def generate_validation_report(self, deep_validation: bool = False) -> str

Generate a comprehensive validation report.

Args: deep_validation: Whether to perform deep validation

Returns: Human-readable validation report

Parameters:

  • self
  • deep_validation (bool) optional

Returns: str

get_registry_stats
def get_registry_stats(self) -> Dict[str, int]

Get registry statistics.

Parameters:

  • self

Returns: Dict[str, int]

clear_cache
def clear_cache(self)

Clear cached source instances.

Parameters:

  • self
reload_sources
def reload_sources(self)

Reload all sources (useful for development).

Parameters:

  • self

Functions

get_source_registry

def get_source_registry(project_root: Optional[Path] = None) -> SourceRegistry

Get the source registry.

When project_root is None, return the cached global singleton, auto-detecting the project root so user sources in Config/sources/active/ are included. When project_root is non-None, always construct and return a fresh instance (singleton bypass - avoids stale-singleton bugs with user-overridden sources).

Parameters:

  • project_root (Optional[Path]) optional

Returns: SourceRegistry

reset_source_registry

def reset_source_registry()

Reset the global source registry (useful for testing).