180 lines
7.1 KiB
Python
180 lines
7.1 KiB
Python
"""
|
|
Endpoints exclusivos para personal del gobierno (EMPLEADO / ADMIN).
|
|
Cierra el loop: ciudadano reporta → personal recibe y resuelve.
|
|
"""
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, desc
|
|
from ..database import get_db
|
|
from ..models.user import User
|
|
from ..models.address import Address
|
|
from ..models.report import Report, ServiceRating
|
|
from ..services import eta_service
|
|
from .deps import require_staff, require_admin
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
# ─── 1. DASHBOARD: estadísticas globales ─────────────────────────────────────
|
|
@router.get("/stats")
|
|
def get_dashboard_stats(db: Session = Depends(get_db), _=Depends(require_admin)):
|
|
total_users = db.query(func.count(User.id)).filter(User.role == "CIUDADANO").scalar()
|
|
total_addresses = db.query(func.count(Address.id)).scalar()
|
|
total_reports = db.query(func.count(Report.id)).scalar()
|
|
|
|
by_status = dict(
|
|
db.query(Report.status, func.count(Report.id)).group_by(Report.status).all()
|
|
)
|
|
by_type = dict(
|
|
db.query(Report.report_type, func.count(Report.id)).group_by(Report.report_type).all()
|
|
)
|
|
|
|
# Reportes en últimas 24h
|
|
yesterday = datetime.utcnow() - timedelta(days=1)
|
|
recent_count = db.query(func.count(Report.id)).filter(Report.created_at >= yesterday).scalar()
|
|
|
|
# Promedio de calificaciones
|
|
avg_rating = db.query(func.avg(ServiceRating.rating)).scalar() or 0
|
|
|
|
return {
|
|
"total_ciudadanos": total_users,
|
|
"total_domicilios": total_addresses,
|
|
"total_reportes": total_reports,
|
|
"reportes_24h": recent_count,
|
|
"promedio_calificacion": round(float(avg_rating), 2),
|
|
"reportes_por_estado": by_status,
|
|
"reportes_por_tipo": by_type,
|
|
"rutas_activas": len(eta_service.get_all_routes_summary()),
|
|
}
|
|
|
|
|
|
# ─── 2. REPORTES: ver todos / cambiar estado ─────────────────────────────────
|
|
@router.get("/reports")
|
|
def list_all_reports(
|
|
status: Optional[str] = Query(None, description="Filtrar por estado"),
|
|
report_type: Optional[str] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
_=Depends(require_admin),
|
|
):
|
|
q = db.query(Report).order_by(desc(Report.created_at))
|
|
if status:
|
|
q = q.filter(Report.status == status)
|
|
if report_type:
|
|
q = q.filter(Report.report_type == report_type)
|
|
|
|
results = []
|
|
for r in q.all():
|
|
addr = db.query(Address).filter(Address.id == r.address_id).first()
|
|
user = db.query(User).filter(User.id == r.user_id).first()
|
|
results.append({
|
|
"id": r.id,
|
|
"folio": r.folio,
|
|
"report_type": r.report_type,
|
|
"description": r.description,
|
|
"status": r.status,
|
|
"created_at": r.created_at.isoformat(),
|
|
"updated_at": r.updated_at.isoformat() if r.updated_at else None,
|
|
"user_name": user.full_name if user else "?",
|
|
"user_email": user.email if user else None,
|
|
"address_label": addr.label if addr else "?",
|
|
"address_street": addr.street if addr else "?",
|
|
"address_colony": addr.colony if addr else None,
|
|
"route_id": addr.route_id if addr else None,
|
|
})
|
|
return results
|
|
|
|
|
|
@router.patch("/reports/{report_id}/status")
|
|
def update_report_status(
|
|
report_id: int,
|
|
status: str = Query(..., description="PENDIENTE | EN_PROCESO | RESUELTO | CERRADO"),
|
|
db: Session = Depends(get_db),
|
|
_=Depends(require_admin),
|
|
):
|
|
if status not in ("PENDIENTE", "EN_PROCESO", "RESUELTO", "CERRADO"):
|
|
raise HTTPException(status_code=400, detail="Estado inválido")
|
|
report = db.query(Report).filter(Report.id == report_id).first()
|
|
if not report:
|
|
raise HTTPException(status_code=404, detail="Reporte no encontrado")
|
|
report.status = status
|
|
report.updated_at = datetime.utcnow()
|
|
db.commit()
|
|
db.refresh(report)
|
|
return {"id": report.id, "folio": report.folio, "status": report.status}
|
|
|
|
|
|
# ─── 3. RUTAS: estado operativo de la flotilla ───────────────────────────────
|
|
@router.get("/routes")
|
|
def list_routes_status(_=Depends(require_admin)):
|
|
"""Lista todas las rutas con su estado actual (cálculo desde el simulador)."""
|
|
return eta_service.get_all_routes_summary()
|
|
|
|
|
|
# ─── 4. USUARIOS: gestionar ciudadanos / empleados ───────────────────────────
|
|
@router.get("/users")
|
|
def list_users(
|
|
role: Optional[str] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
_=Depends(require_admin), # Solo ADMIN puede ver usuarios
|
|
):
|
|
q = db.query(User).order_by(desc(User.created_at))
|
|
if role:
|
|
q = q.filter(User.role == role)
|
|
|
|
results = []
|
|
for u in q.all():
|
|
n_addrs = db.query(func.count(Address.id)).filter(Address.user_id == u.id).scalar()
|
|
n_reports = db.query(func.count(Report.id)).filter(Report.user_id == u.id).scalar()
|
|
results.append({
|
|
"id": u.id,
|
|
"full_name": u.full_name,
|
|
"email": u.email,
|
|
"phone": u.phone,
|
|
"role": u.role,
|
|
"is_active": u.is_active,
|
|
"created_at": u.created_at.isoformat(),
|
|
"total_domicilios": n_addrs,
|
|
"total_reportes": n_reports,
|
|
})
|
|
return results
|
|
|
|
|
|
@router.patch("/users/{user_id}/role")
|
|
def update_user_role(
|
|
user_id: int,
|
|
role: str = Query(..., description="CIUDADANO | EMPLEADO | ADMIN"),
|
|
db: Session = Depends(get_db),
|
|
_=Depends(require_admin),
|
|
):
|
|
if role not in ("CIUDADANO", "EMPLEADO", "ADMIN"):
|
|
raise HTTPException(status_code=400, detail="Rol inválido")
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Usuario no encontrado")
|
|
user.role = role
|
|
db.commit()
|
|
return {"id": user.id, "role": user.role}
|
|
|
|
|
|
# ─── 5. ANUNCIOS / COMUNICACIÓN ──────────────────────────────────────────────
|
|
@router.get("/feedback")
|
|
def list_recent_feedback(db: Session = Depends(get_db), _=Depends(require_admin)):
|
|
"""Últimas calificaciones del servicio para ver feedback ciudadano."""
|
|
ratings = db.query(ServiceRating).order_by(desc(ServiceRating.created_at)).limit(50).all()
|
|
results = []
|
|
for r in ratings:
|
|
user = db.query(User).filter(User.id == r.user_id).first()
|
|
addr = db.query(Address).filter(Address.id == r.address_id).first()
|
|
results.append({
|
|
"id": r.id,
|
|
"rating": r.rating,
|
|
"comment": r.comment,
|
|
"created_at": r.created_at.isoformat(),
|
|
"user_name": user.full_name if user else "?",
|
|
"address_label": addr.label if addr else "?",
|
|
"route_id": addr.route_id if addr else None,
|
|
})
|
|
return results
|