AV-Watch/avwatch/config.py

"""Configuration loading.

Secrets (SMTP password, API keys) are read from environment variables so the
TOML file can live in version control or a shared drive without leaking
credentials. Env vars win over anything in the file.
"""

from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class EmailConfig:
    enabled: bool = False
    smtp_host: str = ""
    smtp_port: int = 587
    use_starttls: bool = True
    use_ssl: bool = False
    username: str = ""
    password: str = ""          # from AVWATCH_SMTP_PASSWORD
    from_addr: str = ""
    from_name: str = "AVWatch"
    to: list[str] = field(default_factory=list)
    cc: list[str] = field(default_factory=list)
    subject_prefix: str = "[AVWatch]"
    attach_html: bool = True
    send_when_nothing_new: bool = True


@dataclass
class RiskConfig:
    enabled: bool = True
    max_queries_per_item: int = 2
    max_evidence_per_item: int = 8
    lookback_days: int = 400
    use_duckduckgo: bool = True
    use_reddit: bool = True
    use_github_issues: bool = True
    use_hackernews: bool = False
    reddit_client_id: str = ""       # from AVWATCH_REDDIT_CLIENT_ID
    reddit_client_secret: str = ""   # from AVWATCH_REDDIT_CLIENT_SECRET
    subreddits: list[str] = field(default_factory=lambda: [
        "livesound", "VIDEOENGINEERING", "churchtech", "ProPresenter",
        "blackmagicdesign", "audioengineering", "shure", "commandcenter",
    ])
    # Only run community scans for items we are actually behind on, to keep
    # the query budget sane.
    only_when_behind: bool = True


@dataclass
class LLMConfig:
    enabled: bool = True
    # Set model = "claude-sonnet-5" in config.toml to cut per-run cost roughly
    # in half if the Opus verdicts read as more than you need.
    model: str = "claude-opus-5"
    api_key: str = ""            # from ANTHROPIC_API_KEY
    max_items_per_run: int = 40
    max_output_tokens: int = 1400
    # Escalate to the strongest model for show-critical gear.
    escalate_model: str = "claude-opus-5"
    escalate_for_criticality: list[str] = field(default_factory=lambda: ["showstopper"])


@dataclass
class LookupConfig:
    """Claude-with-web-search version lookup, for the many vendors whose
    support sites are JavaScript-rendered and cannot be scraped."""

    enabled: bool = True
    # A factual retrieval job, not a judgement call, so it runs on the cheaper
    # model by default. Raise to claude-opus-5 if you see it getting versions
    # wrong on obscure gear.
    model: str = "claude-sonnet-5"
    api_key: str = ""              # from ANTHROPIC_API_KEY
    ttl_days: float = 6.0          # Monday pays; mid-week scans reuse
    max_items_per_run: int = 45    # hard ceiling on cost per run
    max_searches_per_item: int = 4


@dataclass
class PolicyConfig:
    """Local rules that shape recommendations independent of what the
    internet says."""

    # A release younger than this is 'wait' by default for critical gear.
    soak_days_showstopper: int = 30
    soak_days_high: int = 21
    soak_days_normal: int = 10
    # Never recommend installing during these ISO weekday numbers (1=Mon).
    blackout_weekdays: list[int] = field(default_factory=lambda: [6, 7])
    # Treat a major version jump on critical gear as needing a human.
    major_bump_needs_human: bool = True


@dataclass
class Config:
    root: Path
    data_dir: Path
    cache_dir: Path
    db_path: Path
    report_dir: Path
    inventory_path: Path
    cache_ttl: int = 3600
    http_timeout: int = 30
    log_level: str = "INFO"
    email: EmailConfig = field(default_factory=EmailConfig)
    risk: RiskConfig = field(default_factory=RiskConfig)
    llm: LLMConfig = field(default_factory=LLMConfig)
    lookup: LookupConfig = field(default_factory=LookupConfig)
    policy: PolicyConfig = field(default_factory=PolicyConfig)


def _as_list(v) -> list[str]:
    if v is None:
        return []
    if isinstance(v, str):
        return [x.strip() for x in v.split(",") if x.strip()]
    return [str(x) for x in v]


def _fill(dc, table: dict) -> None:
    """Copy known keys from a TOML table onto a dataclass instance."""
    for key, val in table.items():
        if not hasattr(dc, key):
            continue
        cur = getattr(dc, key)
        if isinstance(cur, list):
            val = _as_list(val)
        elif isinstance(cur, bool):
            val = bool(val)
        elif isinstance(cur, int) and not isinstance(val, bool):
            try:
                val = int(val)
            except (TypeError, ValueError):
                continue
        setattr(dc, key, val)


def load(config_path: Path | None = None, root: Path | None = None) -> Config:
    root = Path(root or Path(__file__).resolve().parent.parent)
    config_path = Path(config_path) if config_path else root / "config.toml"

    raw: dict = {}
    if config_path.exists():
        with open(config_path, "rb") as fh:
            raw = tomllib.load(fh)

    paths = raw.get("paths", {})
    data_dir = Path(paths.get("data_dir") or root / "data")
    cfg = Config(
        root=root,
        data_dir=data_dir,
        cache_dir=Path(paths.get("cache_dir") or data_dir / "cache"),
        db_path=Path(paths.get("db_path") or data_dir / "avwatch.sqlite3"),
        report_dir=Path(paths.get("report_dir") or root / "reports"),
        inventory_path=Path(paths.get("inventory") or root / "inventory.yaml"),
    )

    general = raw.get("general", {})
    cfg.cache_ttl = int(general.get("cache_ttl", cfg.cache_ttl))
    cfg.http_timeout = int(general.get("http_timeout", cfg.http_timeout))
    cfg.log_level = str(general.get("log_level", cfg.log_level)).upper()

    _fill(cfg.email, raw.get("email", {}))
    _fill(cfg.risk, raw.get("risk", {}))
    _fill(cfg.llm, raw.get("llm", {}))
    _fill(cfg.lookup, raw.get("lookup", {}))
    _fill(cfg.policy, raw.get("policy", {}))

    # Secrets: environment always wins.
    cfg.email.password = os.environ.get("AVWATCH_SMTP_PASSWORD", cfg.email.password)
    api_key = os.environ.get("ANTHROPIC_API_KEY", "")
    cfg.llm.api_key = api_key or cfg.llm.api_key
    cfg.lookup.api_key = api_key or cfg.lookup.api_key or cfg.llm.api_key
    cfg.risk.reddit_client_id = os.environ.get(
        "AVWATCH_REDDIT_CLIENT_ID", cfg.risk.reddit_client_id)
    cfg.risk.reddit_client_secret = os.environ.get(
        "AVWATCH_REDDIT_CLIENT_SECRET", cfg.risk.reddit_client_secret)

    if not cfg.llm.api_key:
        cfg.llm.enabled = False
    if not cfg.lookup.api_key:
        cfg.lookup.enabled = False

    for d in (cfg.data_dir, cfg.cache_dir, cfg.report_dir):
        d.mkdir(parents=True, exist_ok=True)

    return cfg