news/news/config.py

"""Configuration loading.

Three files, with a deliberate split so a new user knows what to edit:

    config.toml   machinery: HTTP, embedding model, clustering and scoring
                  constants. Nothing personal. Sane defaults.
    profile.toml  who the reader is: interests, vetoes, demotes, importance,
                  delivery preferences.
    sources.toml  what to read.

Every path is overridable by environment variable so the same code can run
against a different reader's configuration without editing anything.
"""

from __future__ import annotations

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

REPO = Path(__file__).resolve().parent.parent


def config_dir() -> Path:
    """Resolved per call, not at import.

    Reading the environment at module scope meant NEWS_CONFIG_DIR was captured
    once when `news.config` was first imported, so setting it afterwards -- in a
    test, a subprocess, or anything that configures its environment after
    importing -- silently did nothing.
    """
    return Path(os.environ.get("NEWS_CONFIG_DIR", REPO / "profile"))


def _path(env: str, default: Path) -> Path:
    return Path(os.environ.get(env, default))


def _load(path: Path) -> dict[str, Any]:
    return tomllib.loads(path.read_text()) if path.exists() else {}


@dataclass
class Config:
    db_path: Path = field(default_factory=lambda: _path("NEWS_DB", REPO / "data" / "news.db"))
    config_path: Path = field(
        default_factory=lambda: _path("NEWS_CONFIG", config_dir() / "config.toml"))
    sources_path: Path = field(
        default_factory=lambda: _path("NEWS_SOURCES", config_dir() / "sources.toml"))
    profile_path: Path = field(
        default_factory=lambda: _path("NEWS_PROFILE", config_dir() / "profile.toml"))

    def __post_init__(self) -> None:
        self._raw = _load(self.config_path)

    # -- machinery ---------------------------------------------------------
    def section(self, name: str) -> dict[str, Any]:
        return self._raw.get(name, {})

    @property
    def user_agent(self) -> str:
        return self.section("http").get(
            "user_agent", "news/0.1 (+https://example.invalid; personal feed reader)")

    @property
    def http(self) -> dict[str, Any]:
        h = self.section("http")
        return {
            "timeout_secs": h.get("timeout_secs", 30),
            "connect_timeout_secs": h.get("connect_timeout_secs", 10),
            "concurrency": h.get("concurrency", 8),
            "max_body_bytes": h.get("max_body_bytes", 10 * 1024 * 1024),
        }

    @property
    def embedding(self) -> dict[str, Any]:
        e = self.section("embedding")
        return {"model": e.get("model", "BAAI/bge-small-en-v1.5"), "dim": e.get("dim", 384)}

    @property
    def clustering(self) -> dict[str, Any]:
        c = self.section("clustering")
        return {
            "threshold": c.get("threshold", 0.82),
            "window_hours": c.get("window_hours", 48),
            "syndication_threshold": c.get("syndication_threshold", 0.97),
        }

    @property
    def scoring(self) -> dict[str, Any]:
        s = self.section("scoring")
        return {
            "normalize": s.get("normalize", "zscore-per-block"),
            "veto_absolute": s.get("veto_absolute", 0.60),
            "veto_margin_over_interest": s.get("veto_margin_over_interest", 0.3),
            "corroboration_pickup_weight": s.get("corroboration_pickup_weight", 0.3),
            "corroboration_depth_weight": s.get("corroboration_depth_weight", 0.2),
            "demote_absolute": s.get("demote_absolute", 0.55),
            "importance_weights": s.get("importance_weights", {}),
        }

    @property
    def routing(self) -> dict[str, Any]:
        r = self.section("routing")
        return {
            "interest_per_day": r.get("interest_per_day", 3),
            "max_alerts_per_group_per_day": r.get("max_alerts_per_group_per_day", 1),
            "important_per_day": r.get("important_per_day", 2),
            "important_per_night": r.get("important_per_night", 1),
            "interest_threshold": r.get("interest_threshold", 2.0),
            "important_threshold": r.get("important_threshold", 2.0),
            "digest_floor": r.get("digest_floor", 0.5),
            "hn_min_score": r.get("hn_min_score", 150),
            "max_age_hours": r.get("max_age_hours", 48),
            "digest_max_items": r.get("digest_max_items", 25),
            "serendipity_max_interest": r.get("serendipity_max_interest", 0.5),
        }

    # -- reader ------------------------------------------------------------
    def sources(self) -> list[dict]:
        raw = _load(self.sources_path)
        defaults = raw.get("defaults", {})
        out = []
        for entry in raw.get("source", []):
            merged = {**defaults, **entry}
            merged.setdefault("kind", "rss")
            merged.setdefault("bucket", merged["name"].lower().replace(" ", "-"))
            merged.setdefault("weight", 1.0)
            merged.setdefault("prominence", 0)
            merged.setdefault("enabled", True)
            merged.setdefault("poll_interval_secs", 900)
            out.append(merged)
        return out

    def profile(self) -> dict[str, Any]:
        return _load(self.profile_path)

    def block_groups(self) -> dict[str, str]:
        """block id -> quota group, defaulting to the block's own id."""
        profile = self.profile()
        out: dict[str, str] = {}
        for kind in ("interest", "importance"):
            for b in profile.get(kind, []):
                out[b["id"]] = b.get("group", b["id"])
        return out

    @property
    def delivery(self) -> dict[str, Any]:
        return self.profile().get("delivery", {})