dingdanquanliucheng/backend/app/services/audit_service.py

import json

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

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:
        filters = filters or {}
        if session is not None:
            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:
                pass

        return {
            "total": 1,
            "page_no": 1,
            "page_size": 20,
            "list": [
                {
                    "log_id": 90001,
                    "operator_name": "系统",
                    "operate_type": "order_status_change",
                    "biz_type": "sales_order",
                    "biz_id": 9001,
                    "result": "success",
                    "operate_time": "2026-05-14 10:00:00",
                    "remark": "演示审计记录",
                }
            ],
        }

    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()