""" 配置管理 - 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="1.7.1", description="应用版本") DEBUG: bool = Field(default=False, description="调试模式") ENV: Literal["development", "staging", "production"] = Field( default="development", description="运行环境" ) # 服务器配置 HOST: str = Field(default="0.0.0.0", description="监听地址") # nosec: B104 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=10, description="连接池临时溢出上限（建议 ≤ pool_size）") DATABASE_POOL_RECYCLE: int = Field( default=1800, description="连接回收时间（秒），防止长连接被数据库静默断开" ) DATABASE_POOL_TIMEOUT: int = Field( default=30, 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( ..., description="JWT 签名密钥（生产环境必须修改）", ) ACCESS_TOKEN_EXPIRE_MINUTES: int = Field( default=30, description="Access Token 过期时间（分钟）", ) REFRESH_TOKEN_EXPIRE_DAYS: int = Field( default=30, description="Refresh Token 过期时间（天）", ) 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,tauri://localhost,http://tauri.localhost,https://tauri.localhost", description="允许的跨域来源（逗号分隔）", ) # AI 模型配置 # 字节跳动 - 火山方舟 # 文档：https://www.volcengine.com/docs/82379/1399009 # base_url 已从 Settings 移除，改用 config/platform-config.yaml 配置 VOLCENGINE_API_KEY: str | None = Field(default=None, description="火山方舟 API Key") VOLCENGINE_MODEL: str = Field( default="doubao-seed-2-0-pro-260215", description="火山方舟默认模型（Model ID）", ) # 火山引擎音视频字幕服务 VOLCENGINE_CAPTION_APPID: str | None = Field(default=None, description="火山字幕 AppID") VOLCENGINE_CAPTION_TOKEN: str | None = Field(default=None, description="火山字幕 Token") # 火山引擎 MediaKit 服务（背景移除等多媒体处理） VOLCENGINE_MEDIAKIT_TOKEN: str | None = Field(default=None, description="火山引擎 MediaKit Token") # Vidu 密钥（base_url 已从 Settings 移除，改用 config/platform-config.yaml 配置） VIDU_API_KEY: str | None = Field(default=None, description="Vidu API Key") # 七牛云存储配置 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="图片存储域名") # 微信支付配置（APIv2） WXPAY_MCHID: str | None = Field(default=None, description="微信支付商户号") WXPAY_APPID: str | None = Field(default=None, description="微信支付 AppID") WXPAY_API_KEY: str | None = Field(default=None, description="微信支付 APIv2 密钥") WXPAY_NOTIFY_URL: str | None = Field( default=None, description="微信支付回调地址（完整 URL）" ) # B2M 短信平台配置 SMS_APP_ID: str | None = Field(default=None, description="B2M 短信平台 AppID") SMS_SECRET_KEY: str | None = Field(default=None, description="B2M 短信平台 AES 密钥") SMS_BASE_URL: str | None = Field( default=None, description="B2M 短信平台接口地址（如 http://sms.b2m.cn:8080）" ) SMS_EXTENDED_CODE: str | None = Field( default=None, description="B2M 短信平台扩展码（选填）" ) SMS_CODE_WHITELIST: str = Field( default="", description="免验证码登录白名单（逗号分隔的手机号，如 13800138000,13900139000）", ) # 文件上传限制（字节） UPLOAD_MAX_VIDEO_SIZE: int = Field( default=500 * 1024 * 1024, description="视频最大上传大小（字节）" ) UPLOAD_MAX_IMAGE_SIZE: int = Field( default=20 * 1024 * 1024, description="图片最大上传大小（字节）" ) UPLOAD_MAX_AUDIO_SIZE: int = Field( default=100 * 1024 * 1024, description="音频最大上传大小（字节）" ) # 日志配置 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) @property def sms_code_whitelist_set(self) -> set[str]: """免验证码登录白名单（去重、去空格）""" if not self.SMS_CODE_WHITELIST: return set() return { mobile.strip() for mobile in self.SMS_CODE_WHITELIST.split(",") if mobile.strip() } @lru_cache def get_settings() -> Settings: """获取配置单例（带缓存）""" settings = Settings() # 生产环境安全检查 if settings.ENV == "production": if not settings.SECRET_KEY: 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 def reload_settings() -> Settings: """ 重新加载配置（仅用于开发调试）。 ⚠️ 生产环境推荐滚动重启，不要依赖此方法。 清除 lru_cache 后重新读取 .env，返回新的配置实例。 注意：已导入模块的模块级变量（如 `settings = get_settings()`）不会自动更新， 仅后续新的 `get_settings()` 调用会返回新实例。 """ get_settings.cache_clear() return get_settings()