meida_backend/packages/shared/config.py

"""Typed application configuration.

Every value comes from the environment (架构文档 §11.2: 密钥不进代码/镜像).
`.env` is only a developer convenience; dev/staging/prod inject via SSM/KMS.
"""

from __future__ import annotations

import json
from decimal import Decimal
from functools import lru_cache
from typing import Any, Literal

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

Environment = Literal["local", "dev", "staging", "prod"]
CachePrefixMode = Literal["strict", "warn", "off"]

_PLACEHOLDER_JWT_SECRET = "change-me-local-only-do-not-use-outside-local"  # noqa: S105


class Settings(BaseSettings):
    """Root settings object. Read it through :func:`get_settings`."""

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

    # --- Application -------------------------------------------------------
    app_env: Environment = "local"
    app_name: str = "meida-backend"
    api_prefix: str = "/api/v1"
    debug: bool = False
    log_level: str = "INFO"
    log_json: bool = True
    log_extra_redact_keys: str = ""

    # --- Database ----------------------------------------------------------
    database_url: str = "postgresql+asyncpg://meida:meida@localhost:55432/meida"
    db_pool_size: int = 10
    db_max_overflow: int = 20
    db_echo: bool = False

    # --- Redis / Celery ----------------------------------------------------
    redis_url: str = "redis://localhost:56379/0"
    celery_broker_url: str = "redis://localhost:56379/1"
    celery_result_backend: str = "redis://localhost:56379/2"

    # --- Auth --------------------------------------------------------------
    jwt_secret: str = _PLACEHOLDER_JWT_SECRET
    jwt_algorithm: str = "HS256"
    jwt_access_ttl_seconds: int = 900
    jwt_issuer: str = "meida"
    dev_auth_enabled: bool = True

    # --- LLM routing -------------------------------------------------------
    llm_primary_provider: str = "deepseek"
    llm_backup_provider: str = "zhipu"
    llm_structured_max_retries: int = 2
    llm_circuit_failure_threshold: int = 3
    llm_circuit_cooldown_seconds: float = 30.0
    llm_scene_overrides: dict[str, dict[str, Any]] = Field(
        default_factory=dict, alias="LLM_SCENE_OVERRIDES_JSON"
    )

    # --- DeepSeek (primary, OpenAI-compatible) -----------------------------
    deepseek_api_key: str = ""
    deepseek_base_url: str = "https://api.deepseek.com/v1"
    deepseek_model: str = "deepseek-v4-flash"
    deepseek_timeout_seconds: float = 30.0

    # --- 智谱 GLM (backup slot, not implemented in V1) ---------------------
    zhipu_api_key: str = ""
    zhipu_base_url: str = "https://open.bigmodel.cn/api/paas/v4"
    zhipu_model: str = "glm-4-flash"
    zhipu_timeout_seconds: float = 30.0

    # --- Prompt caching discipline (架构文档 §7.2) -------------------------
    llm_cache_prefix_mode: CachePrefixMode = "strict"
    prompts_dir: str = "prompts"

    # --- Cost attribution (架构文档 §13) -----------------------------------
    llm_cost_rates: dict[str, dict[str, Decimal]] = Field(
        default_factory=dict, alias="LLM_COST_RATES_JSON"
    )
    llm_default_cost_rate: dict[str, Decimal] = Field(
        default_factory=lambda: {
            "input_cents_per_mtok": Decimal("100"),
            "cached_input_cents_per_mtok": Decimal("20"),
            "output_cents_per_mtok": Decimal("200"),
        },
        alias="LLM_DEFAULT_COST_RATE_JSON",
    )

    # --- Global daily cost circuit breaker ---------------------------------
    llm_daily_budget_enabled: bool = True
    llm_daily_budget_cents: Decimal = Decimal("100000")
    llm_budget_timezone_offset_hours: int = 8

    # --- Observability -----------------------------------------------------
    otel_enabled: bool = False
    otel_service_name: str = "meida-api"
    otel_exporter_otlp_endpoint: str = ""
    prometheus_enabled: bool = True

    @field_validator(
        "llm_scene_overrides", "llm_cost_rates", "llm_default_cost_rate", mode="before"
    )
    @classmethod
    def _parse_json_object(cls, value: Any) -> Any:
        if isinstance(value, str):
            stripped = value.strip()
            if not stripped:
                return {}
            return json.loads(stripped)
        return value

    @field_validator("jwt_secret")
    @classmethod
    def _reject_placeholder_secret_outside_local(cls, value: str, info: ValidationInfo) -> str:
        env = info.data.get("app_env", "local")
        if env in ("staging", "prod") and (value == _PLACEHOLDER_JWT_SECRET or len(value) < 32):
            msg = "JWT_SECRET must be set to a >=32 char random value outside local/dev"
            raise ValueError(msg)
        return value

    @property
    def is_local(self) -> bool:
        return self.app_env in ("local", "dev")

    @property
    def extra_redact_keys(self) -> tuple[str, ...]:
        return tuple(k.strip() for k in self.log_extra_redact_keys.split(",") if k.strip())

    @property
    def sync_database_url(self) -> str:
        """psycopg-style URL, used by tooling that cannot speak asyncpg."""
        return self.database_url.replace("+asyncpg", "")


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """Process-wide settings singleton."""
    return Settings()