"""监控模块接口（操作日志等）""" from __future__ import annotations from datetime import datetime from typing import List, Optional from fastapi import APIRouter, Depends from pydantic import BaseModel, ConfigDict, Field from sqlalchemy.orm import Session from app.core.database import get_db from app.core.menu_guard import require_admin_or_super_admin from app.core.response import error_response, success_response from app.crud.monitor.oper_log import get_oper_log, list_oper_logs from app.schemas.monitor.operlog import OperLogDetailBody, OperLogListBody from app.utils.excel_export import ( EXPORT_MAX_ROWS, ExportColumn, ExportValidationError, export_rows_to_xlsx_response, filter_export_columns, ) router = APIRouter(prefix="/web/monitor", tags=["系统监控"]) def _user_info(user: dict) -> dict: dept_id = (user.get("region_id") or user.get("dept_id") or "").strip() return { "user_id": (user.get("user_id") or user.get("id") or "").strip(), "role": (user.get("role") or "").strip(), "tenant_no": (user.get("tenant_no") or "").strip(), "dept_id": dept_id, "region_id": dept_id, } @router.post("/operlog/list") def operlog_list( body: OperLogListBody = None, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """操作日志分页列表（基于 API 请求日志适配 RuoYi OperLog 结构）""" try: page = body.page if body else 1 page_size = body.page_size if body else 10 info = _user_info(user) rows, total = list_oper_logs( db, page=page, page_size=page_size, title=body.title if body else None, business_type=body.business_type if body else None, business_id=body.business_id if body else None, business_title=body.business_title if body else None, oper_name=body.oper_name if body else None, title_in=body.title_in if body else None, oper_name_in=body.oper_name_in if body else None, status=body.status if body else None, user_info=info, tenant_no=info.get("tenant_no"), ) total_pages = (total + page_size - 1) // page_size if page_size else 0 return success_response( { "rows": rows, "list": rows, "total": total, "page": page, "pageSize": page_size, "totalPages": total_pages, } ) except Exception as e: return error_response(500, str(e)) class OperLogColumnFacetBody(BaseModel): """操作日志列头可选值（去重）""" model_config = ConfigDict(populate_by_name=True) field: str = Field(..., description="title | operName") keyword: Optional[str] = Field(default=None, description="可选值关键字") limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限") @router.post("/operlog/column-facet-values") def operlog_column_facet_values( body: OperLogColumnFacetBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """操作日志列头可选值（操作模块/操作人员去重；与列表同样的租户和数据范围）""" from sqlalchemy import func from app.models.models import SysApiRequestLog if body.field not in ("title", "operName"): return error_response(400, "不支持的列") try: info = _user_info(user) column = SysApiRequestLog.business_title if body.field == "title" else SysApiRequestLog.create_name q = ( db.query(column.label("value"), func.count(SysApiRequestLog.id).label("cnt")) .filter( SysApiRequestLog.is_del == 0, column.isnot(None), column != "", ) ) tenant_no = info.get("tenant_no") if tenant_no: q = q.filter(SysApiRequestLog.tenant_no == tenant_no) from app.crud.monitor.oper_log import _apply_oper_log_scope q = _apply_oper_log_scope(q, db, info, tenant_no or "") kw = (body.keyword or "").strip() if kw: q = q.filter(column.like(f"%{kw}%")) lim = min(max(1, int(body.limit or 500)), 500) rows = q.group_by(column).order_by(column.asc()).limit(lim).all() out = [] for row in rows: value = str(row.value).strip() if row.value is not None else "" if not value: continue out.append({"value": value, "label": value, "count": int(row.cnt or 0)}) return success_response({"list": out}) except ValueError as e: return error_response(400, str(e)) except Exception as e: return error_response(500, str(e)) _OPER_STATUS_LABEL = {0: "成功", 1: "失败"} _BUSINESS_TYPE_LABEL = { 0: "其它", 1: "新增", 2: "修改", 3: "删除", 4: "授权", 5: "导出", 6: "导入", 7: "强退", 8: "生成代码", 9: "清空数据", } OPERLOG_EXPORT_COLUMNS = [ ExportColumn(title="操作模块", field="title"), ExportColumn(title="操作类型", field="businessTypeLabel"), ExportColumn(title="操作人员", field="operName"), ExportColumn(title="操作时间", field="operTime"), ExportColumn(title="状态", field="statusLabel"), ExportColumn(title="耗时(ms)", field="costTime"), ] class OperLogExportBody(BaseModel): model_config = ConfigDict(populate_by_name=True) title: Optional[str] = None business_type: Optional[int] = Field(default=None, alias="businessType") oper_name: Optional[str] = Field(default=None, alias="operName") status: Optional[int] = None ids: Optional[List[str]] = None columns: Optional[List[str]] = None @router.post("/operlog/export") def operlog_export( body: OperLogExportBody = None, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """导出操作日志（当前筛选全量，或 ids 已选）。""" try: body = body or OperLogExportBody() info = _user_info(user) rows, _ = list_oper_logs( db, page=1, page_size=EXPORT_MAX_ROWS, title=body.title, business_type=body.business_type, oper_name=body.oper_name, status=body.status, ids=body.ids, max_page_size=EXPORT_MAX_ROWS, user_info=info, tenant_no=info.get("tenant_no"), ) for row in rows: bt = row.get("businessType") row["businessTypeLabel"] = _BUSINESS_TYPE_LABEL.get(int(bt) if bt is not None else -1, str(bt if bt is not None else "")) st = row.get("status") row["statusLabel"] = _OPER_STATUS_LABEL.get(int(st) if st is not None else -1, str(st if st is not None else "")) requested = None if body.columns: requested = [] for key in body.columns: if key == "businessType": requested.append("businessTypeLabel") elif key == "status": requested.append("statusLabel") else: requested.append(key) cols = filter_export_columns(OPERLOG_EXPORT_COLUMNS, requested) date_str = datetime.now().strftime("%Y%m%d") return export_rows_to_xlsx_response( db, tenant_no=info.get("tenant_no") or None, sheet_title="操作日志", columns=cols, rows=rows, filename_stem=f"操作日志_{date_str}", ) except ExportValidationError as e: return error_response(400, str(e)) except Exception as e: return error_response(500, str(e)) @router.post("/operlog/detail") def operlog_detail( body: OperLogDetailBody, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """操作日志详情""" try: info = _user_info(user) row = get_oper_log(db, body.id, user_info=info, tenant_no=info.get("tenant_no")) if not row: return error_response(404, "操作日志不存在") return success_response(row) except Exception as e: return error_response(500, str(e))