xiaoxia-saas/packages/config/base.py

"""统一配置基类 — 所有服务共享的基础配置。

数据库/Redis/OSS/Celery/AI服务等通用配置统一定义在此。
API 和 Worker 各自的 Settings 类继承本类，只追加服务特有字段。
单例模式和 env 文件加载逻辑也统一在这里实现。
"""

import os
from typing import Optional, TypeVar

from pydantic import AliasChoices, Field
from pydantic_settings import BaseSettings, SettingsConfigDict

T = TypeVar("T", bound=BaseSettings)


class SharedSettings(BaseSettings):
    """所有服务共享的基础配置。

    API 和 Worker 都继承本类，确保：
    1. 数据库/Redis/OSS/Celery 等核心配置默认值一致
    2. 环境变量命名统一（snake_case，pydantic-settings 自动兼容大写）
    3. env 文件加载逻辑只实现一次
    """

    # ── 环境 ──────────────────────────────────────────────────────────────
    environment: str = "development"
    debug: bool = True
    auto_create_schema: bool = False

    # ── 数据库 ────────────────────────────────────────────────────────────
    database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas"
    database_pool_size: int = 20
    database_max_overflow: int = 10  # pool_size(20) + max_overflow(10) = 最大30连接
    database_pool_timeout: int = 30
    database_pool_recycle: int = 3600

    # 测试用：使用 SQLite 内存数据库（CI 环境无需 PostgreSQL）
    use_in_memory_db: bool = False

    # ── Redis ────────────────────────────────────────────────────────────
    redis_url: str = "redis://localhost:6379/0"

    # ── Celery ───────────────────────────────────────────────────────────
    celery_broker_url: str = "redis://localhost:6379/0"
    celery_result_backend: str = "redis://localhost:6379/1"

    # ── OSS 阿里云 ──────────────────────────────────────────────────────
    oss_endpoint: str = "oss-cn-hangzhou.aliyuncs.com"
    # 内网 endpoint：ECS VPC 内访问 OSS 用（千兆带宽、免公网流量费）。
    # 为空时自动从 oss_endpoint 推导：若 oss_endpoint 是阿里云公网域名（形如
    # oss-cn-<region>.aliyuncs.com），自动加 -internal 得到内网域名；其他情况
    # （自定义域名/本地 MinIO/非阿里云）回退使用 oss_endpoint。
    # 显式填同值可以覆盖自动推导、强制所有流量都走公网。
    oss_internal_endpoint: str = ""
    oss_access_key_id: str = ""
    oss_access_key_secret: str = ""
    oss_bucket_name: str = "xiaoxia-autocut"
    oss_direct_upload_max_mb: int = 2000
    oss_direct_upload_expire_seconds: int = 900

    @property
    def effective_oss_internal_endpoint(self) -> str:
        """实际用于 SDK 内网访问的 endpoint（带 -internal 自动推导）。"""
        if self.oss_internal_endpoint:
            return self.oss_internal_endpoint
        ep = self.oss_endpoint.strip()
        scheme = ""
        host = ep
        if ep.startswith("https://"):
            scheme = "https://"
            host = ep[len("https://") :]
        elif ep.startswith("http://"):
            scheme = "http://"
            host = ep[len("http://") :]
        # 阿里云公网域名自动推导：oss-cn-<region>.aliyuncs.com → oss-cn-<region>-internal.aliyuncs.com
        if host.endswith(".aliyuncs.com") and "-internal" not in host and host.startswith("oss-cn-"):
            host = host[: -len(".aliyuncs.com")] + "-internal.aliyuncs.com"
        return f"{scheme}{host}" if scheme else host

    # ── CosyVoice (阿里云百炼语音合成) ───────────────────────────────────
    cosyvoice_api_key: str = ""
    cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1"
    cosyvoice_model: str = "cosyvoice-v3-flash"
    cosyvoice_voice: str = "longxiaochun_v3"  # 默认音色（v3 系列系统音色带 _v3 后缀）
    cosyvoice_sample_rate: int = 22050
    cosyvoice_format: str = "mp3"  # 输出格式：mp3/wav/pcm
    # 音色克隆模型名（固定为 voice-enrollment）
    cosyvoice_clone_model: str = "voice-enrollment"

    # ── 豆包大模型（火山引擎方舟） ────────────────────────────────────────
    doubao_api_key: str = ""
    doubao_model: str = "doubao-seed-1-6-250615"  # 推理模型（通用兜底）
    doubao_fast_model: str = "doubao-1-5-pro-32k-250115"  # 快速结构化输出模型（编导脚本/意图解析/审核）
    doubao_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
    doubao_timeout: int = 30
    doubao_max_retries: int = 2
    doubao_vision_model: str = "doubao-1-5-vision-pro-250915"  # 高精度视觉（备用）
    doubao_vision_lite_model: str = "doubao-1-5-vision-lite-250915"  # 快速视觉（商品识别默认，速度优先）
    doubao_vision_use_lite: bool = True  # viral-video 图片分析默认用 lite 提速
    doubao_embedding_model: str = "doubao-embedding-large-text-240915"
    doubao_video_model: str = "doubao-seedance-2-5-260628"
    doubao_video_timeout: int = 600  # 视频生成轮询总超时（秒）
    doubao_video_poll_interval: int = 10  # 轮询间隔（秒）

    # ── MediaKit (火山引擎 AI 媒体工具) ──────────────────────────────────
    mediakit_api_key: str = ""
    mediakit_base_url: str = "https://mediakit.cn-beijing.volces.com/api/v1"
    mediakit_timeout: int = 60
    mediakit_cover_enabled: bool = False  # 封面抽帧是否走MediaKit(默认false走本地ffmpeg+cv2，<2s完成)

    # ── 积分/会员系统 (#1895) ────────────────────────────────────────────
    # 积分系统总开关（产品要求 #1895：暂停积分系统但保留全部代码/表/接口）。
    # - false（默认）：所有 AI 功能（生成视频/口型/数字人/AI标题/TTS/克隆音色…）
    #   对全部登录用户免费放行，不扣积分、不做余额拦截；积分余额/流水/会员
    #   状态等查询接口保持可用，但数据不再变动。
    # - 未来恢复：只需设置环境变量 ENABLE_CREDIT_SYSTEM=true。
    # 旧开关 POINTS_ENABLED 仍保留作为兼容别名（两者任一为 true 即启用）。
    # 主开关（推荐环境变量名 ENABLE_CREDIT_SYSTEM）
    credits_enabled: bool = Field(
        default=False,
        validation_alias=AliasChoices("ENABLE_CREDIT_SYSTEM", "credits_enabled"),
    )
    # 旧开关兼容（POINTS_ENABLED）；两者任一为 true 即启用
    points_enabled_compat: bool = Field(
        default=False,
        validation_alias=AliasChoices("POINTS_ENABLED", "points_enabled_compat"),
    )

    @property
    def points_enabled(self) -> bool:
        """旧代码/测试使用的属性名，等价于积分系统总开关（兼容别名）。"""
        return bool(self.credits_enabled or self.points_enabled_compat)

    @points_enabled.setter
    def points_enabled(self, value: bool) -> None:
        # 支持旧测试/代码 ``settings.points_enabled = True`` 的写法
        self.credits_enabled = bool(value)
        self.points_enabled_compat = False

    # ── GPU MuseTalk 反向轮询 Worker ────────────────────────────────────
    # Worker 用这个长期 Token 鉴权（不是用户 JWT）。多 Worker 共用同一个 Token；
    # worker_id 用于区分具体机器。生产必须配置；development 留空会跳过校验。
    gpu_worker_token: str = ""
    # GPU 任务超时（秒）：processing 状态超过此时长（以任务心跳为准）才回退
    # pending / failed。#1970：RTX2060 6G 推理 720p 长视频需 5 分钟以上，300→900。
    # Worker 推理期间每 30s 通过 /gpu/register(task_id=...) 续心跳，
    # 只有真正超时或 Worker 明确上报 failed 才会回退。
    gpu_task_timeout_seconds: int = 900
    # 结果预签名 URL 有效期（秒）
    gpu_result_url_expires: int = 3600
    # 输入预签名 URL 有效期（秒，需留出 Worker 下载时间）
    gpu_input_url_expires: int = 3600
    # 业务侧是否启用 GPU 口型同步（开关）；关或无可用 Worker 时回退 MediaKit 云端
    use_gpu_lipsync: bool = False
    # 业务侧轮询 GPU 任务结果的间隔（秒）
    gpu_lipsync_poll_interval: float = 5.0
    # 业务侧等待 GPU 任务结果的总超时（秒）；超时后回退 MediaKit。
    # 应小于等于 gpu_task_timeout_seconds（默认900s）+ 冗余，留足 Worker 下载/上传时间。
    gpu_lipsync_wait_timeout: int = 1200
    # 判断 Worker 可用的心跳新鲜度窗口（秒）—— last_heartbeat_at 在窗口内视为在线
    gpu_worker_stale_seconds: int = 300

    # ── P4000 NVENC 硬件编码 ────────────────────────────────────────────
    # GPU 编码总开关；关闭或 endpoint 为空时始终走本机 CPU libx264
    enable_gpu_encode: bool = Field(
        default=False,
        validation_alias=AliasChoices("ENABLE_GPU_ENCODE", "enable_gpu_encode"),
    )
    # P4000 编码节点地址（Tailscale 内网），例如 http://100.105.75.67:8900
    gpu_encode_endpoint: str = Field(
        default="",
        validation_alias=AliasChoices("GPU_ENCODE_ENDPOINT", "gpu_encode_endpoint"),
    )
    # GPU 回传临时文件走公网/内网 nginx（/gpu-relay/ 已加 location）；
    # 形如 http://100.69.73.60/gpu-relay （不带尾斜杠）
    gpu_encode_relay_base_url: str = Field(
        default="",
        validation_alias=AliasChoices("GPU_ENCODE_RELAY_BASE_URL", "gpu_encode_relay_base_url"),
        description="P4000 回传结果用的外部 URL（worker 通过该 URL 提供给 P4000 PUT），如 http://100.69.73.60:8092",
    )
    # Worker→API 内网直连 URL（Docker DNS），用于 worker 自己下载/清理 relay 文件。
    # 未配置时回退到 relay_base_url（本地开发/单节点）。
    gpu_encode_relay_internal_base_url: str = Field(
        default="",
        validation_alias=AliasChoices("GPU_ENCODE_RELAY_INTERNAL_BASE_URL", "gpu_encode_relay_internal_base_url"),
    )
    # 同步调用超时（秒）：含编码+上传回传，5 分钟足够短视频
    gpu_encode_sync_timeout: int = 300
    # 异步轮询总超时（秒）：长视频走 async + 轮询
    gpu_encode_async_timeout: int = 1800
    # 轮询间隔（秒）
    gpu_encode_poll_interval: float = 3.0
    # 启动探测超时（秒）
    gpu_encode_health_timeout: float = 3.0
    # NVENC 默认编码参数（可被调用方覆盖）
    gpu_encode_vcodec: str = "h264_nvenc"
    gpu_encode_preset: str = "p4"  # NVENC preset: p1(最快)~p7(最好)，p4 为均衡
    gpu_encode_crf: int = 23
    gpu_encode_bitrate: str = ""  # 空则用 crf；非空则用 -b:v 模式
    # GPU 编码失败时是否自动降级到 CPU（默认 True）；设为 False 可在 CI/测试中暴露错误
    gpu_encode_fallback_cpu: bool = Field(
        default=True,
        validation_alias=AliasChoices("GPU_ENCODE_FALLBACK_CPU", "gpu_encode_fallback_cpu"),
    )
    # P4000 → relay 回传鉴权 token（query 参数 token=xxx）。
    # 生产环境必须设置；未设置且非 production 时自动生成随机值（写日志方便排查）。
    gpu_encode_relay_secret: str = Field(
        default="",
        validation_alias=AliasChoices("GPU_ENCODE_RELAY_SECRET", "gpu_encode_relay_secret"),
    )
    # Mezzanine 传输方式：relay=走Tailscale/Docker内网relay PUT（推荐，省公网OSS往返18-20s）；oss=走旧公网OSS路径
    gpu_encode_mezzanine_transport: str = Field(
        default="relay",
        validation_alias=AliasChoices("GPU_ENCODE_MEZZANINE_TRANSPORT", "gpu_encode_mezzanine_transport"),
    )
    # GPU 中间片在 OSS 的临时前缀（mezzanine_transport=oss 时或 relay 失败 fallback 时使用）
    gpu_encode_oss_tmp_prefix: str = Field(
        default="tmp/gpu-mezzanine/",
        validation_alias=AliasChoices("GPU_ENCODE_OSS_TMP_PREFIX", "gpu_encode_oss_tmp_prefix"),
    )
    # relay 写入目录（相对于 generated-files 根目录）
    gpu_encode_relay_dir: str = Field(
        default="gpu_relay",
        validation_alias=AliasChoices("GPU_ENCODE_RELAY_DIR", "gpu_encode_relay_dir"),
    )
    # relay 文件保留时间（秒），worker 下载完成后会主动删除，此为兜底清理 TTL
    gpu_encode_relay_ttl: int = 3600

    @property
    def effective_database_url(self) -> str:
        """返回实际使用的数据库 URL。

        当 USE_IN_MEMORY_DB=True 时返回 SQLite 内存 URL，否则返回 database_url。
        """
        if self.use_in_memory_db:
            return "sqlite:///./test.db"
        return self.database_url

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


# ── 统一单例管理 ────────────────────────────────────────────────────────
# 所有 Settings 类的单例缓存都在这里，消除每处各自实现的重复代码

_settings_cache: dict[str, BaseSettings] = {}


def _get_env_file() -> str:
    """根据 APP_ENV 决定读取哪个 env 文件。"""
    env = os.getenv("APP_ENV", "development")
    env_file = f".env.{env}" if env != "development" else ".env"
    return env_file if os.path.exists(env_file) else ".env"


def get_cached_settings(settings_class: type[T], cache_key: Optional[str] = None) -> T:
    """统一的 Settings 单例获取函数。

    所有服务都通过这个函数获取配置，消除重复的单例实现。
    按类名缓存，同一类只初始化一次。
    """
    key = cache_key or settings_class.__name__
    if key not in _settings_cache:
        env_file = _get_env_file()
        _settings_cache[key] = settings_class(_env_file=env_file)
    return _settings_cache[key]  # type: ignore[return-value]


def reload_settings_cache() -> None:
    """清空配置缓存，下次获取时重新加载。测试用。"""
    _settings_cache.clear()


def get_shared_settings() -> SharedSettings:
    """获取共享配置单例（统一入口）。"""
    return get_cached_settings(SharedSettings)