104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeFilters:
|
|
brands: set[str] = field(default_factory=set)
|
|
models: set[str] = field(default_factory=set)
|
|
years: set[int] = field(default_factory=set)
|
|
body_types: set[str] = field(default_factory=set)
|
|
|
|
def is_enabled(self) -> bool:
|
|
return bool(self.brands or self.models or self.years or self.body_types)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeConfig:
|
|
condition_check_enabled: bool | None = None
|
|
filters: RuntimeFilters = field(default_factory=RuntimeFilters)
|
|
|
|
|
|
def load_runtime_config(path: Path, logger: logging.Logger | None = None) -> RuntimeConfig:
|
|
if not path.exists():
|
|
return RuntimeConfig()
|
|
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception as exc: # noqa: BLE001
|
|
if logger is not None:
|
|
logger.warning("Failed to parse runtime config file '%s': %s", path, exc)
|
|
return RuntimeConfig()
|
|
|
|
if not isinstance(payload, dict):
|
|
if logger is not None:
|
|
logger.warning("Runtime config file '%s' root must be object.", path)
|
|
return RuntimeConfig()
|
|
|
|
sync_block = payload.get("sync")
|
|
filters_block = payload.get("filters")
|
|
if not isinstance(sync_block, dict):
|
|
sync_block = {}
|
|
if not isinstance(filters_block, dict):
|
|
filters_block = {}
|
|
|
|
return RuntimeConfig(
|
|
condition_check_enabled=_to_optional_bool(sync_block.get("condition_check_enabled")),
|
|
filters=RuntimeFilters(
|
|
brands=_to_ci_set(filters_block.get("brands")),
|
|
models=_to_ci_set(filters_block.get("models")),
|
|
years=_to_int_set(filters_block.get("years"), min_value=1900),
|
|
body_types=_to_ci_set(filters_block.get("body_types")),
|
|
)
|
|
)
|
|
|
|
|
|
def _to_optional_bool(value: Any) -> bool | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
normalized = value.strip().lower()
|
|
if normalized in {"true", "1", "yes", "on"}:
|
|
return True
|
|
if normalized in {"false", "0", "no", "off"}:
|
|
return False
|
|
return None
|
|
|
|
|
|
def _to_ci_set(value: Any) -> set[str]:
|
|
if not isinstance(value, list):
|
|
return set()
|
|
result: set[str] = set()
|
|
for item in value:
|
|
if not isinstance(item, str):
|
|
continue
|
|
text = item.strip().casefold()
|
|
if text:
|
|
result.add(text)
|
|
return result
|
|
|
|
|
|
def _to_int_set(value: Any, *, min_value: int) -> set[int]:
|
|
if not isinstance(value, list):
|
|
return set()
|
|
result: set[int] = set()
|
|
for item in value:
|
|
if isinstance(item, bool):
|
|
continue
|
|
if isinstance(item, int):
|
|
if item >= min_value:
|
|
result.add(item)
|
|
continue
|
|
if isinstance(item, str) and item.strip().isdigit():
|
|
parsed = int(item.strip())
|
|
if parsed >= min_value:
|
|
result.add(parsed)
|
|
return result
|