dieya_risk_prediction_backend/app/core/config.py

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

from functools import lru_cache
from pathlib import Path
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 = ""
    baidu_ocr_api_key: SecretStr | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("BAIDU_OCR_API_KEY", "RISK_BAIDU_OCR_API_KEY"),
    )
    baidu_ocr_secret_key: SecretStr | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("BAIDU_OCR_SECRET_KEY", "RISK_BAIDU_OCR_SECRET_KEY"),
    )
    baidu_ocr_image_url: str = "https://aip.baidubce.com/rest/2.0/ocr/v1/webimage"
    baidu_ocr_document_url: str = "https://aip.baidubce.com/rest/2.0/brain/online/v2/paddle-vl-parser/task"
    baidu_ocr_timeout_seconds: int = Field(default=75, ge=10, le=300)
    baidu_ocr_poll_interval_seconds: float = Field(default=2.0, ge=0.5, le=10)
    ocr_upload_dir: Path = Path(".runtime-uploads/ocr")
    ocr_max_file_mb: int = Field(default=10, ge=1, le=20)
    ocr_max_batch_files: int = Field(default=5, ge=1, le=10)
    ocr_review_confidence: float = Field(default=0.8, ge=0, le=1)
    ocr_judgment_confidence: float = Field(default=0.85, ge=0, le=1)
    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"),
    )
    rag_embedding_api_key: SecretStr | None = Field(
        default=None,
        repr=False,
        validation_alias=AliasChoices("RISK_RAG_EMBEDDING_API_KEY", "EMBEDDING_API_KEY"),
    )
    rag_embedding_base_url: str = Field(
        default="",
        validation_alias=AliasChoices("RISK_RAG_EMBEDDING_BASE_URL", "EMBEDDING_BASE_URL"),
    )
    rag_embedding_model: str = Field(
        default="bge-m3",
        validation_alias=AliasChoices("RISK_RAG_EMBEDDING_MODEL", "EMBEDDING_MODEL"),
    )
    rag_embedding_dimensions: int = Field(
        default=3072,
        ge=1,
        le=4096,
        validation_alias=AliasChoices("RISK_RAG_EMBEDDING_DIMENSIONS", "EMBEDDING_DIMENSIONS"),
    )
    rag_embedding_timeout_seconds: int = Field(
        default=10,
        gt=0,
        le=60,
        validation_alias=AliasChoices("RISK_RAG_EMBEDDING_TIMEOUT_SECONDS", "EMBEDDING_TIMEOUT"),
    )
    rag_top_k: int = Field(
        default=5,
        ge=1,
        le=20,
        validation_alias=AliasChoices("RISK_RAG_TOP_K", "RAG_TOP_K"),
    )
    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("rag_embedding_base_url")
    @classmethod
    def normalize_rag_embedding_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()