"""系统管理接口""" import json from typing import Optional from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session from sqlalchemy import desc, or_ from app.core.database import get_db from app.core.auth import get_current_user from app.core.response import success_response from app.core.utils import generate_uuid from app.models.models import SysUpgradeLog, SysComment, BizSystemConfig router = APIRouter(prefix="/web/system", tags=["系统管理"]) # ========================================= # 更新日志管理 # ========================================= @router.get("/upgrade-logs", response_model=dict) async def get_upgrade_logs( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 获取更新日志列表 """ # 构建查询 query = db.query(SysUpgradeLog).filter(SysUpgradeLog.is_del == 0) # 获取总数 total = query.count() # 分页查询 offset = (page - 1) * pageSize logs = ( query.order_by(desc(SysUpgradeLog.date)) .offset(offset) .limit(pageSize) .all() ) # 转换数据 list_data = [] for log in logs: detail_list = [] if log.detail: try: detail_list = json.loads(log.detail) except json.JSONDecodeError: detail_list = [] list_data.append({ "id": log.id, "version": log.version, "title": log.title, "date": log.date.strftime("%Y-%m-%d") if log.date else "", "detail": detail_list, "requireReLogin": bool(log.require_re_login), "remark": log.remark, "createAt": log.create_at.strftime("%Y-%m-%d %H:%M:%S") if log.create_at else "" }) return success_response({ "list": list_data, "total": total, "page": page, "pageSize": pageSize }) @router.get("/upgrade-logs/latest", response_model=dict) async def get_latest_upgrade_log( db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 获取最新一条更新日志（用于升级检测） """ log = ( db.query(SysUpgradeLog) .filter(SysUpgradeLog.is_del == 0) .order_by(desc(SysUpgradeLog.date)) .first() ) if not log: return success_response(None) detail_list = [] if log.detail: try: detail_list = json.loads(log.detail) except json.JSONDecodeError: detail_list = [] return success_response({ "version": log.version, "title": log.title, "date": log.date.strftime("%Y-%m-%d") if log.date else "", "detail": detail_list, "requireReLogin": bool(log.require_re_login), "remark": log.remark }) @router.get("/upgrade-logs/all", response_model=dict) async def get_all_upgrade_logs( db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 获取所有更新日志（用于变更日志页面展示） """ logs = ( db.query(SysUpgradeLog) .filter(SysUpgradeLog.is_del == 0) .order_by(desc(SysUpgradeLog.date)) .all() ) list_data = [] for log in logs: detail_list = [] if log.detail: try: detail_list = json.loads(log.detail) except json.JSONDecodeError: detail_list = [] list_data.append({ "version": log.version, "title": log.title, "date": log.date.strftime("%Y-%m-%d") if log.date else "", "detail": detail_list, "requireReLogin": bool(log.require_re_login), "remark": log.remark }) return success_response(list_data) @router.post("/upgrade-logs", response_model=dict) async def create_upgrade_log( version: str, title: str, date: str, detail: Optional[str] = None, require_re_login: bool = False, remark: Optional[str] = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 创建更新日志 """ # 检查版本号是否已存在 existing = db.query(SysUpgradeLog).filter( SysUpgradeLog.version == version, SysUpgradeLog.is_del == 0 ).first() if existing: return {"code": 400, "message": "版本号已存在", "data": None} from datetime import datetime log = SysUpgradeLog( id=generate_uuid(), version=version, title=title, date=datetime.strptime(date, "%Y-%m-%d"), detail=detail, require_re_login=1 if require_re_login else 0, remark=remark, create_by=current_user.get("id"), create_name=current_user.get("username", "系统") ) db.add(log) db.commit() return success_response({"id": log.id}) @router.put("/upgrade-logs/{log_id}", response_model=dict) async def update_upgrade_log( log_id: str, version: Optional[str] = None, title: Optional[str] = None, date: Optional[str] = None, detail: Optional[str] = None, require_re_login: Optional[bool] = None, remark: Optional[str] = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 更新更新日志 """ log = db.query(SysUpgradeLog).filter( SysUpgradeLog.id == log_id, SysUpgradeLog.is_del == 0 ).first() if not log: return {"code": 404, "message": "更新日志不存在", "data": None} if version: log.version = version if title: log.title = title if date: from datetime import datetime log.date = datetime.strptime(date, "%Y-%m-%d") if detail is not None: log.detail = detail if require_re_login is not None: log.require_re_login = 1 if require_re_login else 0 if remark is not None: log.remark = remark log.update_by = current_user.get("id") db.commit() return success_response({"id": log.id}) @router.delete("/upgrade-logs/{log_id}", response_model=dict) async def delete_upgrade_log( log_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 删除更新日志 """ log = db.query(SysUpgradeLog).filter( SysUpgradeLog.id == log_id, SysUpgradeLog.is_del == 0 ).first() if not log: return {"code": 404, "message": "更新日志不存在", "data": None} log.is_del = 1 log.update_by = current_user.get("id") db.commit() return success_response({"id": log.id}) # ========================================= # 评论管理 # ========================================= @router.get("/comments", response_model=dict) async def get_comments( target_type: Optional[str] = Query(None, description="目标类型"), target_id: Optional[str] = Query(None, description="目标ID"), page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 获取评论列表 """ query = db.query(SysComment).filter(SysComment.is_del == 0) if target_type: query = query.filter(SysComment.target_type == target_type) if target_id: query = query.filter(SysComment.target_id == target_id) total = query.count() offset = (page - 1) * pageSize comments = ( query.order_by(desc(SysComment.create_at)) .offset(offset) .limit(pageSize) .all() ) # 构建树形结构 comment_map = {} root_comments = [] for c in comments: comment_map[c.id] = { "id": c.id, "parentId": c.parent_id, "author": c.author, "authorId": c.author_id, "content": c.content, "targetType": c.target_type, "targetId": c.target_id, "timestamp": c.create_at.strftime("%Y-%m-%d %H:%M") if c.create_at else "", "replies": [] } for c in comments: if c.parent_id and c.parent_id in comment_map: comment_map[c.parent_id]["replies"].append(comment_map[c.id]) elif not c.parent_id: root_comments.append(comment_map[c.id]) return success_response({ "list": root_comments, "total": total, "page": page, "pageSize": pageSize }) @router.post("/comments", response_model=dict) async def create_comment( author: str, content: str, parent_id: Optional[str] = None, author_id: Optional[str] = None, target_type: Optional[str] = None, target_id: Optional[str] = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 创建评论 """ comment = SysComment( id=generate_uuid(), parent_id=parent_id, author=author, author_id=author_id or current_user.get("id"), content=content, target_type=target_type, target_id=target_id, create_by=current_user.get("id"), create_name=current_user.get("username", "匿名") ) db.add(comment) db.commit() return success_response({ "id": comment.id, "author": comment.author, "content": comment.content, "timestamp": comment.create_at.strftime("%Y-%m-%d %H:%M") if comment.create_at else "", "replies": [] }) @router.delete("/comments/{comment_id}", response_model=dict) async def delete_comment( comment_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """ 删除评论 """ comment = db.query(SysComment).filter( SysComment.id == comment_id, SysComment.is_del == 0 ).first() if not comment: return {"code": 404, "message": "评论不存在", "data": None} comment.is_del = 1 comment.update_by = current_user.get("id") db.commit() return success_response({"id": comment_id}) # ========================================= # 系统配置 # ========================================= @router.get("/china-map", response_model=dict) async def get_china_map_data( db: Session = Depends(get_db) ): """ 获取中国地图GeoJSON数据 返回保存在系统配置中的地图数据 """ config = db.query(BizSystemConfig).filter( BizSystemConfig.config_key == "china_map_geojson", BizSystemConfig.is_del == 0 ).first() if config and config.config_value: try: data = json.loads(config.config_value) return success_response(data) except json.JSONDecodeError: pass # 如果没有配置，返回空的GeoJSON结构 return success_response({ "type": "FeatureCollection", "features": [] }) @router.get("/config/{config_key}", response_model=dict) async def get_system_config( config_key: str, db: Session = Depends(get_db) ): """ 获取系统配置 """ config = db.query(BizSystemConfig).filter( BizSystemConfig.config_key == config_key, BizSystemConfig.is_del == 0 ).first() if not config: return success_response(None) # 根据配置类型解析值 value = config.config_value if config.config_type == "number": try: value = float(value) if "." in str(value) else int(value) except (ValueError, TypeError): pass elif config.config_type == "boolean": value = value.lower() in ("true", "1", "yes") if value else False elif config.config_type == "json": try: value = json.loads(value) except json.JSONDecodeError: pass return success_response({ "key": config.config_key, "value": value, "name": config.config_name, "type": config.config_type, "description": config.description }) @router.get("/version", response_model=dict) async def get_current_version( db: Session = Depends(get_db) ): """ 获取当前系统版本号 """ config = db.query(BizSystemConfig).filter( BizSystemConfig.config_key == "system_version", BizSystemConfig.is_del == 0 ).first() version = "v1.0.0" if config and config.config_value: version = config.config_value return success_response({"version": version})