"""统一配置基类 — 所有服务共享的基础配置。 数据库/Redis/OSS/Celery/AI服务等通用配置统一定义在此。 API 和 Worker 各自的 Settings 类继承本类,只追加服务特有字段。 单例模式和 env 文件加载逻辑也统一在这里实现。 """ import os from typing import Optional, Type, 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 @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)