86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.rbac import get_user_permission_names, require_permission
|
|
from app.db.database import get_db
|
|
from app.models.user import User
|
|
from app.repositories.audit_repository import AuditRepository
|
|
from app.schemas.api_response import ApiSuccess
|
|
from app.schemas.audit import ActivityFeedItem, AuditLogListResponse
|
|
from app.services.audit_service import audit_to_activity
|
|
|
|
router = APIRouter(tags=["Audit"])
|
|
|
|
|
|
@router.get("/audit-logs", response_model=ApiSuccess)
|
|
def get_audit_logs(
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=25, ge=1, le=100),
|
|
entity_type: str | None = None,
|
|
actor_user_id: int | None = None,
|
|
action: str | None = None,
|
|
date_from: datetime | None = None,
|
|
date_to: datetime | None = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_permission("audit_logs.read")),
|
|
):
|
|
items, total = AuditRepository.list(
|
|
db,
|
|
page=page,
|
|
page_size=page_size,
|
|
entity_type=entity_type,
|
|
actor_user_id=actor_user_id,
|
|
action=action,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
)
|
|
return ApiSuccess(
|
|
data=AuditLogListResponse(
|
|
items=items,
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
),
|
|
message="Audit Logs geladen",
|
|
)
|
|
|
|
|
|
def can_read_activity(action: str, permissions: set[str]) -> bool:
|
|
if action.startswith("users."):
|
|
return "users.read" in permissions
|
|
if action.startswith("customers.") or action.startswith("customer_contacts."):
|
|
return "customers.read" in permissions
|
|
if action.startswith("roles."):
|
|
return "roles.read" in permissions
|
|
if action.startswith("knowledge."):
|
|
return "knowledge.read" in permissions
|
|
if action.startswith("repairs."):
|
|
return "repairs.read" in permissions
|
|
if action.startswith("inventory."):
|
|
return "inventory.read" in permissions
|
|
if action.startswith("audit_logs."):
|
|
return "audit_logs.read" in permissions
|
|
if action.startswith("auth."):
|
|
return "audit_logs.read" in permissions
|
|
return False
|
|
|
|
|
|
@router.get("/activity-feed", response_model=ApiSuccess)
|
|
def get_activity_feed(
|
|
limit: int = Query(default=10, ge=1, le=50),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_permission("dashboard.read")),
|
|
):
|
|
permissions = get_user_permission_names(current_user)
|
|
activities: list[ActivityFeedItem] = []
|
|
|
|
for audit_log in AuditRepository.latest(db, limit=100):
|
|
if not can_read_activity(audit_log.action, permissions):
|
|
continue
|
|
activities.append(ActivityFeedItem(**audit_to_activity(audit_log)))
|
|
if len(activities) >= limit:
|
|
break
|
|
|
|
return ApiSuccess(data=activities, message="Aktivitäten geladen")
|