"""Centralized application configuration. All settings are loaded from environment variables (or a .env file) once at startup. Provider-specific defaults (model names, base URLs) are resolved by the provider adapters, not here. """ import copy import logging.config from collections.abc import Sequence from enum import StrEnum from functools import cache from pathlib import Path from typing import Annotated from typing import Any from typing import Literal from pydantic import BaseModel from pydantic import ConfigDict from pydantic import Field from pydantic import SecretStr from pydantic import StringConstraints from pydantic import field_validator from pydantic import model_validator from pydantic_settings import BaseSettings from pydantic_settings import SettingsConfigDict logger = logging.getLogger(__name__) class Provider(StrEnum): OPENROUTER = "openrouter" NonEmptyStr = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)] PromptFilename = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, pattern=r"^[^/\\]+$")] Probability = Annotated[float, Field(ge=0.0, le=1.0)] Temperature = Annotated[float, Field(ge=0.0, le=2.0)] DEFAULT_PROVIDER_MODEL = "google/gemini-2.5-flash" class SqliteSettings(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) driver: Literal["sqlite"] = "sqlite" path: NonEmptyStr = "./data/transcription.db" class PostgresSettings(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) driver: Literal["postgres"] = "postgres" host: NonEmptyStr port: int = Field(default=5432, ge=1, le=65535) database: NonEmptyStr user: NonEmptyStr password: SecretStr DatabaseSettings = Annotated[ SqliteSettings | PostgresSettings, Field(discriminator="driver"), ] class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", env_nested_delimiter="__", cli_implicit_flags=True, cli_kebab_case=True, frozen=True, ) # --- NiceGUI Server --- host: str = "0.0.0.0" port: int = 8000 log_level: Literal["critical", "error", "warning", "info", "debug", "trace"] = "info" reload: bool = False log_dir: Path = Path("./data/logs") log_file_name: NonEmptyStr = "transcription.log" log_file_max_bytes: int = Field(default=10 * 1024 * 1024, gt=0) log_file_backup_count: int = Field(default=5, ge=1) # --- AI provider --- provider: Provider = Provider.OPENROUTER openrouter_api_key: SecretStr provider_model: NonEmptyStr | None = DEFAULT_PROVIDER_MODEL provider_models: tuple[NonEmptyStr, ...] = () openrouter_http_referer: NonEmptyStr | None = None openrouter_app_title: NonEmptyStr | None = None default_prompt_name: PromptFilename = "transcribe_document.md" transcription_temperature: Temperature | None = None transcription_top_p: Probability | None = None # --- runtime environment --- environment: Literal["development", "test", "production"] = "development" transcription_commit: NonEmptyStr | None = None # --- persistence --- database: DatabaseSettings = Field(default_factory=SqliteSettings) bootstrap_schema_on_startup: bool = False sqlite_check_same_thread: bool = False # --- filesystem paths --- upload_dir: Path = Path("./data") prompt_dir: Path = Path("./prompts") database_backup_dir: Path = Path("./data/backups") # --- worker reliability --- worker_max_retries: int = Field(default=0, ge=0) # Bounded only from below. Vision transcription of a dense page routinely runs # well past twenty seconds, so an upper cap here would silently fail real work. worker_provider_timeout_seconds: float = Field(default=30.0, gt=0.0) worker_min_transcription_chars: int = Field(default=0, ge=0) worker_min_transcription_lines: int = Field(default=0, ge=0) worker_fail_on_finish_reason_length: bool = False @field_validator("provider_models", mode="before") @classmethod def validate_provider_models_input(cls, value: object) -> object: if value is None: return () if isinstance(value, (list, tuple)) and not value: raise ValueError("PROVIDER_MODELS must contain at least one model") return value @model_validator(mode="before") @classmethod def normalize_provider_models(cls, data: object) -> object: """Build the immutable model selector with the configured default first. This runs before field validation so the derived value is produced by normal construction rather than by mutating a frozen instance. """ if not isinstance(data, dict): return data default_model = data.get("provider_model") or DEFAULT_PROVIDER_MODEL if not isinstance(default_model, str): return data default_model = default_model.strip() configured = data.get("provider_models") if configured is None: configured = () elif isinstance(configured, str): # Left as-is so the field validator can report the malformed value. return {**data, "provider_model": default_model} elif not isinstance(configured, (list, tuple)): return {**data, "provider_model": default_model} elif not configured: # Preserved so validate_provider_models_input can reject it. return {**data, "provider_model": default_model} deduplicated: list[str] = [] for model in (default_model, *configured): if not isinstance(model, str): return {**data, "provider_model": default_model} normalized = model.strip() if normalized not in deduplicated: deduplicated.append(normalized) return {**data, "provider_model": default_model, "provider_models": tuple(deduplicated)} @property def should_bootstrap_schema(self) -> bool: """Return whether startup should auto-create schema for this environment.""" if "bootstrap_schema_on_startup" in self.model_fields_set: return self.bootstrap_schema_on_startup return self.environment in {"development", "test"} @cache def get_settings(**kwargs: Any) -> Settings: """Load cached settings without reading process CLI arguments.""" return Settings(_cli_parse_args=False, **kwargs) def parse_cli_settings(args: Sequence[str] | None = None) -> Settings: """Load settings with CLI arguments at the executable boundary.""" cli_args = True if args is None else list(args) return Settings(_cli_parse_args=cli_args) LOGGING_CONFIG: dict[str, Any] = { "version": 1, "disable_existing_loggers": False, "formatters": { "standard": { "format": "%(asctime)s %(levelname)-8s | %(message)s", "datefmt": "%Y-%m-%d %H:%M:%S", } }, "handlers": { "console": { "class": "logging.StreamHandler", "formatter": "standard", "stream": "ext://sys.stdout", }, "file": { "class": "logging.handlers.RotatingFileHandler", "formatter": "standard", "filename": str(Path("./data/logs") / "transcription.log"), "maxBytes": 10 * 1024 * 1024, "backupCount": 5, "encoding": "utf-8", } }, "root": { "level": "INFO", "handlers": ["console", "file"], }, "loggers": { "transcription": { "level": "DEBUG", "handlers": ["console", "file"], "propagate": False, } }, } def configure_logging(settings: Settings | None = None) -> None: """Configure root logging once at startup.""" cfg = copy.deepcopy(LOGGING_CONFIG) active_settings = settings or get_settings() active_settings.log_dir.mkdir(parents=True, exist_ok=True) file_handler = cfg["handlers"]["file"] file_handler["filename"] = str(active_settings.log_dir / active_settings.log_file_name) file_handler["maxBytes"] = active_settings.log_file_max_bytes file_handler["backupCount"] = active_settings.log_file_backup_count cfg["loggers"]["transcription"]["level"] = active_settings.log_level.upper() logging.config.dictConfig(cfg) logger.debug("Logging configured")