A R R R Associates
113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.core.audit.models import AuditLog
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.iam.scope import UserScope
|
|
|
|
|
|
def _safe_json(value: Any) -> str:
|
|
return json.dumps(value or {}, ensure_ascii=False, default=_json_default, sort_keys=True)
|
|
|
|
|
|
def _json_default(value: Any):
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if hasattr(value, "isoformat"):
|
|
try:
|
|
return value.isoformat()
|
|
except Exception:
|
|
pass
|
|
return str(value)
|
|
|
|
|
|
def _request_meta(request: Request | None) -> tuple[str | None, str | None]:
|
|
if not request:
|
|
return None, None
|
|
ip = request.client.host if request.client else None
|
|
user_agent = request.headers.get("user-agent")
|
|
return ip, user_agent
|
|
|
|
|
|
def write_audit_log(
|
|
db: Session,
|
|
*,
|
|
action: str,
|
|
entity_type: str,
|
|
actor: User | None = None,
|
|
request: Request | None = None,
|
|
entity_id: str | int | None = None,
|
|
entity_name: str | None = None,
|
|
status: str = "success",
|
|
target_tenant_id: int | None = None,
|
|
target_branch_id: int | None = None,
|
|
details: dict[str, Any] | None = None,
|
|
actor_email: str | None = None,
|
|
) -> AuditLog:
|
|
ip_address, user_agent = _request_meta(request)
|
|
log = AuditLog(
|
|
actor_user_id=actor.id if actor else None,
|
|
actor_email=(actor.email if actor else actor_email),
|
|
actor_tenant_id=(actor.tenant_id if actor else None),
|
|
actor_branch_id=(actor.branch_id if actor else None),
|
|
action=action,
|
|
entity_type=entity_type,
|
|
entity_id=str(entity_id) if entity_id is not None else None,
|
|
entity_name=entity_name,
|
|
status=status,
|
|
target_tenant_id=target_tenant_id,
|
|
target_branch_id=target_branch_id,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
details_json=_safe_json(details),
|
|
)
|
|
db.add(log)
|
|
db.commit()
|
|
db.refresh(log)
|
|
return log
|
|
|
|
|
|
def model_snapshot(obj: Any, fields: list[str]) -> dict[str, Any]:
|
|
return {field: getattr(obj, field, None) for field in fields}
|
|
|
|
|
|
def pair_before_after(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
|
|
return {"before": before, "after": after}
|
|
|
|
|
|
def list_audit_logs(db: Session, scope: UserScope, limit: int = 200) -> list[AuditLog]:
|
|
q = select(AuditLog)
|
|
if not scope.is_system_admin:
|
|
q = q.where(AuditLog.target_tenant_id == scope.actor.tenant_id)
|
|
if scope.branch_scoped:
|
|
q = q.where(AuditLog.target_branch_id == scope.actor.branch_id)
|
|
return db.execute(q.order_by(AuditLog.created_at_utc.desc(), AuditLog.id.desc()).limit(limit)).scalars().all()
|
|
|
|
|
|
def parse_details(log: AuditLog) -> dict[str, Any]:
|
|
try:
|
|
return json.loads(log.details_json or "{}")
|
|
except Exception:
|
|
return {"raw": log.details_json}
|
|
|
|
|
|
|
|
def search_audit_logs(db: Session, scope: UserScope, q: str | None = None) -> list[AuditLog]:
|
|
rows = list_audit_logs(db, scope, limit=1000)
|
|
query = (q or "").strip().lower()
|
|
if not query:
|
|
return rows
|
|
result = []
|
|
for row in rows:
|
|
hay = " ".join([str(row.action or ""), str(row.entity_type or ""), str(row.entity_name or ""), str(row.actor_email or ""), str(row.details_json or "")]).lower()
|
|
if query in hay:
|
|
result.append(row)
|
|
return result
|