alien_py_cloud/alien_gateway/config.py

from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Any, Dict, List
from dotenv import load_dotenv
from urllib.parse import quote
import os

# Docker/Jenkins 挂载 /app/.env.produ；本地开发可同路径或环境变量覆盖
load_dotenv(".env.produ")


def _strip_env_quotes(value: str | None) -> str | None:
    if value is None:
        return None
    v = value.strip()
    if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"):
        return v[1:-1].strip()
    return v


class Settings(BaseSettings):
    # 基础配置
    PROJECT_NAME: str = "Alien Cloud Python"
    API_V1_STR: str = "/api/v1"

    # 鉴权配置 (原 alien-gateway 职责)
    SECRET_KEY: str = ""
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 10080

    # 数据库配置（生产须与 Java alien_produ 一致）
    DB_USER: str = "root"
    DB_PASSWORD: str = ""
    DB_HOST: str = ""
    DB_PORT: int = 3306
    DB_NAME: str = "alien_produ"

    # Redis Sentinel 高可用配置
    REDIS_SENTINELS: str = ""
    REDIS_MASTER_NAME: str = "mymaster"
    REDIS_PASSWORD: str = ""
    REDIS_DB: int = 0
    REDIS_SENTINEL_USERNAME: str = ""
    REDIS_SENTINEL_PASSWORD: str = ""
    REDIS_SOCKET_TIMEOUT: float = 0.5
    REDIS_CONNECT_TIMEOUT: float = 1.0
    STORE_BASE_URL: str = ""

    ALIYUN_SMS_SIGN_NAME_CONTRACT: str = ""
    ALIYUN_SMS_TEMPLATE_CODE_CONTRACT: str = ""
    ALIYUN_ACCESS_KEY_ID: str = ""
    ALIYUN_ACCESS_KEY_SECRET: str = ""

    @field_validator(
        "DB_HOST",
        "DB_USER",
        "DB_NAME",
        "DB_PASSWORD",
        "SECRET_KEY",
        "STORE_BASE_URL",
        mode="before",
    )
    @classmethod
    def strip_quoted_env(cls, value: Any) -> Any:
        if isinstance(value, str):
            return _strip_env_quotes(value) or ""
        return value



    @property
    def SQLALCHEMY_DATABASE_URI(self) -> str:
        return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"

    @property
    def REDIS_SENTINEL_NODES(self) -> List[tuple[str, int]]:
        nodes: List[tuple[str, int]] = []
        for item in self.REDIS_SENTINELS.split(","):
            entry = item.strip()
            if not entry:
                continue
            if ":" not in entry:
                raise ValueError(f"Invalid REDIS_SENTINELS entry: {entry}")
            host, port = entry.split(":", 1)
            nodes.append((host.strip(), int(port.strip())))
        return nodes

    @property
    def REDIS_SENTINEL_KWARGS(self) -> Dict[str, Any]:
        kwargs: Dict[str, Any] = {}
        if self.REDIS_SENTINEL_USERNAME:
            kwargs["username"] = self.REDIS_SENTINEL_USERNAME
        if self.REDIS_SENTINEL_PASSWORD:
            kwargs["password"] = self.REDIS_SENTINEL_PASSWORD
        return kwargs

    @property
    def REDIS_SENTINEL_URL(self) -> str:
        # 标准 Sentinel URL:
        # redis+sentinel://[username[:password]@]host1:port1,host2:port2/db?sentinel_master=mymaster
        if self.REDIS_SENTINEL_USERNAME:
            user = quote(self.REDIS_SENTINEL_USERNAME, safe="")
            pwd = quote(self.REDIS_PASSWORD, safe="")
            auth = f"{user}:{pwd}@"
        elif self.REDIS_PASSWORD:
            pwd = quote(self.REDIS_PASSWORD, safe="")
            auth = f":{pwd}@"
        else:
            auth = ""
        hosts = ",".join(f"{host}:{port}" for host, port in self.REDIS_SENTINEL_NODES)
        master = quote(self.REDIS_MASTER_NAME, safe="")
        return f"redis+sentinel://{auth}{hosts}/{self.REDIS_DB}?sentinel_master={master}"

    @property
    def REDIS_CELERY_SENTINEL_URL(self) -> str:
        # Celery/Kombu Sentinel URL 格式：
        # sentinel://:redis_password@sentinel_host:port/db;...
        # URL 中的 auth 密码 = Redis 主从节点密码
        # （Kombu 通过 Sentinel 发现主节点地址后，用此密码向 Redis 主节点认证）
        # Sentinel 自身的密码（如有）通过 broker_transport_options["sentinel_kwargs"]["password"] 传递
        auth = f":{quote(self.REDIS_PASSWORD, safe='')}@" if self.REDIS_PASSWORD else ""
        return ";".join(
            f"sentinel://{auth}{host}:{port}/{self.REDIS_DB}"
            for host, port in self.REDIS_SENTINEL_NODES
        )

    @property
    def REDIS_SENTINEL_TRANSPORT_OPTIONS(self) -> Dict[str, Any]:
        return {
            "master_name": self.REDIS_MASTER_NAME,
            "sentinel_kwargs": self.REDIS_SENTINEL_KWARGS,
            "password": self.REDIS_PASSWORD,
            "db": self.REDIS_DB,
        }

    model_config = SettingsConfigDict(
        case_sensitive=True,
        env_file=".env.produ",
        extra="ignore",
    )

settings = Settings()