Spaces:
Runtime error
Runtime error
File size: 5,481 Bytes
680fa2b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | from typing import List
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import func
from backend.app.core.database import get_db
from backend.app.routers.auth import get_current_user
from backend.app.repositories.users import UserRepository
from backend.app.repositories.alerts import AlertRepository
from backend.app.repositories.bookings import BookingRepository
from backend.app.services.bookings import BookingService
from backend.app.schemas.admin import UserAdminView, AdminAlertResponse, AdminAnalytics
from backend.app.models.users import User, WorkerProfile
from backend.app.models.jobs import Job
from backend.app.models.bookings import Booking
from backend.app.models.alerts import AdminAlert
router = APIRouter(prefix="/admin", tags=["admin"])
def check_admin(current_user: User):
"""Utility to restrict access to Admins only."""
if current_user.role != "admin":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only administrators can access this endpoint."
)
@router.get("/users", response_model=List[UserAdminView])
def get_users(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
List all registered platform users. (Admins only).
"""
check_admin(current_user)
user_repo = UserRepository(db)
return user_repo.get_all_users()
@router.patch("/users/{user_id}/status", response_model=UserAdminView)
def toggle_user_status(
user_id: uuid.UUID,
is_active: bool,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
Suspend or activate a user account. (Admins only).
"""
check_admin(current_user)
user_repo = UserRepository(db)
user = user_repo.update_user_status(user_id, is_active)
if not user:
raise HTTPException(status_code=404, detail="User not found.")
return user
@router.delete("/users/{user_id}", status_code=status.HTTP_200_OK)
def delete_user(
user_id: uuid.UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
Permanently delete a user account from the system. (Admins only).
"""
check_admin(current_user)
user_repo = UserRepository(db)
success = user_repo.delete_user(user_id)
if not success:
raise HTTPException(status_code=404, detail="User not found.")
return {"message": "User deleted successfully."}
@router.get("/alerts", response_model=List[AdminAlertResponse])
def get_alerts(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
List active fraud, dispute, and IVR platform warnings. (Admins only).
"""
check_admin(current_user)
alert_repo = AlertRepository(db)
return alert_repo.get_active()
@router.post("/alerts/{alert_id}/resolve", response_model=AdminAlertResponse)
def resolve_alert(
alert_id: uuid.UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
Mark a warning alert as resolved, removing it from the admin review queue. (Admins only).
"""
check_admin(current_user)
alert_repo = AlertRepository(db)
# Check if alert is a Dispute and needs resolving
alert = alert_repo.get_by_id(alert_id)
if alert and alert.type == "Dispute":
# Resolve dispute by refunding customer as a mock action
# Extract booking code from alert text if possible
import re
match = re.search(r"B-\d+", alert.text)
if match:
code = match.group(0)
booking_repo = BookingRepository(db)
booking = booking_repo.get_by_code(code)
if booking and booking.status == "DISPUTED":
booking_service = BookingService(db)
booking_service.refund_booking(booking.id)
resolved = alert_repo.resolve(alert_id)
if not resolved:
raise HTTPException(status_code=404, detail="Alert not found.")
return resolved
@router.get("/analytics", response_model=AdminAnalytics)
def get_analytics(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""
Calculate platform performance indicators and metrics in real-time. (Admins only).
"""
check_admin(current_user)
workers_count = db.query(User).filter(User.role == "worker").count()
customers_count = db.query(User).filter(User.role == "customer").count()
jobs_count = db.query(Job).count()
bookings = db.query(Booking).all()
bookings_count = len(bookings)
# Sum booking values
gmv = sum(b.amount for b in bookings)
platform_commission = sum(round(b.amount * 0.05) for b in bookings if b.status in ("COMPLETED", "RATED"))
# Alert counts
alerts = db.query(AdminAlert).filter(AdminAlert.status == "active").all()
disputes = sum(1 for a in alerts if a.type == "Dispute")
frauds = sum(1 for a in alerts if a.type == "Fraud")
ivrs = sum(1 for a in alerts if a.type == "IVR")
return AdminAnalytics(
total_workers=workers_count,
total_customers=customers_count,
total_jobs=jobs_count,
active_bookings=bookings_count,
gmv=gmv,
platform_commission=platform_commission,
disputes_count=disputes,
fraud_alerts_count=frauds,
ivr_alerts_count=ivrs
)
|