mcma-backend/app/core/config.py

"""Application settings — single source of truth, sourced from environment.

Nothing is hardcoded. All knobs come from env vars (or a local ``.env`` during
development). Access the cached singleton via :func:`get_settings`.
"""

from functools import lru_cache
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
from typing import Literal

from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

# App identity for outbound API calls (e.g. the MusicBrainz/AcoustID
# User-Agent). Name is fixed; version comes from the installed package.
APP_NAME = "MCMA"
_PROJECT_URL = "https://git.ollyhearn.ru/olly/mcma-backend"


def app_version() -> str:
    try:
        return version("mcma-backend")
    except PackageNotFoundError:
        return "0.0.0"


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
        case_sensitive=False,
    )

    # -- runtime ----------------------------------------------------------
    environment: Literal["dev", "test", "prod"] = "dev"
    log_level: str = "INFO"
    log_json: bool = Field(
        default=False,
        description="Structured JSON logs (enable in prod).",
    )

    # -- database ---------------------------------------------------------
    # Async driver required (asyncpg). Example:
    # postgresql+asyncpg://mcma:mcma@db:5432/mcma
    database_url: str = "postgresql+asyncpg://mcma:mcma@localhost:5432/mcma"
    db_echo: bool = False
    db_pool_size: int = 5
    db_max_overflow: int = 10

    # -- redis (cache + arq broker) --------------------------------------
    redis_url: str = "redis://localhost:6379/0"

    # -- auth -------------------------------------------------------------
    jwt_secret: SecretStr = SecretStr("change-me-in-prod")
    jwt_algorithm: str = "HS256"
    access_token_ttl_seconds: int = 60 * 15  # 15 min
    refresh_token_ttl_seconds: int = 60 * 60 * 24 * 30  # 30 days (offline-first)
    # Public self-service sign-up. When disabled, accounts are created
    # admin-only (POST /admin/users). Registered users are never superusers.
    allow_registration: bool = True

    # -- subsonic ---------------------------------------------------------
    # Symmetric key (any string) used to encrypt each user's recoverable
    # Subsonic app-password at rest. A Fernet key is derived from it; rotating
    # this value renders stored app-passwords undecryptable (rotate them too).
    subsonic_secret_key: SecretStr = SecretStr("change-me-subsonic-key")

    # -- media / storage --------------------------------------------------
    media_path: Path = Path("/data/media")
    transcode_cache_path: Path = Path("/data/transcode-cache")
    max_parallel_downloads: int = 2
    # How many times the download worker retries a failed fetch (yt-dlp fails
    # often) before marking the job ``failed`` — exponential backoff between tries.
    download_max_retries: int = 3
    storage_backend: Literal["local", "s3"] = "local"
    upload_tmp_dir: Path | None = None

    # -- sources ----------------------------------------------------------
    # Mounted folder the ``local`` source indexes (copies into managed storage).
    # Unset → the local source is simply not registered.
    local_media_import_path: Path | None = None

    # -- S3 storage (deferred; set storage_backend="s3" to use) ----------
    s3_endpoint_url: str | None = None
    s3_bucket: str | None = None
    s3_region: str | None = None
    s3_access_key: SecretStr | None = None
    s3_secret_key: SecretStr | None = None

    # -- external services (all optional; graceful degradation) ----------
    ml_service_url: str | None = None
    acoustid_api_key: SecretStr | None = None
    acoustid_api_url: str = "https://api.acoustid.org/v2/lookup"
    # Above this AcoustID match score, trust the acoustic identification over
    # embedded file tags (which are frequently junk on downloaded files —
    # e.g. "Music Track" / "Sound_12345"). Below it, keep the tag-first merge.
    acoustid_trust_score: float = 0.85
    # MusicBrainz/AcoustID require a meaningful User-Agent identifying the
    # application and a way to contact its maintainer (see
    # https://musicbrainz.org/doc/XML_Web_Service/Rate_Limiting). Self-hosted
    # deployments should set their own contact email; see
    # ``musicbrainz_user_agent`` below for how it's used.
    musicbrainz_owner_email: str | None = None
    # ``youtube`` fetch source (search + download via ytmusicapi/yt-dlp). Enabled
    # by default; the source still reports unavailable if the libs aren't present.
    youtube_enabled: bool = True
    # Optional cookies file (Netscape format) for yt-dlp — lets it fetch
    # age-restricted / region-locked items via an authenticated session.
    youtube_cookies_path: Path | None = None

    # -- enrichment -------------------------------------------------------
    # ``fpcalc`` (Chromaprint) binary; resolved on PATH by default. The Docker
    # image installs it via libchromaprint-tools.
    fpcalc_path: str = "fpcalc"

    # Cover Art Archive — network fallback for album covers (after embedded art).
    # Disable to keep enrichment fully offline; embedded artwork still works.
    coverart_enabled: bool = True
    coverart_base_url: str = "https://coverartarchive.org"

    @field_validator("database_url")
    @classmethod
    def _require_async_driver(cls, v: str) -> str:
        if "+asyncpg" not in v:
            raise ValueError("database_url must use the asyncpg driver: postgresql+asyncpg://")
        return v

    @property
    def is_prod(self) -> bool:
        return self.environment == "prod"

    @property
    def musicbrainz_user_agent(self) -> str:
        """User-Agent sent to MusicBrainz/AcoustID: ``MCMA/<version> ( <contact> )``.

        Falls back to the project URL if the deployment hasn't set
        ``musicbrainz_owner_email``.
        """
        contact = self.musicbrainz_owner_email or _PROJECT_URL
        return f"{APP_NAME}/{app_version()} ( {contact} )"


@lru_cache
def get_settings() -> Settings:
    """Cached settings singleton. Patch the cache in tests via ``get_settings.cache_clear()``."""
    return Settings()