"""Typed runtime configuration loaded from non-secret environment values.""" from functools import lru_cache from pathlib import Path from typing import Literal, Self from urllib.parse import urlsplit from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from sqlalchemy import URL from app.core.secrets import read_secret_file class Settings(BaseSettings): """Application settings; secret values stay in mounted files.""" model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore", ) app_env: Literal["development", "test", "production"] = "development" app_name: str = "geological-rag" app_secret_key_file: Path = Path("/run/secrets/app_secret_key") postgres_host: str = "db" postgres_port: int = Field(default=5432, ge=1, le=65535) postgres_db: str = "geological_rag" postgres_user: str = "geological_rag_app" postgres_password_file: Path = Path("/run/secrets/postgres_app_password") upload_root: Path = Path("/data/uploads") max_upload_mb: int = Field(default=100, ge=1, le=2048) bailian_openai_base_url: str = ( "https://.cn-beijing.maas.aliyuncs.com/compatible-mode/v1" ) bailian_native_base_url: str = "https://.cn-beijing.maas.aliyuncs.com/api/v1" bailian_rerank_base_url: str = ( "https://.cn-beijing.maas.aliyuncs.com/compatible-api/v1" ) dashscope_api_key_file: Path = Path("/run/secrets/bailian_api_key") embedding_model: str = "text-embedding-v4" embedding_dimension: Literal[1024] = 1024 rerank_model: str = "qwen3-rerank" llm_model: str = "deepseek-v4-flash" chunk_target_tokens: int = Field(default=512, ge=120, le=800) chunk_max_tokens: int = Field(default=800, ge=120, le=4000) chunk_overlap_tokens: int = Field(default=64, ge=0, le=256) vector_top_k: int = Field(default=50, ge=1, le=500) rerank_top_n: int = Field(default=10, ge=1, le=500) context_top_n: int = Field(default=8, ge=1, le=50) max_context_tokens: int = Field(default=10_000, ge=1, le=100_000) model_timeout_seconds: float = Field(default=90, gt=0, le=600) model_max_retries: int = Field(default=3, ge=0, le=10) model_max_concurrency: int = Field(default=4, ge=1, le=100) worker_capabilities: str = "document_parse,embedding,rerank,evaluation" @field_validator( "bailian_openai_base_url", "bailian_native_base_url", "bailian_rerank_base_url", ) @classmethod def normalize_base_url(cls, value: str) -> str: return value.rstrip("/") @field_validator("embedding_dimension", mode="before") @classmethod def parse_embedding_dimension(cls, value: object) -> int: """Accept Compose's string representation without widening the contract.""" if value == "1024" or ( isinstance(value, int) and not isinstance(value, bool) and value == 1024 ): return 1024 raise ValueError("EMBEDDING_DIMENSION must be exactly 1024") @model_validator(mode="after") def validate_rag_limits(self) -> Self: if self.chunk_overlap_tokens >= self.chunk_target_tokens: raise ValueError("CHUNK_OVERLAP_TOKENS must be smaller than CHUNK_TARGET_TOKENS") if self.chunk_target_tokens > self.chunk_max_tokens: raise ValueError("CHUNK_TARGET_TOKENS must not exceed CHUNK_MAX_TOKENS") if self.context_top_n > self.rerank_top_n: raise ValueError("CONTEXT_TOP_N must not exceed RERANK_TOP_N") if self.rerank_top_n > self.vector_top_k: raise ValueError("RERANK_TOP_N must not exceed VECTOR_TOP_K") return self def bailian_api_key(self) -> str: self.validate_live_bailian_endpoints() return read_secret_file(self.dashscope_api_key_file) def validate_live_bailian_endpoints(self) -> str: endpoints = { self.bailian_openai_base_url: "/compatible-mode/v1", self.bailian_rerank_base_url: "/compatible-api/v1", } hosts: set[str] = set() for endpoint, expected_path in endpoints.items(): parsed = urlsplit(endpoint) host = parsed.hostname or "" if ( parsed.scheme != "https" or parsed.path.rstrip("/") != expected_path or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or "<" in endpoint or not host.endswith(".cn-beijing.maas.aliyuncs.com") ): raise ValueError("Bailian endpoint is not an approved Beijing MaaS URL") hosts.add(host) if len(hosts) != 1: raise ValueError("Bailian endpoints must use the same workspace host") return hosts.pop() def database_url(self) -> URL: return URL.create( drivername="postgresql+psycopg", username=self.postgres_user, password=read_secret_file(self.postgres_password_file), host=self.postgres_host, port=self.postgres_port, database=self.postgres_db, ) @lru_cache(maxsize=1) def get_settings() -> Settings: return Settings()