shanxia2.0/backend/app/api/monitor.py

"""监控模块接口（操作日志等）"""
from __future__ import annotations

from fastapi import APIRouter, Depends
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

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,
            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))


@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))