alien_py_cloud/alien_gateway/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import quote
import os

# 通过环境变量 APP_ENV 选择 .env 文件，默认 dev：
# - 本地开发：APP_ENV=dev  -> .env.dev
# - 测试环境：APP_ENV=sit  -> .env.sit
# - UAT环境：APP_ENV=uat   -> .env.uat
APP_ENV = os.getenv("APP_ENV", "dev")
_ENV_FILE = f".env.{APP_ENV}"


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

    # 鉴权配置（原 alien-gateway 职责）
    SECRET_KEY: str = os.getenv("SECRET_KEY", "")
    ALGORITHM: str = os.getenv("ALGORITHM", "HS256")
    ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "10080"))

    # 数据库配置
    DB_USER: str = os.getenv("DB_USER", "")
    DB_PASSWORD: str = os.getenv("DB_PASSWORD", "")
    DB_HOST: str = os.getenv("DB_HOST", "")
    DB_PORT: int = int(os.getenv("DB_PORT", "3306"))
    DB_NAME: str = os.getenv("DB_NAME", "")

    # Redis 单机配置（与 Sentinel 二选一；REDIS_URL 非空时走单机模式）
    REDIS_URL: str = os.getenv("REDIS_URL", "")

    # Redis Sentinel 高可用配置（REDIS_SENTINELS 非空时走哨兵模式）
    # 例：REDIS_SENTINELS=192.168.2.251:36379,192.168.2.252:36379,192.168.2.253:36379
    REDIS_SENTINELS: str = os.getenv("REDIS_SENTINELS", "")
    REDIS_MASTER_NAME: str = os.getenv("REDIS_MASTER_NAME", "mymaster")
    REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD", "")
    REDIS_DB: int = int(os.getenv("REDIS_DB", "0"))
    REDIS_SENTINEL_USERNAME: str = os.getenv("REDIS_SENTINEL_USERNAME", "")
    REDIS_SENTINEL_PASSWORD: str = os.getenv("REDIS_SENTINEL_PASSWORD", "")
    REDIS_SOCKET_TIMEOUT: float = float(os.getenv("REDIS_SOCKET_TIMEOUT", "0.5"))
    REDIS_CONNECT_TIMEOUT: float = float(os.getenv("REDIS_CONNECT_TIMEOUT", "1.0"))

    # 各服务监听端口（容器内端口；外部映射由 Jenkinsfile / docker-compose 控制）
    GATEWAY_PORT: int = int(os.getenv("GATEWAY_PORT", "33333"))
    STORE_PORT: int = int(os.getenv("STORE_PORT", "8001"))
    CONTRACT_PORT: int = int(os.getenv("CONTRACT_PORT", "8002"))
    LAWYER_PORT: int = int(os.getenv("LAWYER_PORT", "8004"))

    # 下游服务地址（网关反向代理用）
    STORE_BASE_URL: str = os.getenv("STORE_BASE_URL", "http://127.0.0.1:8001")
    CONTRACT_BASE_URL: str = os.getenv("CONTRACT_BASE_URL", "http://127.0.0.1:8002")

    # 阿里云短信配置
    ALIYUN_SMS_SIGN_NAME_CONTRACT: str = os.getenv("ALIYUN_SMS_SIGN_NAME_CONTRACT", "")
    ALIYUN_SMS_TEMPLATE_CODE_CONTRACT: str = os.getenv("ALIYUN_SMS_TEMPLATE_CODE_CONTRACT", "")
    ALIYUN_ACCESS_KEY_ID: str = os.getenv("ALIYUN_ACCESS_KEY_ID", "")
    ALIYUN_ACCESS_KEY_SECRET: str = os.getenv("ALIYUN_ACCESS_KEY_SECRET", "")

    @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_MODE(self) -> str:
        # standalone（单机） / sentinel（哨兵）
        return "sentinel" if self.REDIS_SENTINELS.strip() else "standalone"

    @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;...
        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,
        }

    @property
    def CELERY_BROKER_URL(self) -> str:
        # 单机优先用 REDIS_URL；哨兵则用 Celery Sentinel URL
        return self.REDIS_CELERY_SENTINEL_URL if self.REDIS_MODE == "sentinel" else self.REDIS_URL

    @property
    def CELERY_BACKEND_URL(self) -> str:
        return self.CELERY_BROKER_URL

    @property
    def CELERY_TRANSPORT_OPTIONS(self) -> Optional[Dict[str, Any]]:
        return self.REDIS_SENTINEL_TRANSPORT_OPTIONS if self.REDIS_MODE == "sentinel" else None

    model_config = SettingsConfigDict(
        case_sensitive=True,
        env_file=_ENV_FILE,
        extra="ignore",
    )


settings = Settings()