""" 配置服务 优先从数据库 system_configs 表读取配置,数据库无值时回退到 .env """ import logging from typing import Optional from sqlalchemy.orm import Session from ..database import SessionLocal from ..models.system_config import SystemConfig from ..config import settings as env_settings logger = logging.getLogger(__name__) def get_config_value(key: str, default: Optional[str] = None) -> Optional[str]: """ 获取配置值(数据库优先,.env 兜底) Args: key: 配置键名(如 SILICONFLOW_API_KEY) default: 默认值 Returns: 配置值字符串 """ # 1. 尝试从数据库读取 try: db = SessionLocal() try: config = db.query(SystemConfig).filter(SystemConfig.config_key == key).first() if config and config.config_value: return config.config_value finally: db.close() except Exception as e: logger.warning(f"从数据库读取配置 {key} 失败: {e}") # 2. 回退到 .env / Settings env_value = getattr(env_settings, key, None) if env_value is not None and env_value != "": return str(env_value) return default def get_ai_config() -> dict: """ 获取所有 AI 相关配置 返回字典,方便 ai_generator 使用 """ return { "SILICONFLOW_API_KEY": get_config_value("SILICONFLOW_API_KEY", ""), "SILICONFLOW_BASE_URL": get_config_value("SILICONFLOW_BASE_URL", "https://api.siliconflow.cn/v1"), "VOLCENGINE_API_KEY": get_config_value("VOLCENGINE_API_KEY", ""), "VOLCENGINE_BASE_URL": get_config_value("VOLCENGINE_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"), "AI_IMAGE_MODEL": get_config_value("AI_IMAGE_MODEL", "flux-dev"), "AI_IMAGE_SIZE": int(get_config_value("AI_IMAGE_SIZE", "1024")), }