second-brain/src/second_brain/config.py

"""
Configuration loading for second-brain.

Settings are read from config/settings.toml (or a path set via SECOND_BRAIN_CONFIG).
All values have sensible defaults so the system works out of the box.

The relational store is Postgres (petalbrain DB, `second_brain` schema). Connection
details are resolved by `second_brain.database` — see that module for the URL
resolution order. The vault is still a file-based Obsidian directory; this
project does NOT dual-write to petalbrain.wiki.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

try:
    import tomllib  # Python 3.11+
except ModuleNotFoundError:
    import tomli as tomllib  # type: ignore[no-reuse-impl]

# ---------------------------------------------------------------------------
# Defaults
# ---------------------------------------------------------------------------

_DEFAULTS: dict[str, Any] = {
    "vault_path": "~/Documents/second-brain-vault",
    "media_dir": "~/.local/share/second-brain/media",
    "subtitles_dir": "~/.local/share/second-brain/subtitles",
    "domains": ["development", "content", "business", "homelab"],
    "whisper": {
        # Tower default: faster-whisper large-v3 on cuda/float16. The
        # transcribe helper picks int8 automatically if the device is cpu.
        "whisper_model": "large-v3",
        "whisper_device": "cuda",
        "whisper_compute_type": "",
    },
    "database": {
        # Empty by default — the runtime expects SECOND_BRAIN_DATABASE_URL
        # (or HERBYLAB_DATABASE_URL) in the environment. Fill this only if
        # you prefer keeping the URL inside settings.toml.
        "url": "",
    },
    "extractor": {
        # "cli" runs `claude -p` under the Max OAuth subscription (default).
        # "api" uses the Anthropic SDK and requires ANTHROPIC_API_KEY.
        "backend": "cli",
        "model": "claude-sonnet-4-6",
        "cli_binary": "claude",
        "timeout_sec": 600,
    },
    "embeddings": {
        # Best-effort embedding of extraction summaries into public.embeddings.
        # If disabled, or if Ollama / the DB is unreachable, the pipeline
        # logs a warning and keeps going — file/relational data stays the
        # source of truth.
        "enabled": True,
        # Default targets the `ollama` service on the homelab docker network
        # (same convention vault-mcp uses). For host-side runs, set
        # OLLAMA_URL=http://127.0.0.1:11434 in the env instead.
        "ollama_url": "http://ollama:11434",
        "model": "nomic-embed-text",
        "embed_field": "summary",
    },
    "scheduler": {
        "window_start": "22:00",
        "window_end": "06:00",
        # CLI backend: throttle by request rate (no per-token billing).
        "max_calls_per_hour": 30,
        "min_gap_seconds": 120,
        # Legacy API-backend budget; kept for backwards compat.
        "max_tokens_per_hour": 100_000,
    },
}

DOMAINS = ["development", "content", "business", "homelab"]


def _merge(default: dict, override: dict) -> dict:
    """Shallow merge — override wins per top-level key."""
    out = dict(default)
    out.update(override or {})
    return out


class Config:
    """Holds resolved configuration values."""

    def __init__(self, raw: dict[str, Any]) -> None:
        self._raw = raw

    # --- paths ---

    @property
    def vault_path(self) -> Path:
        return Path(self._raw.get("vault_path", _DEFAULTS["vault_path"])).expanduser()

    @property
    def media_dir(self) -> Path:
        return Path(self._raw.get("media_dir", _DEFAULTS["media_dir"])).expanduser()

    @property
    def subtitles_dir(self) -> Path:
        return Path(
            self._raw.get("subtitles_dir", _DEFAULTS["subtitles_dir"])
        ).expanduser()

    # --- transcription ---

    @property
    def whisper(self) -> dict[str, Any]:
        """Merged [whisper] block. Env vars in `transcribe.resolve_settings`
        still win over these for the tower worker."""
        return _merge(_DEFAULTS["whisper"], self._raw.get("whisper", {}))

    @property
    def whisper_model(self) -> str:
        """Backwards-compat shortcut — older callers (and settings.toml
        in the wild) reference whisper_model at the top level."""
        return (
            self._raw.get("whisper_model")
            or self.whisper.get("whisper_model", "large-v3")
        )

    # --- extraction ---

    @property
    def domains(self) -> list[str]:
        return self._raw.get("domains", _DEFAULTS["domains"])

    @property
    def anthropic_api_key(self) -> str | None:
        return os.getenv("ANTHROPIC_API_KEY")

    # --- structured sections ---

    @property
    def database(self) -> dict[str, Any]:
        return _merge(_DEFAULTS["database"], self._raw.get("database", {}))

    @property
    def extractor(self) -> dict[str, Any]:
        return _merge(_DEFAULTS["extractor"], self._raw.get("extractor", {}))

    @property
    def embeddings(self) -> dict[str, Any]:
        return _merge(_DEFAULTS["embeddings"], self._raw.get("embeddings", {}))

    @property
    def scheduler(self) -> dict[str, Any]:
        return _merge(_DEFAULTS["scheduler"], self._raw.get("scheduler", {}))

    # --- prompts dir ---

    @property
    def prompts_dir(self) -> Path:
        """Directory containing per-domain prompt templates."""
        # src/second_brain/config.py → ../../.. = project root
        here = Path(__file__).resolve().parent.parent.parent
        return here / "prompts"

    def ensure_dirs(self) -> None:
        """Create runtime directories if they don't exist."""
        for d in (self.media_dir, self.subtitles_dir):
            d.mkdir(parents=True, exist_ok=True)


# ---------------------------------------------------------------------------
# Loader
# ---------------------------------------------------------------------------

_config_instance: Config | None = None


def load_config(path: Path | None = None) -> Config:
    """Load (and cache) configuration from a TOML file."""
    global _config_instance
    if _config_instance is not None:
        return _config_instance

    if path is None:
        env_path = os.getenv("SECOND_BRAIN_CONFIG")
        if env_path:
            path = Path(env_path)
        else:
            # Walk up from CWD looking for config/settings.toml
            cwd = Path.cwd()
            for candidate in [cwd / "config" / "settings.toml", cwd / "settings.toml"]:
                if candidate.exists():
                    path = candidate
                    break

    raw: dict[str, Any] = {}
    if path and path.exists():
        with open(path, "rb") as fh:
            raw = tomllib.load(fh)

    _config_instance = Config(raw)
    return _config_instance


def reset_config_singleton() -> None:
    """Test hook — clear the cached Config so a re-read picks up new settings."""
    global _config_instance
    _config_instance = None


__all__ = ["Config", "DOMAINS", "load_config", "reset_config_singleton"]