shanxia2.0/backend/app/crud/monitor/oper_log.py

"""操作日志：基于 sys_api_request_log 适配 RuoYi OperLog 视图"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import desc, func, or_
from sqlalchemy.orm import Session

from app.core.oper_log_infer import business_type_from_path, title_from_path
from app.models.models import SysApiRequestLog, SysUser

_FMT = "%Y-%m-%d %H:%M:%S"


def _oper_status(status_code: int) -> int:
    return 0 if 200 <= int(status_code or 0) < 400 else 1


def _row_to_oper_log(row: Any) -> Dict[str, Any]:
    path = getattr(row, "path", "") or ""
    http_method = getattr(row, "http_method", "") or ""
    status_code = int(getattr(row, "status_code", 0) or 0)
    create_at = getattr(row, "create_at", None)
    oper_time = create_at.strftime(_FMT) if isinstance(create_at, datetime) else ""
    oper_name = getattr(row, "create_name", None) or getattr(row, "username", None) or ""
    request_body = getattr(row, "request_body", None) or ""
    response_body = getattr(row, "response_body", None) or ""

    stored_title = getattr(row, "business_title", None)
    stored_type = getattr(row, "business_type", None)
    title = stored_title or title_from_path(path)
    business_type = stored_type if stored_type is not None else business_type_from_path(path, http_method)
    oper_summary = getattr(row, "oper_summary", None) or ""

    return {
        "id": getattr(row, "id", ""),
        "title": title,
        "businessType": business_type,
        "businessId": getattr(row, "business_id", None) or "",
        "operSummary": oper_summary,
        "method": http_method,
        "operName": oper_name,
        "operUrl": path,
        "operIp": getattr(row, "ip_address", "") or "",
        "operLocation": "",
        "operParam": request_body,
        "jsonResult": response_body,
        "status": _oper_status(status_code),
        "costTime": int(getattr(row, "duration_ms", 0) or 0),
        "operTime": oper_time,
        "errorMsg": "" if _oper_status(status_code) == 0 else f"HTTP {status_code}",
    }


def _apply_oper_log_scope(q, db: Session, user_info: dict, tenant_no: str):
    """操作日志按 oper_user_id（user_id/create_by）应用数据权限。"""
    from app.core.config import settings
    from app.core.data_scope import _collect_scope_ids, _resolve_user_role_scopes

    if not user_info or user_info.get("role") == "super_admin":
        return q

    user_id = user_info.get("user_id")
    if not user_id:
        return q.filter(False)

    effective_tenant = tenant_no or user_info.get("tenant_no") or settings.DEFAULT_TENANT_NO
    role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant)
    if not role_scopes:
        return q.filter(False)
    if any(rs["data_scope"] == "1" for rs in role_scopes):
        return q

    dept_ids, user_ids = _collect_scope_ids(
        db, user_info, user_id, effective_tenant, role_scopes
    )
    conditions = []
    if user_ids:
        conditions.append(
            or_(
                SysApiRequestLog.user_id.in_(user_ids),
                SysApiRequestLog.create_by.in_(user_ids),
            )
        )
    if dept_ids:
        dept_user_ids = (
            db.query(SysUser.id)
            .filter(
                SysUser.dept_id.in_(dept_ids),
                SysUser.del_flag == 0,
                SysUser.tenant_no == effective_tenant,
            )
        )
        conditions.append(SysApiRequestLog.user_id.in_(dept_user_ids))

    if not conditions:
        return q.filter(False)
    return q.filter(or_(*conditions))


def _oper_log_row_in_scope(db: Session, row, user_info: dict, tenant_no: str) -> bool:
    from app.core.data_scope import row_in_data_scope

    class _OperLogScopeRow:
        create_by = getattr(row, "user_id", None) or getattr(row, "create_by", None)
        create_dept_id = None

    return row_in_data_scope(db, _OperLogScopeRow(), user_info, tenant_no)


def apply_oper_log_filters(
    q,
    db: Session,
    *,
    title: Optional[str] = None,
    business_type: Optional[int] = None,
    oper_name: Optional[str] = None,
    title_in: Optional[str] = None,
    oper_name_in: Optional[str] = None,
    status: Optional[int] = None,
    business_id: Optional[str] = None,
    business_title: Optional[str] = None,
    ids: Optional[List[str]] = None,
    user_info: Optional[dict] = None,
    tenant_no: Optional[str] = None,
):
    """列表与列头聚合共用查询条件，包含租户及操作人数据权限。"""
    q = q.filter(SysApiRequestLog.is_del == 0)

    effective_tenant = tenant_no or (user_info or {}).get("tenant_no")
    if effective_tenant:
        q = q.filter(SysApiRequestLog.tenant_no == effective_tenant)

    if user_info:
        q = _apply_oper_log_scope(q, db, user_info, effective_tenant or "")

    if business_id:
        q = q.filter(SysApiRequestLog.business_id == business_id.strip())

    if business_title:
        kw = f"%{business_title.strip()}%"
        q = q.filter(SysApiRequestLog.business_title.like(kw))

    if title:
        title_kw = f"%{title.strip()}%"
        q = q.filter(SysApiRequestLog.business_title.like(title_kw))

    title_in_values = [s.strip() for s in str(title_in or "").split(",") if s.strip()]
    if title_in_values:
        q = q.filter(SysApiRequestLog.business_title.in_(title_in_values))

    if business_type is not None:
        q = q.filter(SysApiRequestLog.business_type == int(business_type))

    if oper_name:
        kw = f"%{oper_name.strip()}%"
        q = q.filter(
            or_(
                SysApiRequestLog.username.like(kw),
                SysApiRequestLog.create_name.like(kw),
            )
        )

    oper_name_in_values = [s.strip() for s in str(oper_name_in or "").split(",") if s.strip()]
    if oper_name_in_values:
        q = q.filter(SysApiRequestLog.create_name.in_(oper_name_in_values))

    if status is not None:
        if int(status) == 0:
            q = q.filter(SysApiRequestLog.status_code >= 200, SysApiRequestLog.status_code < 400)
        else:
            q = q.filter(
                or_(
                    SysApiRequestLog.status_code < 200,
                    SysApiRequestLog.status_code >= 400,
                )
            )

    if ids:
        q = q.filter(SysApiRequestLog.id.in_(ids))

    return q


def list_oper_logs(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    title: Optional[str] = None,
    business_type: Optional[int] = None,
    oper_name: Optional[str] = None,
    title_in: Optional[str] = None,
    oper_name_in: Optional[str] = None,
    status: Optional[int] = None,
    business_id: Optional[str] = None,
    business_title: Optional[str] = None,
    ids: Optional[List[str]] = None,
    max_page_size: Optional[int] = None,
    user_info: Optional[dict] = None,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = apply_oper_log_filters(
        db.query(SysApiRequestLog),
        db,
        title=title,
        business_type=business_type,
        oper_name=oper_name,
        title_in=title_in,
        oper_name_in=oper_name_in,
        status=status,
        business_id=business_id,
        business_title=business_title,
        ids=ids,
        user_info=user_info,
        tenant_no=tenant_no,
    )
    total = int(q.with_entities(func.count(SysApiRequestLog.id)).order_by(None).scalar() or 0)
    page = max(1, page)
    cap = int(max_page_size) if max_page_size else 500
    page_size = min(max(1, page_size), cap)
    start = (page - 1) * page_size
    rows = (
        q.order_by(desc(SysApiRequestLog.create_at))
        .offset(start)
        .limit(page_size)
        .all()
    )
    return [_row_to_oper_log(r) for r in rows], total


def get_oper_log(
    db: Session, log_id: str, *, user_info: Optional[dict] = None, tenant_no: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    row = (
        db.query(SysApiRequestLog)
        .filter(SysApiRequestLog.id == log_id, SysApiRequestLog.is_del == 0)
        .first()
    )
    if not row:
        return None
    if user_info and not _oper_log_row_in_scope(db, row, user_info, tenant_no):
        return None
    if tenant_no and row.tenant_no and row.tenant_no != tenant_no:
        return None
    return _row_to_oper_log(row)