zlwuandCursor
66d5a8714b
DB 列表/仪表盘 scoped 聚合，操作日志按 user_id 过滤，工作流实例详情与审批前校验 scope。 Co-authored-by: Cursor <cursoragent@cursor.com>
224 lines
7.1 KiB
Python
224 lines
7.1 KiB
Python
"""操作日志：基于 sys_api_request_log 适配 RuoYi OperLog 视图"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
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.models.models import SysApiRequestLog, SysUser
|
||
|
||
_FMT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
# path 末段 -> sys_oper_type 字典值
|
||
_ACTION_TYPE_MAP = {
|
||
"add": 1,
|
||
"create": 1,
|
||
"insert": 1,
|
||
"update": 2,
|
||
"edit": 2,
|
||
"modify": 2,
|
||
"delete": 3,
|
||
"remove": 3,
|
||
"export": 5,
|
||
"import": 6,
|
||
"transfer": 2,
|
||
"intopool": 2,
|
||
"deal": 2,
|
||
"claim": 2,
|
||
"release": 2,
|
||
}
|
||
|
||
# path 片段 -> 模块标题
|
||
_MODULE_TITLE_RULES: List[Tuple[re.Pattern[str], str]] = [
|
||
(re.compile(r"/crm/customer", re.I), "CRM客户管理"),
|
||
(re.compile(r"/crm/contact", re.I), "CRM联系人"),
|
||
(re.compile(r"/crm/opportunity", re.I), "CRM商机管理"),
|
||
(re.compile(r"/crm/contract", re.I), "CRM合同管理"),
|
||
(re.compile(r"/crm/invoice-info", re.I), "CRM开票信息"),
|
||
(re.compile(r"/crm/", re.I), "CRM管理"),
|
||
(re.compile(r"/web/admin/", re.I), "系统管理"),
|
||
(re.compile(r"/web/auth/", re.I), "认证授权"),
|
||
]
|
||
|
||
|
||
def _title_from_path(path: str) -> str:
|
||
for pattern, title in _MODULE_TITLE_RULES:
|
||
if pattern.search(path or ""):
|
||
return title
|
||
parts = [p for p in (path or "").split("/") if p]
|
||
return parts[-1] if parts else "操作日志"
|
||
|
||
|
||
def _business_type_from_path(path: str, http_method: str) -> int:
|
||
parts = [p for p in (path or "").split("/") if p]
|
||
if parts:
|
||
last = parts[-1].lower()
|
||
if last in _ACTION_TYPE_MAP:
|
||
return _ACTION_TYPE_MAP[last]
|
||
method = (http_method or "").upper()
|
||
if method == "POST":
|
||
return 1
|
||
if method in ("PUT", "PATCH"):
|
||
return 2
|
||
if method == "DELETE":
|
||
return 3
|
||
return 0
|
||
|
||
|
||
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 ""
|
||
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 ""
|
||
|
||
return {
|
||
"id": getattr(row, "id", ""),
|
||
"title": _title_from_path(path),
|
||
"businessType": _business_type_from_path(path, getattr(row, "http_method", "")),
|
||
"method": getattr(row, "http_method", "") or "",
|
||
"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 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,
|
||
status: Optional[int] = None,
|
||
user_info: Optional[dict] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
q = db.query(SysApiRequestLog).filter(SysApiRequestLog.is_del == 0)
|
||
if user_info:
|
||
q = _apply_oper_log_scope(q, db, user_info, tenant_no or user_info.get("tenant_no"))
|
||
|
||
if oper_name:
|
||
kw = f"%{oper_name.strip()}%"
|
||
q = q.filter(
|
||
or_(
|
||
SysApiRequestLog.username.like(kw),
|
||
SysApiRequestLog.create_name.like(kw),
|
||
)
|
||
)
|
||
|
||
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,
|
||
)
|
||
)
|
||
|
||
# title / businessType 为衍生字段，先拉取再在内存过滤（数据量可控）
|
||
rows = q.order_by(desc(SysApiRequestLog.create_at)).limit(5000).all()
|
||
items = [_row_to_oper_log(r) for r in rows]
|
||
|
||
if title:
|
||
title_kw = title.strip().lower()
|
||
items = [i for i in items if title_kw in (i.get("title") or "").lower()]
|
||
|
||
if business_type is not None:
|
||
items = [i for i in items if i.get("businessType") == int(business_type)]
|
||
|
||
total = len(items)
|
||
start = (max(page, 1) - 1) * page_size
|
||
end = start + page_size
|
||
return items[start:end], 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
|
||
return _row_to_oper_log(row)
|