eryao/backend/src/core/config/settings.py

from __future__ import annotations

from pathlib import Path
from typing import ClassVar, Literal
from urllib.parse import quote

from pydantic import (
    AnyHttpUrl,
    BaseModel,
    Field,
    computed_field,
    field_validator,
    model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict


def _resolve_project_root() -> Path:
    current = Path(__file__).resolve()
    for parent in current.parents:
        if (
            (parent / "pyproject.toml").is_file()
            and (parent / "backend").is_dir()
            and (parent / "infra").is_dir()
        ):
            return parent

    for parent in current.parents:
        if parent.name == "backend":
            return parent.parent

    return Path.cwd().resolve()


class RuntimeSettings(BaseModel):
    environment: Literal["dev", "test", "prod"] = "dev"
    service_name: str = "app"
    debug: bool = True
    log_level: str = "INFO"
    log_json: bool = True
    log_rotation: Literal["time", "size", "none"] = "time"
    log_rotation_when: str = "midnight"
    log_rotation_interval: int = 1
    log_rotation_backup_count: int = 14
    log_rotation_max_bytes: int = 10_000_000
    log_dir: str = "logs"
    log_error_dir: str = "logs/errors"
    log_file_name: str = ""
    log_error_file_name: str = ""
    log_sensitive_fields: list[str] = Field(
        default_factory=lambda: [
            "password",
            "secret",
            "token",
            "api_key",
            "authorization",
            "cookie",
            "client_ip",
            "user_id",
        ]
    )
    sql_log_queries: bool = False
    trusted_proxy_ips: list[str] = Field(default_factory=list)

    @field_validator("log_dir", mode="before")
    @classmethod
    def lock_log_dir(cls, _: object) -> str:
        return "logs"

    @field_validator("log_error_dir", mode="before")
    @classmethod
    def lock_log_error_dir(cls, _: object) -> str:
        return "logs/errors"

    @model_validator(mode="after")
    def ensure_service_scoped_log_file_names(self) -> "RuntimeSettings":
        service = "".join(
            char if char.isalnum() or char in {"-", "_"} else "-"
            for char in self.service_name
        ).strip("-_")
        service_name = service or "app"

        if not self.log_file_name.strip():
            self.log_file_name = f"{service_name}.log"
        if not self.log_error_file_name.strip():
            self.log_error_file_name = f"{service_name}.error.log"

        return self


class CorsSettings(BaseModel):
    allow_origins: list[str] = Field(
        default_factory=lambda: [
            "http://localhost",
            "http://localhost:3000",
        ]
    )
    allow_credentials: bool = True
    allow_methods: list[str] = Field(default_factory=lambda: ["*"])
    allow_headers: list[str] = Field(default_factory=lambda: ["*"])


class RedisSettings(BaseModel):
    host: str = "redis"
    port: int = 6379
    password: str | None = None
    db: int = 0
    socket_connect_timeout: float = 1.0
    socket_timeout: float = 1.0
    max_connections: int = 10

    @computed_field
    @property
    def url(self) -> str:
        if self.password:
            password = quote(self.password, safe="")
            return f"redis://:{password}@{self.host}:{self.port}/{self.db}"
        return f"redis://{self.host}:{self.port}/{self.db}"


class DatabaseSettings(BaseModel):
    host: str = "localhost"
    port: int = 3306
    name: str = "eryao"
    user: str = "root"
    password: str = "CHANGE_ME"

    @computed_field
    @property
    def url(self) -> str:
        password = quote(self.password, safe="")
        return (
            f"mysql+aiomysql://{self.user}:{password}"
            f"@{self.host}:{self.port}/{self.name}"
        )


class AppVersionSettings(BaseModel):
    manifest_path: str = Field(
        default="deploy/static/releases/manifest.json",
        description="发布清单文件路径，相对于项目根目录",
    )
    release_path_prefix: str = Field(
        default="releases",
        description="下载 URL 中文件目录前缀",
    )
    download_base_url: AnyHttpUrl | None = Field(
        default=None,
        description="下载链接基础域名，如 https://your-domain.com",
    )

    @field_validator("download_base_url", mode="before")
    @classmethod
    def empty_download_base_url_to_none(cls, value: object) -> object:
        if value == "":
            return None
        return value

    @field_validator("manifest_path")
    @classmethod
    def validate_manifest_path(cls, value: str) -> str:
        normalized = Path(value)
        if normalized.is_absolute() or ".." in normalized.parts:
            raise ValueError("manifest_path must be a safe relative path")
        return value


class AliyunSmsSettings(BaseModel):
    access_key_id: str = "CHANGE_ME"
    access_key_secret: str = "CHANGE_ME"
    sign_name: str = "CHANGE_ME"
    template_code: str = "CHANGE_ME"
    region_id: str = "cn-hangzhou"
    endpoint: str = "dysmsapi.aliyuncs.com"
    test_mode: bool = False


class AliyunContentSecuritySettings(BaseModel):
    access_key_id: str = "CHANGE_ME"
    access_key_secret: str = "CHANGE_ME"
    endpoint: str = "green-cip.cn-shenzhen.aliyuncs.com"


class AlipaySettings(BaseModel):
    app_id: str = "CHANGE_ME"
    merchant_id: str = "CHANGE_ME"
    public_key: str = "CHANGE_ME"
    private_key: str = "CHANGE_ME"
    sign_type: str = "RSA2"
    notify_url: str = ""
    timeout_express: str = "30m"
    sandbox: bool = False


class DeepSeekSettings(BaseModel):
    api_key: str = "CHANGE_ME"


class AuthSettings(BaseModel):
    token_expiration_days: int = 7
    token_refresh_threshold_hours: int = 2


class VerificationSettings(BaseModel):
    code_length: int = 6
    expiration_minutes: int = 5
    test_mode: bool = False


class SensitiveWordSettings(BaseModel):
    use_aliyun: bool = True
    fallback_to_local: bool = True


class TestSettings(BaseModel):
    phone: str = ""
    password: str = ""


def _resolve_env_file() -> str:
    current = Path(__file__).resolve()
    for parent in [current, *current.parents]:
        candidate = parent / ".env"
        if candidate.is_file():
            return str(candidate)
    return ".env"


PROJECT_ROOT = _resolve_project_root()


class Settings(BaseSettings):
    runtime: RuntimeSettings = RuntimeSettings()
    cors: CorsSettings = CorsSettings()
    redis: RedisSettings = RedisSettings()
    database: DatabaseSettings = DatabaseSettings()
    app_version: AppVersionSettings = AppVersionSettings()
    aliyun_sms: AliyunSmsSettings = AliyunSmsSettings()
    aliyun_content_security: AliyunContentSecuritySettings = (
        AliyunContentSecuritySettings()
    )
    alipay: AlipaySettings = AlipaySettings()
    deepseek: DeepSeekSettings = DeepSeekSettings()
    auth: AuthSettings = AuthSettings()
    verification: VerificationSettings = VerificationSettings()
    sensitive_word: SensitiveWordSettings = SensitiveWordSettings()
    test: TestSettings = Field(default_factory=TestSettings)

    @computed_field
    @property
    def database_url(self) -> str:
        return self.database.url

    model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
        env_file=_resolve_env_file(),
        env_prefix="ERYAO_",
        env_nested_delimiter="__",
        case_sensitive=False,
        extra="ignore",
    )


config = Settings()  # type: ignore[reportCallIssue]