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 apply_oper_log_filters, 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, extra="allow")

    field: str = Field(..., description="title | operName")
    keyword: Optional[str] = Field(default=None, description="列表模块名称关键字")
    list_title: Optional[str] = Field(default=None, alias="listTitle", description="顶部模块名称搜索")
    list_oper_name: Optional[str] = Field(default=None, alias="listOperName", 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
    from app.services.pl.facet_context import strip_simple_facet_self_params

    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
        data = body.model_dump(by_alias=True, exclude_none=True)
        extra = getattr(body, "model_extra", None) or {}
        for key, value in extra.items():
            if key not in data and value not in (None, ""):
                data[key] = value
        params = strip_simple_facet_self_params(data, body.field)
        # 顶部搜索与列头多选分开传递，排除本列时仍保留顶部搜索。
        params["title"] = body.list_title or body.keyword or params.get("title")
        params["operName"] = body.list_oper_name or params.get("operName")
        filters = OperLogListBody.model_validate(params)
        q = (
            db.query(column.label("value"), func.count(SysApiRequestLog.id).label("cnt"))
            .filter(
                SysApiRequestLog.is_del == 0,
                column.isnot(None),
                column != "",
            )
        )
        q = apply_oper_log_filters(
            q,
            db,
            **filters.model_dump(exclude={"page", "page_size"}),
            user_info=info,
            tenant_no=info.get("tenant_no"),
        )
        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))