241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
"""监控模块接口（操作日志等）"""
|
|
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))
|