misc-csechet/frontools/config.py

"""Config loading"""
from gettext import gettext as _
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from re import Pattern
from re import compile as re_compile
from typing import Any, Iterable, Optional, Type, Union

from xdg import xdg_config_dirs, xdg_config_home

from frontools.sources import CachedSource, OverrideSource, Source
from frontools.theme_index import ThemeIndex, UrlEntry, NodeEntry

REMOTE_SOURCE_NAME = "remote"


class ConfigError(Exception):
    """Error raised on config error"""


class Config:
    """Configuration object"""

    def __init__(
        self,
        source_name: Optional[str],
        use_cache: bool,
        include_urls: list[str],
        exclude_urls: list[str],
        include_tags: list[str],
        exclude_tags: list[str],
    ):
        self._sources: dict[str, Source] = {}
        self._theme_index = ThemeIndex()
        self._block_urls: list[Pattern[str]] = []

        self._add_source(
            REMOTE_SOURCE_NAME, CachedSource, REMOTE_SOURCE_NAME, not use_cache
        )
        self._source_name = source_name if source_name else REMOTE_SOURCE_NAME

        self._filter = _Filter(
            include_urls,
            exclude_urls,
            include_tags,
            exclude_tags,
        )

    @staticmethod
    async def load(config_path: Optional[Path], update_index: bool, *args: Any, **kwargs: Any) -> "Config":
        """Load config from the given path"""
        config = Config(*args, **kwargs)

        if config_path is None:
            config_path = _find_config()

        if config_path is not None:
            module_spec = spec_from_file_location("frontools.local_config", config_path)
            assert module_spec is not None
            config_module = module_from_spec(module_spec)

            assert module_spec is not None
            module_spec.loader.exec_module(config_module)  # type: ignore

            if not hasattr(config_module, "CONFIG"):
                raise ConfigError(
                    _(
                        f"Configuration file {config_path} should define a root variable CONFIG."
                    )
                )

            config_loader = getattr(config_module, "CONFIG")
            await config_loader(config)
            await config._theme_index.load(update_index)

        return config

    @property
    def source(self) -> Source:
        """get the default source for this context"""
        return self.get_source(self._source_name)

    @property
    def urls(self) -> Iterable[tuple[str, str]]:
        """Return themes configured for this context"""
        for theme, url, tags in self._theme_index.urls:
            if self._filter(url, tags):
                yield theme, url

    def add_urls(self, *urls: UrlEntry) -> None:
        """Add an url for a theme"""
        self._theme_index.add_urls(*urls)

    def add_nodes(self, *nodes: NodeEntry) -> None:
        """Add an url for a theme"""
        self._theme_index.add_nodes(*nodes)

    def add_yaml(self, yaml_path: Union[str, Path]) -> None:
        """Load a yaml file containing dictionnary of urls to add as themes."""
        self._theme_index.add_yaml(Path(yaml_path))

    def block_urls(self, *patterns: str) -> None:
        """Will return 500 error for urls matching this pattern."""
        for pattern in patterns:
            self._block_urls.append(re_compile(pattern))

    def override(
        self,
        source_name: str,
        mappings: list[tuple[str, str]],
        next_source_name: Optional[str] = None,
    ) -> None:
        """Add a source overriding given patterns"""
        assert source_name not in self._sources
        next_source = self.get_source(
            next_source_name if next_source_name else REMOTE_SOURCE_NAME
        )
        self._add_source(source_name, OverrideSource, mappings, next_source)

    def get_source(self, name: str) -> Source:
        """Get an alternate source in the configured ones"""
        if name not in self._sources:
            raise ConfigError(f"No source configured matching {name}")
        return self._sources[name]

    def _add_source(
        self, name: str, source_class: Type[Source], *args: Any, **kwargs: Any
    ) -> None:
        if name in self._sources:
            raise ConfigError(f"Source {name} already configured")
        self._sources[name] = source_class(self._block_urls, *args, **kwargs)


class _Filter:
    def __init__(
        self,
        include_urls: list[str],
        exclude_urls: list[str],
        include_tags: list[str],
        exclude_tags: list[str],
    ):
        self._include_urls = [re_compile(it) for it in include_urls]
        self._exclude_urls = [re_compile(it) for it in exclude_urls]
        self._include_tags = set(include_tags)
        self._exclude_tags = set(exclude_tags)

    def __call__(self, url: str, tags: set[str]) -> bool:
        if self._include_urls:
            if all(not it.match(url) for it in self._include_urls):
                return False

        if self._exclude_urls:
            if any(it.match(url) for it in self._exclude_urls):
                return False

        if self._include_tags and not self._include_tags & tags:
            return False

        if self._exclude_tags and self._exclude_tags & tags:
            return False

        return True


def _find_config() -> Optional[Path]:
    local_config = Path(".frontools.py")
    if local_config.is_file():
        return local_config

    for config_dir in xdg_config_dirs():
        config_path = config_dir / "frontools/config.py"
        if config_path.is_file():
            return config_path

    config_path = xdg_config_home() / "frontools/config.py"
    if config_path.is_file():
        return config_path

    return None