""" 系统配置管理接口 """ from fastapi import APIRouter, Depends, Query, Request from sqlalchemy.orm import Session from pydantic import BaseModel from typing import Optional, Any from app.database import get_db from app.api.deps import get_current_admin from app.models.system_config import SystemConfig from app.utils.timezone import beijing_now router = APIRouter() class SystemConfigCreate(BaseModel): """创建系统配置请求""" config_key: str config_value: str config_type: str = 'string' app_name: str = 'backend-admin' # 所属应用:backend-admin、backend-client,多个用逗号拼接 description: Optional[str] = None is_editable: bool = True class SystemConfigUpdate(BaseModel): """更新系统配置请求""" config_value: str config_type: Optional[str] = None app_name: Optional[str] = None # 支持修改所属应用 description: Optional[str] = None is_editable: Optional[bool] = None def format_config_response(config: SystemConfig) -> dict: """格式化配置响应""" return { "id": config.id, "config_key": config.config_key, "config_value": config.config_value, "config_type": config.config_type, "app_name": config.app_name or '', "description": config.description, "is_system": bool(config.is_system), "is_editable": bool(config.is_editable), "created_at": config.created_at.isoformat() if config.created_at else None, "updated_at": config.updated_at.isoformat() if config.updated_at else None } @router.get("") async def get_system_config_list( page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100), keyword: Optional[str] = None, config_type: Optional[str] = None, app_name: Optional[str] = Query(None, description="所属应用过滤:backend-admin、backend-client"), db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """获取系统配置列表""" query = db.query(SystemConfig).filter(SystemConfig.is_deleted == 0) # 搜索条件 if keyword: query = query.filter( (SystemConfig.config_key.contains(keyword)) | (SystemConfig.description.contains(keyword)) ) if config_type: query = query.filter(SystemConfig.config_type == config_type) # 按所属应用过滤(支持模糊匹配,包含指定应用) if app_name: query = query.filter(SystemConfig.app_name.contains(app_name)) # 计算总数 total = query.count() # 分页 configs = query.order_by(SystemConfig.id.desc()).offset((page - 1) * page_size).limit(page_size).all() return { "code": 200, "msg": "成功", "data": [format_config_response(c) for c in configs], "total": total, "page": page, "page_size": page_size } @router.get("/{config_id}") async def get_system_config_detail( config_id: int, db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """获取系统配置详情""" config = db.query(SystemConfig).filter( SystemConfig.id == config_id, SystemConfig.is_deleted == 0 ).first() if not config: return {"code": 404, "msg": "配置不存在"} return { "code": 200, "msg": "成功", "data": format_config_response(config) } @router.post("") async def create_system_config( req: SystemConfigCreate, request: Request, db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """创建系统配置""" # 检查配置键是否已存在 existing = db.query(SystemConfig).filter( SystemConfig.config_key == req.config_key, SystemConfig.is_deleted == 0 ).first() if existing: return {"code": 400, "msg": "配置键已存在"} config = SystemConfig( config_key=req.config_key, config_value=req.config_value, config_type=req.config_type, app_name=req.app_name, description=req.description, is_system=0, # 用户创建的配置 is_editable=1 if req.is_editable else 0, created_at=beijing_now(), updated_at=beijing_now() ) db.add(config) db.flush() # 先写 his 再提交(创建时 his 在 flush 后写,保证有 id) from app.utils.history import create_system_config_history create_system_config_history(db, config, "INSERT") db.commit() db.refresh(config) return { "code": 200, "msg": "创建成功", "data": format_config_response(config) } @router.put("/{config_id}") async def update_system_config( config_id: int, req: SystemConfigUpdate, request: Request, db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """更新系统配置""" config = db.query(SystemConfig).filter( SystemConfig.id == config_id, SystemConfig.is_deleted == 0 ).first() if not config: return {"code": 404, "msg": "配置不存在"} # 检查是否可编辑 if not config.is_editable: return {"code": 403, "msg": "系统配置不可编辑"} # 先写历史表,再改/删 from app.utils.history import create_system_config_history create_system_config_history(db, config, "UPDATE") # 更新字段 config.config_value = req.config_value if req.config_type: config.config_type = req.config_type if req.app_name is not None: config.app_name = req.app_name if req.description is not None: config.description = req.description if req.is_editable is not None: config.is_editable = 1 if req.is_editable else 0 config.updated_at = beijing_now() db.commit() return { "code": 200, "msg": "更新成功", "data": format_config_response(config) } @router.delete("/{config_id}") async def delete_system_config( config_id: int, request: Request, db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """删除系统配置(软删除)""" config = db.query(SystemConfig).filter( SystemConfig.id == config_id, SystemConfig.is_deleted == 0 ).first() if not config: return {"code": 404, "msg": "配置不存在"} # 检查是否可删除 if config.is_system: return {"code": 403, "msg": "系统配置不可删除"} # 先写历史表,再改/删 from app.utils.history import create_system_config_history create_system_config_history(db, config, "DELETE") # 软删除 config.is_deleted = 1 config.updated_at = beijing_now() db.commit() return {"code": 200, "msg": "删除成功"} @router.put("/{config_id}/editable") async def toggle_config_editable( config_id: int, db: Session = Depends(get_db), current_admin = Depends(get_current_admin) ): """切换配置可编辑状态""" config = db.query(SystemConfig).filter( SystemConfig.id == config_id, SystemConfig.is_deleted == 0 ).first() if not config: return {"code": 404, "msg": "配置不存在"} # 先写历史表,再改/删 from app.utils.history import create_system_config_history create_system_config_history(db, config, "UPDATE") # 切换状态 config.is_editable = 0 if config.is_editable else 1 config.updated_at = beijing_now() db.commit() return { "code": 200, "msg": "更新成功", "data": { "id": config.id, "is_editable": bool(config.is_editable) } }