Files
Цвылев Александр Вадимович 313af3a070 feat(transcode): optimize + on-the-fly quality + HLS streaming (§6.6)
POST /tracks/{id}/optimize enqueues a transcode; GET /stream/{id}?quality=
serves a cached Opus rendition (miss → master + background warm); GET
/stream/{id}/hls/{playlist.m3u8,segment} serves the worker-generated HLS
rendition (AAC-in-TS), with ?token= propagated onto segment URLs for players
that can't set headers. Hexagonal: Transcoder port, FfmpegTranscoder adapter,
TranscodeService + cache-path helpers, transcode_track worker (idempotent),
schemas + deps wiring. ffmpeg already in the image.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-28 21:19:36 +03:00

172 lines
7.5 KiB
Python

"""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
# -- CORS -------------------------------------------------------------
# Origins allowed to call the API from a browser. The web UI is multi-
# instance — it connects to whatever origin the operator types on the
# connect screen — so a page served from origin A may call this backend at
# origin B (e.g. the direct :8000 port, a LAN IP, or 127.0.0.1 vs localhost).
# Auth rides in the ``Authorization`` bearer header (not cookies), so the
# wildcard default is safe here — it is paired with ``allow_credentials=False``.
# Set explicit origins in hardened deployments. Accepts a comma-separated
# string in ``.env`` (``CORS_ALLOW_ORIGINS=https://a,https://b``) or ``*``.
cors_allow_origins: list[str] = Field(default_factory=lambda: ["*"])
# -- 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")
# ffmpeg binary for transcoding/HLS (on PATH in the image); override for a
# non-standard location.
ffmpeg_path: str = "ffmpeg"
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
@field_validator("cors_allow_origins", mode="before")
@classmethod
def _split_cors_origins(cls, v: object) -> object:
# Allow a plain comma-separated string in .env (pydantic would otherwise
# try to JSON-decode a list field): "a, b" -> ["a", "b"]; "*" -> ["*"].
if isinstance(v, str):
return [origin.strip() for origin in v.split(",") if origin.strip()]
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()