dieya_risk_prediction_backend/app/core/config.py

"""Environment-backed application settings."""

from functools import lru_cache
from typing import Annotated, Literal

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


class Settings(BaseSettings):
    app_name: str = "Dieya Risk Prediction Backend"
    app_version: str = "0.1.0"
    environment: str = "development"
    debug: bool = False
    database_url: str = Field(
        default="postgresql+asyncpg://postgres:postgres@localhost:5432/dieya_risk_prediction",
        repr=False,
    )
    log_level: str = "INFO"
    jwt_secret_key: SecretStr = SecretStr("CHANGE_THIS_IN_PRODUCTION")
    jwt_algorithm: str = "HS256"
    jwt_expire_minutes: int = Field(default=480, gt=0)
    dingtalk_app_key: str = ""
    dingtalk_app_secret: SecretStr | None = Field(default=None, repr=False)
    dingtalk_corp_id: str = ""
    dingtalk_agent_id: str = ""
    llm_api_key: SecretStr | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices(
            "RISK_LLM_API_KEY",
            "OPENAI_API_KEY",
            "YUNWU_API_KEY",
        ),
    )
    llm_base_url: str = Field(
        default="https://api.openai.com/v1",
        validation_alias=AliasChoices("RISK_LLM_BASE_URL", "LLM_BASE_URL", "OPENAI_API_BASE"),
    )
    llm_model: str = Field(
        default="gpt-4o-mini",
        validation_alias=AliasChoices("RISK_LLM_MODEL", "LLM_MODEL"),
    )
    llm_timeout_seconds: int = Field(
        default=120,
        gt=0,
        le=600,
        validation_alias=AliasChoices("RISK_LLM_TIMEOUT_SECONDS", "LLM_TIMEOUT"),
    )
    llm_temperature: float = Field(
        default=0.2,
        ge=0,
        le=2,
        validation_alias=AliasChoices("RISK_LLM_TEMPERATURE", "LLM_TEMPERATURE"),
    )
    llm_fallback_api_key: SecretStr | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_API_KEY", "DEEPSEEK_API_KEY"),
    )
    llm_fallback_base_url: str = Field(
        default="https://api.deepseek.com",
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_BASE_URL", "DEEPSEEK_BASE_URL"),
    )
    llm_fallback_model: str = Field(
        default="deepseek-v4-pro",
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_MODEL", "DEEPSEEK_MODEL"),
    )
    llm_fallback_api_mode: Literal["chat_completions", "responses"] = Field(
        default="responses",
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_API_MODE", "DEEPSEEK_API_MODE"),
    )
    llm_fallback_timeout_seconds: int = Field(
        default=30,
        gt=0,
        le=600,
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_TIMEOUT_SECONDS", "DEEPSEEK_TIMEOUT"),
    )
    llm_fallback_max_retries: int = Field(
        default=1,
        ge=0,
        le=10,
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_MAX_RETRIES", "DEEPSEEK_MAX_RETRIES"),
    )
    llm_fallback_reasoning_effort: Literal["low", "high", "max"] = Field(
        default="high",
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_REASONING_EFFORT", "DEEPSEEK_REASONING_EFFORT"),
    )
    llm_fallback_max_output_tokens: int = Field(
        default=16_384,
        ge=1,
        le=393_216,
        validation_alias=AliasChoices("RISK_LLM_FALLBACK_MAX_OUTPUT_TOKENS", "DEEPSEEK_MAX_OUTPUT_TOKENS"),
    )
    cors_allow_origins: Annotated[tuple[str, ...], NoDecode] = (
        "http://localhost:39158",
        "http://127.0.0.1:39158",
    )

    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="RISK_",
        extra="ignore",
        case_sensitive=False,
    )

    @field_validator("database_url")
    @classmethod
    def validate_database_url(cls, value: str) -> str:
        if not value.startswith("postgresql+asyncpg://"):
            raise ValueError("database_url must use the postgresql+asyncpg:// scheme")
        return value

    @field_validator("llm_base_url")
    @classmethod
    def normalize_llm_base_url(cls, value: str) -> str:
        return value.rstrip("/")

    @field_validator("llm_fallback_base_url")
    @classmethod
    def normalize_llm_fallback_base_url(cls, value: str) -> str:
        return value.rstrip("/")

    @field_validator("cors_allow_origins", mode="before")
    @classmethod
    def split_cors_origins(cls, value: object) -> object:
        if isinstance(value, str):
            value = tuple(item.strip() for item in value.split(",") if item.strip())
        if isinstance(value, (list, tuple)) and "*" in value:
            raise ValueError("cors_allow_origins must not contain '*' wildcard")
        return value


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()