Spaces:
Sleeping
Sleeping
File size: 6,727 Bytes
74de430 c1ba75f 74de430 ac99973 c1ba75f cf19fe3 c1ba75f cf19fe3 ebf074a c1ba75f 322c33c c1ba75f 74de430 cf19fe3 74de430 c1ba75f 52f8234 c1ba75f 322c33c c1ba75f cf19fe3 ac99973 cf19fe3 ac99973 cf19fe3 ac99973 cf19fe3 |
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 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 |
"""
Analytics & Statistics Endpoints
"""
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from sqlalchemy import func, and_
from typing import Dict, Any
from datetime import datetime, timedelta
import logging
from app.api.deps import get_db, get_current_user, get_current_active_user
from app.core.permissions import require_role
from app.models.user import User
from app.models.client import Client
from app.models.contractor import Contractor
from app.models.ticket import Ticket
from app.models.ticket_assignment import TicketAssignment
from app.models.project import Project
from app.models.timesheet import Timesheet
from app.models.audit_log import AuditLog
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("/platform-admin/dashboard")
@require_role(["platform_admin"])
def get_platform_admin_dashboard_stats(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
) -> Dict[str, Any]:
"""
Get comprehensive statistics for Platform Admin dashboard
Returns:
- User statistics (total, active, by role)
- Organization statistics (clients, contractors)
- Ticket statistics (total, by status, by type)
- Project statistics (total, active)
- Recent activity (latest audit logs)
"""
# User Statistics
total_users = db.query(func.count(User.id)).scalar()
active_users = db.query(func.count(User.id)).filter(User.is_active == True).scalar()
users_by_role = db.query(
User.role,
func.count(User.id)
).group_by(User.role).all()
user_stats = {
"total": total_users,
"active": active_users,
"inactive": total_users - active_users,
"by_role": {role: count for role, count in users_by_role}
}
# Organization Statistics
total_clients = db.query(func.count(Client.id)).scalar()
active_clients = db.query(func.count(Client.id)).filter(Client.is_active == True).scalar()
total_contractors = db.query(func.count(Contractor.id)).scalar()
active_contractors = db.query(func.count(Contractor.id)).filter(Contractor.is_active == True).scalar()
organization_stats = {
"clients": {
"total": total_clients,
"active": active_clients,
"inactive": total_clients - active_clients
},
"contractors": {
"total": total_contractors,
"active": active_contractors,
"inactive": total_contractors - active_contractors
}
}
# Ticket Statistics
total_tickets = db.query(func.count(Ticket.id)).scalar()
tickets_by_status = db.query(
Ticket.status,
func.count(Ticket.id)
).group_by(Ticket.status).all()
tickets_by_type = db.query(
Ticket.ticket_type,
func.count(Ticket.id)
).group_by(Ticket.ticket_type).all()
ticket_stats = {
"total": total_tickets,
"by_status": {status: count for status, count in tickets_by_status},
"by_type": {ticket_type: count for ticket_type, count in tickets_by_type}
}
# Project Statistics
total_projects = db.query(func.count(Project.id)).scalar()
active_projects = db.query(func.count(Project.id)).filter(
Project.status.in_(["planning", "active", "on_hold"])
).scalar()
project_stats = {
"total": total_projects,
"active": active_projects
}
# Assignment Statistics
total_assignments = db.query(func.count(TicketAssignment.id)).scalar()
active_assignments = db.query(func.count(TicketAssignment.id)).filter(
TicketAssignment.status.in_(["assigned", "en_route", "in_progress"])
).scalar()
assignment_stats = {
"total": total_assignments,
"active": active_assignments
}
# Recent Activity (last 10 audit logs)
recent_activity = db.query(AuditLog).order_by(
AuditLog.created_at.desc()
).limit(10).all()
activity_items = [{
"id": str(log.id),
"user_email": log.user_email,
"action": log.action,
"entity_type": log.entity_type,
"description": log.description,
"created_at": log.created_at
} for log in recent_activity]
# System Health
thirty_days_ago = (datetime.utcnow() - timedelta(days=30)).isoformat()
new_users_30d = db.query(func.count(User.id)).filter(
User.created_at >= thirty_days_ago
).scalar()
new_tickets_30d = db.query(func.count(Ticket.id)).filter(
Ticket.created_at >= thirty_days_ago
).scalar()
system_health = {
"new_users_last_30_days": new_users_30d,
"new_tickets_last_30_days": new_tickets_30d
}
return {
"users": user_stats,
"organizations": organization_stats,
"tickets": ticket_stats,
"projects": project_stats,
"assignments": assignment_stats,
"recent_activity": activity_items,
"system_health": system_health
}
@router.get("/user/overview")
def get_user_overview(
limit: int = Query(50, ge=1, le=100, description="Max items in work queue (for field agents)"),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user)
) -> Dict[str, Any]:
"""
Get user overview dashboard (cross-project aggregated stats)
Perfect for landing page / home dashboard.
Shows aggregated statistics across all projects the user has access to.
Returns different stats based on user role:
- Projects (total, active)
- Team members (for managers)
- Tickets (total, open, in_progress)
- Unread notifications
- Expenses (for managers/admins)
- Sales orders (for sales roles)
- Inventory (for managers/admins)
**For Field Agents/Sales Agents:**
- field_agent_stats: hours worked, pending expenses, inventory on hand, tickets completed this week
- work_queue: List of pending ticket assignments sorted by execution order
**Query Parameters:**
- limit: Max items in work queue (default 50, max 100)
**Authorization:** Any authenticated user
"""
try:
from app.services.dashboard_service import DashboardService
overview = DashboardService.get_user_overview(db, current_user, limit)
return overview
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get user overview: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to get user overview: {str(e)}"
)
|