233 lines
7.2 KiB
Python
233 lines
7.2 KiB
Python
"""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
|
|
|
|
from xdg import xdg_config_dirs, xdg_config_home
|
|
from yaml import Loader
|
|
from yaml import load as load_yaml
|
|
|
|
from frontools.cache import Cache, FileCache, NullCache
|
|
from frontools.sources import CachedSource, OverrideSource, Source
|
|
from frontools.utils import ErrorSummary
|
|
|
|
REMOTE_SOURCE_NAME = "remote"
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""Error raised on config error"""
|
|
|
|
|
|
class UrlConfig:
|
|
"""Config for an url"""
|
|
|
|
def __init__(self) -> None:
|
|
self.tags: set[str] = set()
|
|
|
|
|
|
class SiteConfig:
|
|
"""Configuration object for a particular website"""
|
|
|
|
def __init__(self) -> None:
|
|
self.urls: dict[str, UrlConfig] = {}
|
|
|
|
|
|
class Config:
|
|
"""Configuration object"""
|
|
|
|
def __init__(
|
|
self,
|
|
use_cache: bool,
|
|
default_source_name: Optional[str],
|
|
include_urls: list[str],
|
|
exclude_urls: list[str],
|
|
include_tags: list[str],
|
|
exclude_tags: list[str],
|
|
):
|
|
self._use_cache = use_cache
|
|
self._sources: dict[str, Source] = {}
|
|
self._sites: dict[str, SiteConfig] = {}
|
|
self._block_urls: list[Pattern[str]] = []
|
|
|
|
if default_source_name is None:
|
|
self._default_source_name = REMOTE_SOURCE_NAME
|
|
else:
|
|
self._default_source_name = default_source_name
|
|
|
|
self._error_summary = ErrorSummary()
|
|
if use_cache:
|
|
remote_cache: Cache = FileCache(REMOTE_SOURCE_NAME)
|
|
else:
|
|
remote_cache = NullCache()
|
|
self._add_source(REMOTE_SOURCE_NAME, CachedSource, remote_cache)
|
|
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)
|
|
|
|
@staticmethod
|
|
async def load(
|
|
config_path: Optional[Path],
|
|
default_source_name: Optional[str],
|
|
use_cache: bool,
|
|
include_urls: list[str],
|
|
exclude_urls: list[str],
|
|
include_tags: list[str],
|
|
exclude_tags: list[str],
|
|
) -> "Config":
|
|
"""Load config from the given path"""
|
|
config = Config(
|
|
use_cache,
|
|
default_source_name,
|
|
include_urls,
|
|
exclude_urls,
|
|
include_tags,
|
|
exclude_tags,
|
|
)
|
|
|
|
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)
|
|
|
|
return config
|
|
|
|
@property
|
|
def remote_source(self) -> Source:
|
|
"""get the default source for this context"""
|
|
return self.get_source(REMOTE_SOURCE_NAME)
|
|
|
|
@property
|
|
def default_source(self) -> Source:
|
|
"""get the default source for this context"""
|
|
return self.get_source(self._default_source_name)
|
|
|
|
@property
|
|
def urls(self) -> Iterable[tuple[str, str]]:
|
|
"""Return sites configured for this context"""
|
|
for site_name, site in self._sites.items():
|
|
for url, config in site.urls.items():
|
|
if self._filter(url, config.tags):
|
|
yield site_name, url
|
|
|
|
def add_site_url(
|
|
self, name: str, url: str, tags: Optional[Iterable[str]] = None
|
|
) -> None:
|
|
"""Add an url for a site"""
|
|
|
|
site = self._sites.get(name, None)
|
|
if site is None:
|
|
site = SiteConfig()
|
|
self._sites[name] = site
|
|
|
|
if tags is None:
|
|
new_tags = set()
|
|
else:
|
|
new_tags = set(tags)
|
|
|
|
url_config = site.urls.get(url, None)
|
|
if url_config is None:
|
|
url_config = UrlConfig()
|
|
site.urls[url] = url_config
|
|
|
|
url_config.tags.update(new_tags)
|
|
|
|
def load_sites_from_yaml(self, yaml_path: str) -> None:
|
|
"""Load a yaml file containing dictionnary of urls to add as sites."""
|
|
with open(yaml_path, "r", encoding="utf-8") as yaml_file:
|
|
yaml_document = load_yaml(yaml_file, Loader)
|
|
for site_name, urls in yaml_document.items():
|
|
for url, tags in urls.items():
|
|
self.add_site_url(site_name, url, tags)
|
|
|
|
def block_url_patterns(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 add_override_source(
|
|
self,
|
|
name: str,
|
|
mappings: list[tuple[str, str]],
|
|
next_source_name: Optional[str] = None,
|
|
) -> None:
|
|
"""Add a source overriding given patterns"""
|
|
assert name not in self._sources
|
|
if next_source_name is None:
|
|
next_source = self.default_source
|
|
else:
|
|
next_source = self.get_source(next_source_name)
|
|
self._add_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 Exception(f"No source configured matching {name}")
|
|
return self._sources[name]
|
|
|
|
def echo_error_summary(self) -> None:
|
|
"""Echo the error summary to the user."""
|
|
self._error_summary.echo()
|
|
|
|
def _add_source(
|
|
self, name: str, source_class: Type[Source], *args: Any, **kwargs: Any
|
|
) -> None:
|
|
if name in self._sources:
|
|
raise Exception(f"Source {name} already configured")
|
|
self._sources[name] = source_class(
|
|
self._error_summary, self._block_urls, *args, **kwargs
|
|
)
|
|
|
|
def _filter(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
|