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

"""监控模块接口（操作日志等）"""
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,
            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))



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