"""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" 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" ) # DeepSeek 自 2026-08-17 起峰谷定价，高峰单价是空闲时段的 2 倍 # （高峰：北京时间 09:00-12:00、14:00-18:00）。留空则两个时段同价。 llm_cost_rates_peak: dict[str, dict[str, Decimal]] = Field( default_factory=dict, alias="LLM_COST_RATES_PEAK_JSON" ) llm_peak_pricing_enabled: bool = True 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 # 四档分级降级（架构 §13.3，T-P1-25b）。JSON 为列表： # [{"threshold":0.8,"name":"degrade_secondary","blocked":["hd_export"],...}] # blocked 支持 "*"; allow 仅 block_all 档有意义（纯文本对话放行）。 llm_budget_tiers: list[dict[str, Any]] = Field( default_factory=lambda: [ {"threshold": 0.6, "name": "warn", "blocked": []}, { "threshold": 0.8, "name": "degrade_secondary", "blocked": ["hd_export", "report_regen"], }, {"threshold": 0.9, "name": "degrade_generation", "blocked": ["hairstyle", "tryon"]}, {"threshold": 1.0, "name": "block_all", "blocked": "*", "allow": ["agent_chat"]}, ], alias="LLM_BUDGET_TIERS_JSON", ) # --- Auth: refresh sessions (FR-ACC-05) -------------------------------- refresh_token_ttl_seconds: int = 2_592_000 # 30 天 # --- 任务奖励（FR-BILL-03/04，配置化）----------------------------------- # amount 全部是占位值 1：正式数值待产品在产品方案 §3.2 定稿后改 env，不改代码。 # claim_period 决定幂等粒度（同一任务同周期只能领一次）。 task_bonus_rules: dict[str, dict[str, Any]] = Field( default_factory=lambda: { "sign_in": {"quota_key": "agent_chat", "amount": 1, "claim_period": "day"}, "profile_complete": { "quota_key": "agent_chat", "amount": 1, "claim_period": "lifetime", }, "report_feedback": {"quota_key": "report_regen", "amount": 1, "claim_period": "day"}, "share_report": {"quota_key": "tryon", "amount": 1, "claim_period": "day"}, "invite_friend": {"quota_key": "tryon", "amount": 1, "claim_period": "day"}, }, alias="TASK_BONUS_RULES_JSON", ) # --- 客户端埋点（FR-SYS-06）--------------------------------------------- events_max_batch: int = 50 events_props_max_bytes: int = 8192 # --- 腾讯云 COS 直传（FR-SYS-03 / T-P1-27）------------------------------ cos_region: str = "ap-shanghai" cos_secret_id: str = "" cos_secret_key: str = "" cos_bucket_temp: str = "meida-temp" # 临时桶，2 小时生命周期（CR-02） cos_bucket_user: str = "meida-user" cos_bucket_kb: str = "meida-kb" cos_bucket_report: str = "meida-report" cos_presign_expire_seconds: int = 600 upload_max_bytes: int = 20_971_520 # 20MB upload_allowed_types: str = "image/jpeg,image/png,image/webp" # scene → {"bucket": "temp"|"user"|"kb"|"report"}。新增上传场景只改这里。 upload_scenes: dict[str, dict[str, Any]] = Field( default_factory=lambda: { "report_photo": {"bucket": "temp"}, "avatar": {"bucket": "temp"}, }, alias="UPLOAD_SCENES_JSON", ) # --- 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_cost_rates_peak", "llm_default_cost_rate", "llm_budget_tiers", "task_bonus_rules", "upload_scenes", 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()