harness_zbt/backend/src/zbt/core/config.py

"""集中读取和校验项目配置。

配置优先从后端目录的 ``.env`` 读取；若某项未配置，则使用字段右侧的默认值。
Pydantic 会自动把环境变量中的字符串转换成声明的 Python 类型。
"""

from functools import lru_cache
from pathlib import Path
from typing import Literal

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


def resolve_env_file(start_directory: Path | None = None) -> Path:
    """兼容从 backend、项目根目录和可编辑安装环境启动。"""

    start = (start_directory or Path.cwd()).resolve()
    editable_backend = Path(__file__).resolve().parents[3]
    candidates = (
        start / ".env",
        start / "backend" / ".env",
        editable_backend / ".env",
    )
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    return start / ".env"


ENV_FILE = resolve_env_file()
BACKEND_ROOT = ENV_FILE.parent


class Settings(BaseSettings):
    """应用运行时需要的全部配置及其默认值。"""

    model_config = SettingsConfigDict(
        env_file=ENV_FILE,
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
        populate_by_name=True,
    )

    # Literal 限制只能使用列出的三种环境名称，拼写错误会在启动时直接报错。
    app_env: Literal["development", "test", "production"] = "development"
    app_host: str = "127.0.0.1"
    app_port: int = 8000
    app_timezone: str = "Asia/Shanghai"
    app_cors_origins: list[str] = [
        "http://127.0.0.1:5173",
        "http://127.0.0.1:5174",
    ]

    mysql_host: str = "127.0.0.1"
    mysql_port: int = 3306
    mysql_user: str = "zbt_s2"
    mysql_password: str = ""
    mysql_core_database: str = "insurance_s2_core"
    mysql_agent_database: str = "insurance_s2_agent"
    mysql_analytics_database: str = "insurance_s2_analytics"

    redis_url: str = "redis://127.0.0.1:6379/0"
    redis_prefix: str = "ins:s2:"
    agent_rate_limit: int = Field(default=20, ge=1, le=1000)
    agent_rate_window_seconds: int = Field(default=60, ge=1, le=3600)
    agent_hot_state_ttl_seconds: int = Field(default=3600, ge=60, le=86400)
    agent_event_ttl_seconds: int = Field(default=1800, ge=60, le=86400)

    milvus_uri: str = "http://127.0.0.1:19530"
    milvus_token: str = ""
    milvus_collection_prefix: str = "ins_s2_"
    bge_m3_model_path: str = "BAAI/bge-m3"
    bge_m3_device: str = "cpu"
    bge_reranker_model_path: str = "BAAI/bge-reranker-v2-m3"
    bge_reranker_device: str = "cpu"
    embedding_dimension: int = Field(default=1024, ge=128, le=4096)
    knowledge_candidate_multiplier: int = Field(default=4, ge=2, le=20)

    deepseek_api_key: str = ""
    deepseek_base_url: str = "https://api.deepseek.com"
    deepseek_model: str = "deepseek-chat"
    dashscope_api_key: str = Field(
        default="",
        validation_alias=AliasChoices("DASHSCOPE_API_KEY", "QWEN_API_KEY"),
    )
    dashscope_base_url: str = Field(
        default="https://dashscope.aliyuncs.com/compatible-mode/v1",
        validation_alias=AliasChoices("DASHSCOPE_BASE_URL", "QWEN_BASE_URL"),
    )
    dashscope_vision_model: str = Field(
        default="qwen3.5-plus",
        validation_alias=AliasChoices("DASHSCOPE_VISION_MODEL", "QWEN_VISION_MODEL"),
    )
    service_material_storage_path: Path = BACKEND_ROOT / "var" / "service-materials"
    langsmith_api_key: str = ""
    langsmith_endpoint: str = "https://api.smith.langchain.com"
    langsmith_project: str = "智保通-第二阶段-知识服务与数据分析"
    langsmith_tracing: bool = True

    jwt_access_secret: str = Field(default="development-access-secret-change-me", min_length=32)
    jwt_refresh_secret: str = Field(default="development-refresh-secret-change-me", min_length=32)
    field_encryption_key: str = Field(default="development-field-key-change-me-now", min_length=32)
    dev_fixed_otp: str = Field(default="147258", pattern=r"^\d{6}$")
    dev_tools_enabled: bool = True
    mock_payment_callback_secret: str = Field(
        default="development-mock-payment-secret-change-me",
        min_length=32,
    )

    access_token_minutes: int = 15
    refresh_token_days: int = 7

    # 以下验证器限制程序只能访问本项目约定的数据库，避免误连或误操作其他库。
    @field_validator("mysql_core_database")
    @classmethod
    def validate_core_database(cls, value: str) -> str:
        if value not in {"insurance_s2_core", "insurance_s2_test_core"}:
            raise ValueError("第二阶段Core数据库名称不在白名单")
        return value

    @field_validator("mysql_agent_database")
    @classmethod
    def validate_agent_database(cls, value: str) -> str:
        if value not in {"insurance_s2_agent", "insurance_s2_test_agent"}:
            raise ValueError("第二阶段Agent数据库名称不在白名单")
        return value

    @field_validator("mysql_analytics_database")
    @classmethod
    def validate_analytics_database(cls, value: str) -> str:
        if value not in {"insurance_s2_analytics", "insurance_s2_test_analytics"}:
            raise ValueError("第二阶段Analytics数据库名称不在白名单")
        return value

    @field_validator("redis_prefix")
    @classmethod
    def validate_redis_prefix(cls, value: str) -> str:
        if value != "ins:s2:" and not value.startswith("ins:s2:test:"):
            raise ValueError("第二阶段Redis前缀不在白名单")
        return value

    @field_validator("milvus_collection_prefix")
    @classmethod
    def validate_milvus_prefix(cls, value: str) -> str:
        if not value.startswith("ins_s2_"):
            raise ValueError("第二阶段Milvus集合前缀不在白名单")
        return value


@lru_cache
def get_settings() -> Settings:
    """创建并缓存配置对象，后续调用复用同一个实例。"""
    return Settings()