67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.repositories.order_approve_log_repository import OrderApproveLogRepository
|
|
|
|
|
|
class AuditService:
|
|
def __init__(self) -> None:
|
|
self.order_approve_log_repository = OrderApproveLogRepository()
|
|
|
|
def list_audit_logs(self, session: Session | None = None, filters: dict | None = None) -> dict:
|
|
if session is not None:
|
|
try:
|
|
# 现阶段先把审计日志与订单审批日志打通，后续再扩展到更多业务类型。
|
|
if filters and filters.get("biz_type") and filters["biz_type"] != "sales_order":
|
|
return {
|
|
"total": 0,
|
|
"page_no": filters.get("page_no", 1),
|
|
"page_size": filters.get("page_size", 20),
|
|
"list": [],
|
|
}
|
|
|
|
logs = self.order_approve_log_repository.list_by_filters(session, filters or {})
|
|
start = max((filters or {}).get("page_no", 1) - 1, 0) * (filters or {}).get("page_size", 20)
|
|
page_size = (filters or {}).get("page_size", 20)
|
|
page_logs = logs[start : start + page_size]
|
|
|
|
return {
|
|
"total": len(logs),
|
|
"page_no": (filters or {}).get("page_no", 1),
|
|
"page_size": page_size,
|
|
"list": [
|
|
{
|
|
"log_id": item.id,
|
|
"operator_name": "系统审批",
|
|
"operate_type": item.approve_type,
|
|
"biz_type": "sales_order",
|
|
"biz_id": item.order_id,
|
|
"result": "success",
|
|
"operate_time": item.created_at.strftime("%Y-%m-%d %H:%M:%S") if item.created_at else "",
|
|
}
|
|
for item in page_logs
|
|
],
|
|
}
|
|
except SQLAlchemyError:
|
|
pass
|
|
|
|
return {
|
|
"total": 1,
|
|
"page_no": 1,
|
|
"page_size": 20,
|
|
"list": [
|
|
{
|
|
"log_id": 90001,
|
|
"operator_name": "系统审批",
|
|
"operate_type": "order",
|
|
"biz_type": "sales_order",
|
|
"biz_id": 9001,
|
|
"result": "success",
|
|
"operate_time": "2026-05-14 10:00:00",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
audit_service = AuditService()
|