dingdanquanliucheng/backend/app/services/audit_service.py

import json

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.repositories.audit_repository import AuditRepository


class AuditService:
    def __init__(self) -> None:
        self.repository = AuditRepository()

    def list_audit_logs(self, session: Session | None = None, filters: dict | None = None) -> dict:
        if session is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
        filters = filters or {}
        try:
            logs = self.repository.list_logs(session, filters)
            page_no = filters.get("page_no", 1)
            page_size = filters.get("page_size", 20)
            start = max(page_no - 1, 0) * page_size
            page_logs = logs[start : start + page_size]
            return {
                "total": len(logs),
                "page_no": page_no,
                "page_size": page_size,
                "list": [
                    {
                        "log_id": item.id,
                        "operator_name": item.operator_name or "",
                        "operate_type": item.operate_type,
                        "biz_type": item.biz_type,
                        "biz_id": item.biz_id,
                        "result": item.result,
                        "operate_time": item.operate_time.strftime("%Y-%m-%d %H:%M:%S") if item.operate_time else "",
                        "remark": item.remark,
                    }
                    for item in page_logs
                ],
            }
        except SQLAlchemyError as exc:
            raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc

    def write_log(self, session: Session, payload: dict) -> None:
        before_value = payload.get("before_value")
        after_value = payload.get("after_value")
        self.repository.create_log(
            session,
            {
                "operator_id": payload.get("operator_id"),
                "operator_name": payload.get("operator_name") or "系统",
                "operate_type": payload["operate_type"],
                "biz_type": payload["biz_type"],
                "biz_id": payload["biz_id"],
                "before_value": self._serialize_value(before_value),
                "after_value": self._serialize_value(after_value),
                "result": payload.get("result", "success"),
                "remark": payload.get("remark"),
            },
        )

    def _serialize_value(self, value) -> str | None:
        if value is None:
            return None
        if isinstance(value, str):
            return value
        return json.dumps(value, ensure_ascii=False, default=str)


audit_service = AuditService()