S.H.O.N.A.R._Desktop_Companion/backend/shonar/core/config.py

"""Application settings.

All configuration comes from environment variables (and optionally an
admin config file pointed at by SHONAR_CONFIG_FILE). API keys and other
secrets must NEVER be hard-coded.
"""

from __future__ import annotations

from functools import lru_cache

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


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

    # --- Core -------------------------------------------------------------
    app_name: str = "S.H.O.N.A.R."
    debug: bool = False
    # Secret used to sign access tokens. MUST be set in production.
    secret_key: str = Field(default="", repr=False)
    access_token_ttl_minutes: int = 15
    refresh_token_ttl_days: int = 30
    # Comma-separated list of allowed registration modes: "open", "invite"
    allow_registration: bool = True

    # --- Database -----------------------------------------------------------
    database_url: str = "postgresql+asyncpg://shonar:shonar@localhost:5432/shonar"
    db_pool_size: int = 5
    db_max_overflow: int = 10

    # --- Storage ------------------------------------------------------------
    # "local" or "s3"
    storage_backend: str = "local"
    storage_path: str = "./data/storage"
    # S3 (used when storage_backend == "s3")
    s3_endpoint_url: str = ""
    s3_bucket: str = ""
    s3_region: str = "us-east-1"
    s3_access_key_id: str = Field(default="", repr=False)
    s3_secret_access_key: str = Field(default="", repr=False)

    # --- Upload limits --------------------------------------------------------
    max_upload_bytes: int = 2 * 1024 * 1024 * 1024  # 2 GiB
    max_chunk_bytes: int = 16 * 1024 * 1024
    allowed_audio_mime_types: list[str] = [
        "audio/mp4",
        "audio/m4a",
        "audio/aac",
        "audio/wav",
        "audio/x-wav",
        "audio/ogg",
        "audio/opus",
        "audio/webm",
        "audio/mpeg",
    ]

    # --- Worker / queue -------------------------------------------------------
    redis_url: str = "redis://localhost:6379/0"
    # "arq" = Redis transport (server deployments). "inline" = in-process
    # asyncio runner (desktop bundled-lite engine: no Redis, one job at a
    # time). DB rows are the queue of record either way.
    queue_backend: str = "arq"  # arq | inline
    # Apply 'alembic upgrade head' at startup. The desktop bundled engine
    # owns its SQLite file and must self-migrate; server deployments run
    # migrations in their deploy flow instead.
    auto_migrate: bool = False

    # --- Audio processing -----------------------------------------------------
    # Optional server-side conversion via FFmpeg. Off by default.
    audio_conversion_enabled: bool = False
    ffmpeg_bin: str = "ffmpeg"

    # --- AI providers -----------------------------------------------------
    # "none" disables all AI processing (recording/sync/playback still work).
    transcription_provider: str = "none"  # none | whisper_http | faster_whisper
    transcription_model: str = "base"
    transcription_base_url: str = ""  # whisper-compatible HTTP server
    transcription_api_key: str = Field(default="", repr=False)

    llm_provider: str = "none"  # none | openai_compat | ollama
    llm_model: str = ""
    llm_base_url: str = ""
    llm_api_key: str = Field(default="", repr=False)

    # Rescue summarizer tried when the PRIMARY one fails mid-job (desktop:
    # LAN GPU server primary, local Ollama fallback). "none" disables the
    # fallback; the primary's error then stands as today.
    llm_fallback_provider: str = "none"  # none | openai_compat | ollama
    llm_fallback_model: str = ""
    llm_fallback_base_url: str = ""
    llm_fallback_api_key: str = Field(default="", repr=False)

    # Search needs no setting: services/search.py picks its path by SQL
    # dialect (SQLite substring scan / Postgres tsvector). A Meilisearch or
    # OpenSearch backend would replace that module, not add a config knob.

    # --- Retention --------------------------------------------------------
    # Grace window before the sweep hard-deletes soft-deleted recordings
    # and accounts (rows + stored files). Cancellations are valid until
    # the sweep fires.
    retention_grace_days: int = 30

    # --- Misc -------------------------------------------------------------
    rate_limit_auth: str = "10/minute"
    rate_limit_default: str = "120/minute"

    @field_validator("allowed_audio_mime_types", mode="before")
    @classmethod
    def _split_mime(cls, v):  # noqa: ANN001, ANN206
        if isinstance(v, str):
            return [item.strip() for item in v.split(",") if item.strip()]
        return v

    @property
    def ai_enabled(self) -> bool:
        return self.transcription_provider != "none" or self.llm_provider != "none"

    def validate_production(self) -> list[str]:
        """Return a list of configuration warnings (empty == OK)."""
        warnings: list[str] = []
        if not self.secret_key or len(self.secret_key) < 32:
            warnings.append(
                "SHONAR_SECRET_KEY is missing or shorter than 32 characters. "
                "Set a strong random secret in production."
            )
        if self.storage_backend == "s3" and not self.s3_bucket:
            warnings.append("storage_backend=s3 but SHONAR_S3_BUCKET is empty.")
        if self.transcription_provider == "whisper_http" and not self.transcription_base_url:
            warnings.append(
                "transcription_provider=whisper_http requires SHONAR_TRANSCRIPTION_BASE_URL."
            )
        if self.llm_provider == "openai_compat" and not self.llm_base_url:
            warnings.append("llm_provider=openai_compat requires SHONAR_LLM_BASE_URL.")
        return warnings


@lru_cache
def get_settings() -> Settings:
    return Settings()