imetting_backend/app/api/endpoints/admin.py

120 lines
4.7 KiB
Python

from fastapi import APIRouter, Depends
from app.core.auth import get_current_admin_user, get_current_user
from app.core.config import LLM_CONFIG, DEFAULT_RESET_PASSWORD, MAX_FILE_SIZE, VOICEPRINT_CONFIG, TIMELINE_PAGESIZE
from app.core.response import create_api_response
from pydantic import BaseModel
import json
from pathlib import Path
router = APIRouter()
# 配置文件路径
CONFIG_FILE = Path(__file__).parent.parent.parent.parent / "config" / "system_config.json"
class SystemConfigModel(BaseModel):
model_name: str
template_text: str
DEFAULT_RESET_PASSWORD: str
MAX_FILE_SIZE: int # 字节为单位
TIMELINE_PAGESIZE: int # 分页数量
def load_config_from_file():
"""从文件加载配置,如果文件不存在则返回默认配置"""
try:
if CONFIG_FILE.exists():
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception:
pass
# 返回默认配置
return {
'model_name': LLM_CONFIG['model_name'],
'template_text': VOICEPRINT_CONFIG['template_text'],
'DEFAULT_RESET_PASSWORD': DEFAULT_RESET_PASSWORD,
'MAX_FILE_SIZE': MAX_FILE_SIZE,
'TIMELINE_PAGESIZE': TIMELINE_PAGESIZE
}
def save_config_to_file(config_data):
"""将配置保存到文件"""
try:
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(config_data, f, ensure_ascii=False, indent=2)
return True
except Exception as e:
print(f"保存配置文件失败: {e}")
return False
@router.get("/admin/system-config")
async def get_system_config(current_user=Depends(get_current_user)):
"""
获取系统配置
普通用户也可以获取
"""
try:
config = load_config_from_file()
response_data = {
'model_name': config.get('model_name', LLM_CONFIG['model_name']),
'template_text': config.get('template_text', VOICEPRINT_CONFIG['template_text']),
'DEFAULT_RESET_PASSWORD': config.get('DEFAULT_RESET_PASSWORD', DEFAULT_RESET_PASSWORD),
'MAX_FILE_SIZE': config.get('MAX_FILE_SIZE', MAX_FILE_SIZE),
'TIMELINE_PAGESIZE': config.get('TIMELINE_PAGESIZE', TIMELINE_PAGESIZE),
}
return create_api_response(code="200", message="配置获取成功", data=response_data)
except Exception as e:
return create_api_response(code="500", message=f"获取配置失败: {str(e)}")
@router.put("/admin/system-config")
async def update_system_config(
config: SystemConfigModel,
current_user=Depends(get_current_admin_user)
):
"""
更新系统配置
只有管理员才能访问
"""
try:
config_data = {
'model_name': config.model_name,
'template_text': config.template_text,
'DEFAULT_RESET_PASSWORD': config.DEFAULT_RESET_PASSWORD,
'MAX_FILE_SIZE': config.MAX_FILE_SIZE,
'TIMELINE_PAGESIZE': config.TIMELINE_PAGESIZE
}
if not save_config_to_file(config_data):
return create_api_response(code="500", message="配置保存到文件失败")
# 更新运行时配置
LLM_CONFIG['model_name'] = config.model_name
VOICEPRINT_CONFIG['template_text'] = config.template_text
import app.core.config as config_module
config_module.DEFAULT_RESET_PASSWORD = config.DEFAULT_RESET_PASSWORD
config_module.MAX_FILE_SIZE = config.MAX_FILE_SIZE
config_module.TIMELINE_PAGESIZE = config.TIMELINE_PAGESIZE
return create_api_response(
code="200",
message="配置更新成功,重启服务后完全生效",
data=config_data
)
except Exception as e:
return create_api_response(code="500", message=f"更新配置失败: {str(e)}")
# 在应用启动时加载配置
def load_system_config():
"""在应用启动时调用,加载保存的配置"""
try:
config = load_config_from_file()
LLM_CONFIG['model_name'] = config.get('model_name', LLM_CONFIG['model_name'])
VOICEPRINT_CONFIG['template_text'] = config.get('template_text', VOICEPRINT_CONFIG['template_text'])
import app.core.config as config_module
config_module.DEFAULT_RESET_PASSWORD = config.get('DEFAULT_RESET_PASSWORD', DEFAULT_RESET_PASSWORD)
config_module.MAX_FILE_SIZE = config.get('MAX_FILE_SIZE', MAX_FILE_SIZE)
config_module.TIMELINE_PAGESIZE = config.get('TIMELINE_PAGESIZE', TIMELINE_PAGESIZE)
print(f"系统配置加载成功: model={config.get('model_name')}, pagesize={config.get('TIMELINE_PAGESIZE')}")
except Exception as e:
print(f"加载系统配置失败,使用默认配置: {e}")