xiaoxia-saas/packages/config/base.py

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

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

import os
from typing import Optional, TypeVar

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"
    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

    # ── 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_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
    doubao_timeout: int = 30
    doubao_max_retries: int = 2

    # ── MediaKit (火山引擎 AI 媒体工具) ──────────────────────────────────
    mediakit_api_key: str = ""
    mediakit_base_url: str = "https://mediakit.cn-beijing.volces.com/api/v1"
    mediakit_timeout: int = 60

    # ── 积分/会员系统 (#1895) ────────────────────────────────────────────
    # 总开关：默认 false（对所有用户零影响），P2 路由逐个接入时用
    # `if settings.points_enabled:` 包裹，防止未完善的扣点逻辑影响现有用户。
    points_enabled: bool = False

    @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)