Mutual_Fund/repositories/audit_log.py

"""audit_log 审计日志仓储：本人操作记录筛选（台账）。"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import func, or_, select

from model.audit_log import AuditLog
from repositories.base import BaseRepository


class AuditLogRepo(BaseRepository):
    model = AuditLog

    def _conds(self, *, user_id, module, action, customer_id, keyword, start, end):
        conds = [AuditLog.user_id == user_id, AuditLog.module == module]
        if action:
            conds.append(AuditLog.action == action)
        if customer_id is not None:
            customer = str(customer_id)
            conds.append(
                or_(
                    AuditLog.target == customer,
                    AuditLog.detail.like(f'%"customer_id": {customer}%'),
                    AuditLog.detail.like(f'%"customer_id":{customer}%'),
                )
            )
        if start:
            conds.append(AuditLog.create_time >= start)
        if end:
            conds.append(AuditLog.create_time <= end)
        if keyword:
            like = f"%{keyword}%"
            conds.append(or_(AuditLog.target.like(like), AuditLog.detail.like(like)))
        return conds

    async def list_by_advisor(
        self,
        *,
        user_id: int,
        module: str = "advisor",
        action: str | None = None,
        customer_id: int | None = None,
        keyword: str | None = None,
        start: datetime | None = None,
        end: datetime | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[AuditLog]:
        stmt = (
            select(AuditLog)
            .where(*self._conds(user_id=user_id, module=module, action=action, customer_id=customer_id, keyword=keyword, start=start, end=end))
            .order_by(AuditLog.id.desc())
            .limit(limit)
            .offset(offset)
        )
        return list((await self.db.scalars(stmt)).all())

    async def count_by_advisor(
        self,
        *,
        user_id: int,
        module: str = "advisor",
        action: str | None = None,
        customer_id: int | None = None,
        keyword: str | None = None,
        start: datetime | None = None,
        end: datetime | None = None,
    ) -> int:
        stmt = (
            select(func.count())
            .select_from(AuditLog)
            .where(*self._conds(user_id=user_id, module=module, action=action, customer_id=customer_id, keyword=keyword, start=start, end=end))
        )
        return (await self.db.scalar(stmt)) or 0