mirror of
https://github.com/snailyp/gemini-balance.git
synced 2026-09-06 16:16:37 +08:00
feat: 集成数据库配置管理并添加错误日志查看器
主要变更：
1. **数据库集成**:
* 引入 MySQL 数据库支持，使用 SQLAlchemy 和 `databases` 库持久化存储应用程序设置。
* 添加了 `app/database` 目录，包含数据库连接、模型和初始化逻辑。
* 更新 `requirements.txt` 添加数据库相关依赖 (`pymysql`, `sqlalchemy`, `aiomysql`, `databases`, `python-dotenv`)。
2. **配置管理重构**:
* 重构 `ConfigService` (`app/service/config/config_service.py`)，使其从数据库加载和保存设置，并支持从 `.env` 文件同步初始配置到数据库。
* 修改 `Settings` 模型 (`app/config/config.py`) 以包含数据库连接信息，并添加了从数据库加载/同步配置的逻辑。
* 配置相关的路由 (`app/router/config_routes.py`) 更新为异步，并调用新的 `ConfigService` 方法。
* `KeyManager` (`app/service/key/key_manager.py`) 现在可以在配置更新后重置和重新初始化。
3. **错误日志查看器**:
* 新增 `/logs` 页面 (`app/templates/error_logs.html`) 用于展示应用程序错误日志。
* 添加了相应的路由 (`app/router/log_routes.py`)、静态资源 (`app/static/css/error_logs.css`, `app/static/js/error_logs.js`) 和日志记录器 (`app/log/logger.py`)。
* 在配置页面和密钥管理页面的导航栏中添加了指向日志页面的链接。
4. **异步操作**:
* 将配置服务和相关路由转换为异步 (`async def`) 以支持异步数据库操作。
5. **其他**:
* 更新了应用程序初始化逻辑 (`app/core/application.py`, `app/core/initialization.py`) 以包含数据库连接的建立和关闭。
This commit is contained in:
snaily
+230
-5
@@ -1,14 +1,34 @@
|
||||
"""
|
||||
应用程序配置模块
|
||||
"""
|
||||
from typing import List
|
||||
import datetime
|
||||
import json
|
||||
from typing import List, Any, Dict, Type
|
||||
|
||||
from pydantic import ValidationError
|
||||
from pydantic_settings import BaseSettings
|
||||
from dotenv import find_dotenv, load_dotenv
|
||||
from sqlalchemy import insert, update, select
|
||||
|
||||
from app.core.constants import API_VERSION, DEFAULT_CREATE_IMAGE_MODEL, DEFAULT_FILTER_MODELS, DEFAULT_MODEL, DEFAULT_STREAM_CHUNK_SIZE, DEFAULT_STREAM_LONG_TEXT_THRESHOLD, DEFAULT_STREAM_MAX_DELAY, DEFAULT_STREAM_MIN_DELAY, DEFAULT_STREAM_SHORT_TEXT_THRESHOLD, DEFAULT_TIMEOUT
|
||||
from app.log.logger import get_config_logger
|
||||
# 延迟导入以避免循环依赖，仅在 sync_initial_settings 中使用
|
||||
# from app.database.connection import database
|
||||
# from app.database.models import Settings as SettingsModel
|
||||
# from app.database.services import get_all_settings # get_all_settings 可能不适合启动时调用，直接查询
|
||||
|
||||
logger = get_config_logger()
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""应用程序配置"""
|
||||
# 数据库配置
|
||||
MYSQL_HOST: str
|
||||
MYSQL_PORT: int
|
||||
MYSQL_USER: str
|
||||
MYSQL_PASSWORD: str
|
||||
MYSQL_DATABASE: str
|
||||
|
||||
# API相关配置
|
||||
API_KEYS: List[str]
|
||||
ALLOWED_TOKENS: List[str]
|
||||
@@ -48,10 +68,215 @@ class Settings(BaseSettings):
|
||||
# 设置默认AUTH_TOKEN（如果未提供）
|
||||
if not self.AUTH_TOKEN and self.ALLOWED_TOKENS:
|
||||
self.AUTH_TOKEN = self.ALLOWED_TOKENS[0]
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
|
||||
|
||||
# 创建全局配置实例
|
||||
settings = Settings()
|
||||
|
||||
# 添加重新加载配置的函数
|
||||
def reload_settings():
|
||||
"""重新加载环境变量并更新配置"""
|
||||
global settings
|
||||
# 显式加载 .env 文件，覆盖现有环境变量
|
||||
# find_dotenv() 会查找 .env 文件
|
||||
load_dotenv(find_dotenv(), override=True)
|
||||
settings = Settings()
|
||||
# 可以在这里添加日志记录，确认配置已重新加载
|
||||
# print("Settings reloaded") # 用于调试
|
||||
|
||||
# --- Initial Settings Synchronization ---
|
||||
|
||||
def _parse_db_value(key: str, db_value: str, target_type: Type) -> Any:
|
||||
"""尝试将数据库字符串值解析为目标 Python 类型"""
|
||||
try:
|
||||
if target_type == List[str]:
|
||||
# 尝试解析 JSON 列表，如果失败则按逗号分割
|
||||
try:
|
||||
parsed = json.loads(db_value)
|
||||
if isinstance(parsed, list):
|
||||
return [str(item) for item in parsed]
|
||||
except json.JSONDecodeError:
|
||||
# 回退到逗号分割，去除空格
|
||||
return [item.strip() for item in db_value.split(',') if item.strip()]
|
||||
# 如果解析后不是列表或解析失败，返回空列表或进行其他处理
|
||||
logger.warning(f"Could not parse '{db_value}' as List[str] for key '{key}', falling back to comma split or empty list.")
|
||||
return [item.strip() for item in db_value.split(',') if item.strip()] # Fallback
|
||||
elif target_type == bool:
|
||||
return db_value.lower() in ('true', '1', 'yes', 'on')
|
||||
elif target_type == int:
|
||||
return int(db_value)
|
||||
elif target_type == float:
|
||||
return float(db_value)
|
||||
else: # 默认为 str 或其他 pydantic 能处理的类型
|
||||
return db_value
|
||||
except (ValueError, TypeError, json.JSONDecodeError) as e:
|
||||
logger.warning(f"Failed to parse db_value '{db_value}' for key '{key}' as type {target_type}: {e}. Using original string value.")
|
||||
return db_value # 解析失败则返回原始字符串
|
||||
|
||||
async def sync_initial_settings():
|
||||
"""
|
||||
应用启动时同步配置：
|
||||
1. 从数据库加载设置。
|
||||
2. 将数据库设置合并到内存 settings (数据库优先)。
|
||||
3. 将最终的内存 settings 同步回数据库。
|
||||
"""
|
||||
# 延迟导入以避免循环依赖和确保数据库连接已初始化
|
||||
from app.database.connection import database
|
||||
from app.database.models import Settings as SettingsModel
|
||||
|
||||
global settings
|
||||
logger.info("Starting initial settings synchronization...")
|
||||
|
||||
if not database.is_connected:
|
||||
try:
|
||||
await database.connect()
|
||||
logger.info("Database connection established for initial sync.")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to database for initial settings sync: {e}. Skipping sync.")
|
||||
return
|
||||
|
||||
try:
|
||||
# 1. 从数据库加载设置
|
||||
db_settings_raw: List[Dict[str, Any]] = []
|
||||
try:
|
||||
query = select(SettingsModel.key, SettingsModel.value)
|
||||
results = await database.fetch_all(query)
|
||||
db_settings_raw = [{"key": row["key"], "value": row["value"]} for row in results]
|
||||
logger.info(f"Fetched {len(db_settings_raw)} settings from database.")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch settings from database: {e}. Proceeding with environment/dotenv settings.")
|
||||
# 即使数据库读取失败，也要继续执行，确保基于 env/dotenv 的配置能同步到数据库
|
||||
|
||||
db_settings_map: Dict[str, str] = {s['key']: s['value'] for s in db_settings_raw}
|
||||
|
||||
# 2. 将数据库设置合并到内存 settings (数据库优先)
|
||||
updated_in_memory = False
|
||||
|
||||
for key, db_value in db_settings_map.items():
|
||||
if hasattr(settings, key):
|
||||
target_type = Settings.__annotations__.get(key)
|
||||
if target_type:
|
||||
try:
|
||||
parsed_db_value = _parse_db_value(key, db_value, target_type)
|
||||
memory_value = getattr(settings, key)
|
||||
|
||||
# 比较解析后的值和内存中的值
|
||||
# 注意：对于列表等复杂类型，直接比较可能不够健壮，但这里简化处理
|
||||
if parsed_db_value != memory_value:
|
||||
# 检查类型是否匹配，以防解析函数返回了不兼容的类型
|
||||
# 优先处理 List[str] 类型，避免直接对泛型使用 isinstance
|
||||
if target_type == List[str]:
|
||||
if isinstance(parsed_db_value, list):
|
||||
# 可以选择性地添加对列表元素的检查，但这里保持简化
|
||||
setattr(settings, key, parsed_db_value)
|
||||
logger.info(f"Updated setting '{key}' in memory from database value (List[str]).")
|
||||
updated_in_memory = True
|
||||
else:
|
||||
logger.warning(f"Parsed DB value type mismatch for key '{key}'. Expected List[str], got {type(parsed_db_value)}. Skipping update.")
|
||||
# 对于其他非泛型类型，使用常规的 isinstance 检查
|
||||
elif isinstance(parsed_db_value, target_type):
|
||||
setattr(settings, key, parsed_db_value)
|
||||
logger.info(f"Updated setting '{key}' in memory from database value.")
|
||||
updated_in_memory = True
|
||||
else:
|
||||
logger.warning(f"Parsed DB value type mismatch for key '{key}'. Expected {target_type}, got {type(parsed_db_value)}. Skipping update.")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing database setting for key '{key}': {e}")
|
||||
else:
|
||||
logger.warning(f"Database setting '{key}' not found in Settings model definition. Ignoring.")
|
||||
|
||||
|
||||
# 如果内存中有更新，重新验证 Pydantic 模型（可选但推荐）
|
||||
if updated_in_memory:
|
||||
try:
|
||||
# 重新加载以确保类型转换和验证
|
||||
settings = Settings(**settings.model_dump())
|
||||
logger.info("Settings object re-validated after merging database values.")
|
||||
except ValidationError as e:
|
||||
logger.error(f"Validation error after merging database settings: {e}. Settings might be inconsistent.")
|
||||
|
||||
|
||||
# 3. 将最终的内存 settings 同步回数据库
|
||||
final_memory_settings = settings.model_dump()
|
||||
settings_to_update: List[Dict[str, Any]] = []
|
||||
settings_to_insert: List[Dict[str, Any]] = []
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
|
||||
existing_db_keys = set(db_settings_map.keys())
|
||||
|
||||
for key, value in final_memory_settings.items():
|
||||
# 序列化值为字符串或 JSON 字符串
|
||||
if isinstance(value, list):
|
||||
db_value = json.dumps(value)
|
||||
elif isinstance(value, bool):
|
||||
db_value = str(value).lower()
|
||||
else:
|
||||
db_value = str(value)
|
||||
|
||||
data = {
|
||||
'key': key,
|
||||
'value': db_value,
|
||||
'description': f"{key} configuration setting", # 默认描述
|
||||
'updated_at': now
|
||||
}
|
||||
|
||||
if key in existing_db_keys:
|
||||
# 仅当值与数据库中的不同时才更新
|
||||
if db_settings_map[key] != db_value:
|
||||
settings_to_update.append(data)
|
||||
else:
|
||||
# 如果键不在数据库中，则插入
|
||||
data['created_at'] = now
|
||||
settings_to_insert.append(data)
|
||||
|
||||
# 在事务中执行批量插入和更新
|
||||
if settings_to_insert or settings_to_update:
|
||||
try:
|
||||
async with database.transaction():
|
||||
if settings_to_insert:
|
||||
# 获取现有描述以避免覆盖
|
||||
query_existing = select(SettingsModel.key, SettingsModel.description).where(SettingsModel.key.in_([s['key'] for s in settings_to_insert]))
|
||||
existing_desc = {row['key']: row['description'] for row in await database.fetch_all(query_existing)}
|
||||
for item in settings_to_insert:
|
||||
item['description'] = existing_desc.get(item['key'], item['description'])
|
||||
|
||||
query_insert = insert(SettingsModel).values(settings_to_insert)
|
||||
await database.execute(query=query_insert)
|
||||
logger.info(f"Synced (inserted) {len(settings_to_insert)} settings to database.")
|
||||
|
||||
if settings_to_update:
|
||||
# 获取现有描述以避免覆盖
|
||||
query_existing = select(SettingsModel.key, SettingsModel.description).where(SettingsModel.key.in_([s['key'] for s in settings_to_update]))
|
||||
existing_desc = {row['key']: row['description'] for row in await database.fetch_all(query_existing)}
|
||||
|
||||
for setting_data in settings_to_update:
|
||||
setting_data['description'] = existing_desc.get(setting_data['key'], setting_data['description'])
|
||||
query_update = (
|
||||
update(SettingsModel)
|
||||
.where(SettingsModel.key == setting_data['key'])
|
||||
.values(
|
||||
value=setting_data['value'],
|
||||
description=setting_data['description'],
|
||||
updated_at=setting_data['updated_at']
|
||||
)
|
||||
)
|
||||
await database.execute(query=query_update)
|
||||
logger.info(f"Synced (updated) {len(settings_to_update)} settings to database.")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to sync settings to database during startup: {str(e)}")
|
||||
else:
|
||||
logger.info("No setting changes detected between memory and database during initial sync.")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"An unexpected error occurred during initial settings sync: {e}")
|
||||
finally:
|
||||
if database.is_connected:
|
||||
try:
|
||||
# Don't disconnect if it's managed elsewhere (e.g., FastAPI lifespan)
|
||||
# await database.disconnect()
|
||||
# logger.info("Database connection closed after initial sync.")
|
||||
pass # Assume connection lifecycle is managed by the application lifespan
|
||||
except Exception as e:
|
||||
logger.error(f"Error disconnecting database after initial sync: {e}")
|
||||
|
||||
logger.info("Initial settings synchronization finished.")
|
||||
|
||||
Reference in New Issue
Block a user