Fast IAAI implementation

iaai_sync_service/runtime_config.py

@@ -0,0 +1,103 @@
+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