148 lines
4.9 KiB
Python
148 lines
4.9 KiB
Python
"""Configuration loading and path resolution.
|
|
|
|
All tunable behaviour (URLs, retry policy, cache locations, alert sink) lives in
|
|
config.yaml. This module loads it, resolves relative paths against the repo root,
|
|
and exposes a small, validated object. Nothing in the rest of the package reads
|
|
files directly -- it goes through Config.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError as exc: # pragma: no cover - PyYAML is a hard dep
|
|
raise SystemExit(
|
|
"PyYAML is required. Install with: python3 -m pip install --user pyyaml"
|
|
) from exc
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
DEFAULT_CONFIG_PATH = PROJECT_ROOT / "config.yaml"
|
|
|
|
# Sensible fallbacks so the package still runs if a key is absent from config.yaml.
|
|
_DEFAULTS: Dict[str, Any] = {
|
|
"http": {
|
|
"user_agent": "Mozilla/5.0 (research-scanner/1.0)",
|
|
"timeout_seconds": 20,
|
|
"max_retries": 3,
|
|
"backoff_base_seconds": 1.0,
|
|
"backoff_cap_seconds": 30.0,
|
|
"retry_status": [500, 502, 503, 504],
|
|
"antibot_status": [403],
|
|
},
|
|
"cache": {
|
|
"raw_dir": "cache/raw",
|
|
"request_log": "cache/request_log.jsonl",
|
|
},
|
|
"alerts": {
|
|
"enabled": True,
|
|
"sink": "log",
|
|
},
|
|
# spec §4 LLM analysis (task t_950780e0). Kept as fallbacks so the package
|
|
# still runs even if config.yaml omits the `llm:` section. The authoritative
|
|
# values live in config.yaml; these only fill missing keys.
|
|
"llm": {
|
|
"enabled": False,
|
|
"backend": "ollama",
|
|
"endpoint": "http://localhost:11434",
|
|
"model": "llama3",
|
|
"default_tags": 8,
|
|
"timeout_seconds": 30,
|
|
},
|
|
"sources": {},
|
|
}
|
|
|
|
_REQUIRED_SOURCES = ("assembly_daily_file", "senate_daily_file")
|
|
|
|
|
|
class Config:
|
|
"""Thin validated wrapper around the parsed config.yaml."""
|
|
|
|
def __init__(self, data: Dict[str, Any], root: Path, path: Path):
|
|
self.data = data
|
|
self.root = root
|
|
self.path = path
|
|
self._http = _merge(_DEFAULTS["http"], data.get("http") or {})
|
|
self._cache = _merge(_DEFAULTS["cache"], data.get("cache") or {})
|
|
self._alerts = _merge(_DEFAULTS["alerts"], data.get("alerts") or {})
|
|
self._llm = _merge(_DEFAULTS["llm"], data.get("llm") or {})
|
|
self._sources = data.get("sources") or {}
|
|
self._validate_sources()
|
|
|
|
# -- paths ---------------------------------------------------------
|
|
@property
|
|
def raw_dir(self) -> Path:
|
|
return self._resolve(self._cache["raw_dir"])
|
|
|
|
@property
|
|
def request_log(self) -> Path:
|
|
return self._resolve(self._cache["request_log"])
|
|
|
|
# -- config slices -------------------------------------------------
|
|
@property
|
|
def http(self) -> Dict[str, Any]:
|
|
return self._http
|
|
|
|
@property
|
|
def alerts(self) -> Dict[str, Any]:
|
|
return self._alerts
|
|
|
|
@property
|
|
def llm(self) -> Dict[str, Any]:
|
|
return self._llm
|
|
|
|
@property
|
|
def sources(self) -> Dict[str, Dict[str, Any]]:
|
|
return self._sources
|
|
|
|
def source(self, name: str) -> Dict[str, Any]:
|
|
if name not in self._sources:
|
|
raise KeyError(f"Unknown source '{name}'. Known: {list(self._sources)}")
|
|
return self._sources[name]
|
|
|
|
def source_names(self) -> List[str]:
|
|
return list(self._sources.keys())
|
|
|
|
# -- helpers -------------------------------------------------------
|
|
def _resolve(self, rel: str) -> Path:
|
|
p = Path(rel)
|
|
return p if p.is_absolute() else (self.root / p)
|
|
|
|
def _validate_sources(self) -> None:
|
|
for name in _REQUIRED_SOURCES:
|
|
if name not in self._sources:
|
|
raise ValueError(
|
|
f"config.yaml is missing sources.{name} "
|
|
f"(required by the research spec t_a171f4d1)."
|
|
)
|
|
asm = self._sources.get("assembly_daily_file", {})
|
|
sen = self._sources.get("senate_daily_file", {})
|
|
if not asm.get("url"):
|
|
raise ValueError("sources.assembly_daily_file.url is required")
|
|
if not sen.get("index_url"):
|
|
raise ValueError("sources.senate_daily_file.index_url is required")
|
|
|
|
|
|
def _merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
|
|
out = dict(base)
|
|
out.update(override or {})
|
|
return out
|
|
|
|
|
|
def load_config(path: str | os.PathLike | None = None) -> Config:
|
|
"""Load config.yaml; falls back to defaults with a warning if absent."""
|
|
p = Path(path) if path else DEFAULT_CONFIG_PATH
|
|
p = p if p.is_absolute() else (Path.cwd() / p)
|
|
if not p.exists():
|
|
raise SystemExit(f"Config file not found: {p}")
|
|
with open(p, "r", encoding="utf-8") as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"Config file {p} did not parse to a mapping.")
|
|
root = p.resolve().parent
|
|
return Config(data, root=root, path=p.resolve())
|