meijiaka-zy/python-api/app/config.py

"""
配置管理 - Pydantic Settings
==========================

所有配置项通过环境变量或 .env 文件注入。
"""

from functools import lru_cache
from typing import Literal

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """应用配置"""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
        arbitrary_types_allowed=True,
    )

    # 应用基础配置
    APP_NAME: str = Field(default="美家卡智影 API", description="应用名称")
    APP_VERSION: str = Field(default="0.1.0", description="应用版本")
    DEBUG: bool = Field(default=True, description="调试模式")
    ENV: Literal["development", "staging", "production"] = Field(
        default="development", description="运行环境"
    )

    # 服务器配置
    HOST: str = Field(default="0.0.0.0", description="监听地址")
    PORT: int = Field(default=8000, description="监听端口")
    APP_BASE_URL: str = Field(
        default="",
        description="应用公网地址（用于第三方回调，如 https://dev.tapi.meijiaka.cn）",
    )
    WORKERS: int = Field(default=1, description="工作进程数（生产环境建议 > 1）")

    # 数据库配置（统一使用 PostgreSQL）
    DATABASE_URL: str = Field(
        default="postgresql+asyncpg://postgres:postgres@localhost:5432/meijiaka_zy",
        description="数据库连接字符串（PostgreSQL）",
    )
    DATABASE_POOL_SIZE: int = Field(default=10, description="数据库连接池大小")
    DATABASE_MAX_OVERFLOW: int = Field(default=20, description="连接池溢出上限")

    # Redis 配置
    REDIS_HOST: str = Field(
        default="localhost",
        description="Redis 主机地址",
    )
    REDIS_PORT: int = Field(
        default=6379,
        description="Redis 端口",
    )
    REDIS_DB: int = Field(
        default=0,
        description="Redis 数据库编号",
    )
    REDIS_PASSWORD: str | None = Field(
        default=None,
        description="Redis 密码（无密码请留空）",
    )

    # 安全配置
    SECRET_KEY: str = Field(
        default="your-secret-key-here-change-in-production",
        description="JWT 签名密钥（生产环境必须修改）",
    )
    ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(
        default=60 * 24 * 7,  # 7 天
        description="访问令牌过期时间（分钟）",
    )
    ALGORITHM: str = Field(default="HS256", description="JWT 算法")

    # CORS 配置
    CORS_ORIGINS: str = Field(
        default="http://localhost:1420,http://127.0.0.1:1420,http://localhost:8080,http://127.0.0.1:8080",
        description="允许的跨域来源（逗号分隔）",
    )

    # AI 模型配置
    # 字节跳动 - 火山方舟
    # 文档：https://www.volcengine.com/docs/82379/1399009
    VOLCENGINE_API_KEY: str | None = Field(default=None, description="火山方舟 API Key")
    VOLCENGINE_BASE_URL: str = Field(
        default="https://ark.cn-beijing.volces.com/api/v3",
        description="火山方舟 Base URL",
    )
    VOLCENGINE_MODEL: str = Field(
        default="doubao-seed-2-0-lite-260215",
        description="火山方舟默认模型（Model ID）",
    )

    # 火山引擎音视频字幕服务（字幕功能仍使用火山引擎）
    VOLCENGINE_CAPTION_APPID: str | None = Field(default=None, description="火山字幕 AppID")
    VOLCENGINE_CAPTION_TOKEN: str | None = Field(default=None, description="火山字幕 Token")

    # OpenAI
    OPENAI_API_KEY: str | None = Field(default=None, description="OpenAI API Key")
    OPENAI_BASE_URL: str = Field(default="https://api.openai.com/v1", description="OpenAI Base URL")
    OPENAI_DEFAULT_MODEL: str = Field(default="gpt-3.5-turbo", description="默认 OpenAI 模型")

    # 文心一言 (百度)
    WENXIN_API_KEY: str | None = Field(default=None, description="文心一言 API Key")
    WENXIN_SECRET_KEY: str | None = Field(default=None, description="文心一言 Secret Key")

    # 通义千问 (阿里云)
    QIANWEN_API_KEY: str | None = Field(default=None, description="通义千问 API Key")

    # 数字人服务配置
    DIGITAL_HUMAN_PROVIDER: Literal["heygen", "did", "mock"] = Field(
        default="mock",
        description="数字人服务提供商",
    )
    HEYGEN_API_KEY: str | None = Field(default=None, description="HeyGen API Key")
    DID_API_KEY: str | None = Field(default=None, description="D-ID API Key")

    # Vidu 配置
    VIDU_API_KEY: str | None = Field(default=None, description="Vidu API Key")
    VIDU_BASE_URL: str = Field(
        default="https://api.vidu.cn",
        description="Vidu Base URL",
    )

    # 七牛云存储配置
    QINIU_ACCESS_KEY: str | None = Field(default=None, description="七牛云 Access Key")
    QINIU_SECRET_KEY: str | None = Field(default=None, description="七牛云 Secret Key")
    QINIU_VIDEO_BUCKET: str = Field(default="media-liche", description="视频存储 Bucket")
    QINIU_VIDEO_DOMAIN: str = Field(default="media.liche.cn", description="视频存储域名")
    QINIU_IMAGE_BUCKET: str = Field(default="img-liche", description="图片存储 Bucket")
    QINIU_IMAGE_DOMAIN: str = Field(default="img.liche.cn", description="图片存储域名")

    # AnyToCopy 文案提取服务
    ANYTOCOPY_API_KEY: str | None = Field(default=None, description="AnyToCopy API Key")
    ANYTOCOPY_API_SECRET: str | None = Field(default=None, description="AnyToCopy API Secret")
    ANYTOCOPY_BASE_URL: str = Field(
        default="https://api.anytocopy.com/vip/open-api/v1",
        description="AnyToCopy Base URL",
    )

    # Async Engine 槽位配置
    ANYTOCOPY_MAX_CONCURRENT: int = Field(default=5, description="AnyToCopy文案提取最大并发数")
    VOLC_SUBTITLE_MAX_CONCURRENT: int = Field(default=5, description="火山字幕生成最大并发数")

    # 任务超时配置（秒）
    VOLC_SUBTITLE_TIMEOUT: int = Field(default=600, description="火山字幕生成超时时间（秒）")

    # AnyToCopy 轮询配置
    ANYTOCOPY_POLL_INTERVAL: float = Field(default=3.0, description="AnyToCopy轮询间隔（秒）")
    ANYTOCOPY_MAX_POLL: int = Field(default=60, description="AnyToCopy最大轮询次数")

    # 日志配置
    LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = Field(
        default="DEBUG",
        description="日志级别",
    )

    @property
    def cors_origins_list(self) -> list[str]:
        """将 CORS_ORIGINS 字符串解析为列表"""
        return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]

    @property
    def app_base_url(self) -> str:
        """应用公网地址（用于第三方回调）"""
        if self.APP_BASE_URL:
            return self.APP_BASE_URL.rstrip("/")
        if self.ENV == "production":
            return "https://tapi.meijiaka.cn"
        if self.ENV == "staging":
            return "https://dev.tapi.meijiaka.cn"
        return f"http://{self.HOST}:{self.PORT}"

    @property
    def use_redis(self) -> bool:
        """是否使用 Redis"""
        return bool(self.REDIS_HOST)


@lru_cache
def get_settings() -> Settings:
    """获取配置单例（带缓存）"""
    settings = Settings()

    # 生产环境安全检查
    if settings.ENV == "production":
        default_keys = [
            "your-secret-key-here-change-in-production",
            "change-me-in-production",
            "secret-key",
            "",
        ]
        if not settings.SECRET_KEY or settings.SECRET_KEY in default_keys:
            raise ValueError(
                "生产环境必须设置强随机 SECRET_KEY！"
                "请在 .env 文件中设置一个随机字符串（至少 32 位）。"
            )

        # 检查 CORS 配置
        if settings.CORS_ORIGINS and "localhost" in settings.CORS_ORIGINS.lower():
            import warnings

            warnings.warn(
                "生产环境 CORS 配置中包含 localhost，建议限制为实际域名",
                RuntimeWarning,
                stacklevel=2,
            )

    return settings