feat(phase-4): SMS gateway dashboard + stats + health + retry
Backend (backend/app/routers/sms.py):
- GET /api/sms/stats: aggregate totals by status, daily volume chart
(configurable N days), per-school top-10 breakdown, delivery rate,
live queue depth
- POST /api/sms/jobs/{id}/retry: reset failed/cancelled job to pending
- GET /api/sms/health: Semaphore API connectivity check + credit balance
- POST /api/sms/trigger-queue: manually fire sms.process_queue Celery task
Frontend (frontend/src/pages/SmsPage.vue) — rebuilt from scratch:
- Semaphore health badge with live status indicator and balance display
- 'Process Queue' button calling POST /api/sms/trigger-queue
- KPI row: Total, Sent, Failed, Pending (queue depth), Delivery Rate %;
period picker switching between 7/30/90 day views
- Volume bar chart: daily sent vs failed bars with hover tooltips,
x-axis date labels, legend; no charting library required
- School breakdown panel: top 10 schools by volume with delivery
percentage bars
- Jobs table: status + school + phone number filters; inline error
message on failed rows; Retry button for failed/cancelled jobs;
retry count display; pagination
api.ts: getSmsStats, retrySmsJob, getSmsHealth, triggerSmsQueue
PAUL: Phase 4 marked complete, STATE + ROADMAP updated
This commit is contained in:
@@ -1,10 +1,10 @@
|
||||
"""SMS gateway endpoints."""
|
||||
from typing import Optional
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timezone, date, timedelta
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, func, desc, and_
|
||||
from sqlalchemy import select, func, desc, and_, cast, Date
|
||||
|
||||
from app.auth.dependencies import require_super_admin, require_school_admin, get_current_user
|
||||
from app.database import get_db
|
||||
@@ -107,3 +107,177 @@ async def get_credit_ledger(
|
||||
],
|
||||
"total": total,
|
||||
}
|
||||
|
||||
|
||||
# ── SMS Stats ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/stats")
|
||||
async def get_sms_stats(
|
||||
school_id: Optional[str] = Query(None),
|
||||
days: int = Query(30, ge=1, le=90),
|
||||
_admin: HubUser = Depends(require_super_admin),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Aggregate SMS stats for the super admin dashboard.
|
||||
Returns: totals by status, daily volume for last N days, per-school breakdown.
|
||||
"""
|
||||
since = datetime.now(timezone.utc) - timedelta(days=days)
|
||||
base_filter = [SmsJob.created_at >= since]
|
||||
if school_id:
|
||||
base_filter.append(SmsJob.school_id == school_id)
|
||||
|
||||
# Totals by status
|
||||
status_rows = (await db.execute(
|
||||
select(SmsJob.status, func.count().label("cnt"))
|
||||
.where(and_(*base_filter))
|
||||
.group_by(SmsJob.status)
|
||||
)).all()
|
||||
totals = {r.status.value: r.cnt for r in status_rows}
|
||||
|
||||
total_all = sum(totals.values())
|
||||
total_sent = totals.get("sent", 0)
|
||||
delivery_rate = round((total_sent / total_all * 100), 1) if total_all > 0 else 0.0
|
||||
|
||||
# Daily volume for chart (last N days)
|
||||
daily_rows = (await db.execute(
|
||||
select(
|
||||
cast(SmsJob.created_at, Date).label("day"),
|
||||
SmsJob.status,
|
||||
func.count().label("cnt"),
|
||||
)
|
||||
.where(and_(*base_filter))
|
||||
.group_by(cast(SmsJob.created_at, Date), SmsJob.status)
|
||||
.order_by(cast(SmsJob.created_at, Date))
|
||||
)).all()
|
||||
|
||||
# Build day-keyed dict
|
||||
day_map: dict = {}
|
||||
for row in daily_rows:
|
||||
key = str(row.day)
|
||||
if key not in day_map:
|
||||
day_map[key] = {"date": key, "sent": 0, "failed": 0, "pending": 0, "total": 0}
|
||||
day_map[key][row.status.value] = row.cnt
|
||||
day_map[key]["total"] += row.cnt
|
||||
|
||||
# Fill missing days with zeros
|
||||
chart = []
|
||||
for i in range(days):
|
||||
d = (datetime.now(timezone.utc) - timedelta(days=days - 1 - i)).date().isoformat()
|
||||
chart.append(day_map.get(d, {"date": d, "sent": 0, "failed": 0, "pending": 0, "total": 0}))
|
||||
|
||||
# Per-school breakdown (top 10 by volume, super admin only, no school filter)
|
||||
school_breakdown: list = []
|
||||
if not school_id:
|
||||
school_rows = (await db.execute(
|
||||
select(
|
||||
SmsJob.school_id,
|
||||
func.count().label("total"),
|
||||
func.sum(
|
||||
func.cast(SmsJob.status == SmsJobStatus.sent, func.Integer())
|
||||
).label("sent"),
|
||||
)
|
||||
.where(SmsJob.created_at >= since)
|
||||
.group_by(SmsJob.school_id)
|
||||
.order_by(desc("total"))
|
||||
.limit(10)
|
||||
)).all()
|
||||
# Fetch school names
|
||||
for sr in school_rows:
|
||||
school = (await db.execute(
|
||||
select(School.name).where(School.id == sr.school_id)
|
||||
)).scalar_one_or_none()
|
||||
school_breakdown.append({
|
||||
"school_id": sr.school_id,
|
||||
"school_name": school or sr.school_id,
|
||||
"total": sr.total,
|
||||
"sent": int(sr.sent or 0),
|
||||
})
|
||||
|
||||
# Pending jobs count (for the "queue depth" indicator)
|
||||
pending_count = (await db.execute(
|
||||
select(func.count()).where(SmsJob.status == SmsJobStatus.pending)
|
||||
)).scalar_one()
|
||||
|
||||
return {
|
||||
"period_days": days,
|
||||
"total": total_all,
|
||||
"sent": total_sent,
|
||||
"failed": totals.get("failed", 0),
|
||||
"pending": totals.get("pending", 0),
|
||||
"cancelled": totals.get("cancelled", 0),
|
||||
"delivery_rate": delivery_rate,
|
||||
"queue_depth": pending_count,
|
||||
"chart": chart,
|
||||
"by_school": school_breakdown,
|
||||
}
|
||||
|
||||
|
||||
# ── Retry failed job ──────────────────────────────────────────────────────────
|
||||
|
||||
@router.post("/jobs/{job_id}/retry")
|
||||
async def retry_sms_job(
|
||||
job_id: str,
|
||||
_admin: HubUser = Depends(require_super_admin),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Reset a failed/cancelled SMS job back to pending so it gets re-processed."""
|
||||
job = (await db.execute(select(SmsJob).where(SmsJob.id == job_id))).scalar_one_or_none()
|
||||
if not job:
|
||||
raise HTTPException(404, "Job not found")
|
||||
if job.status not in (SmsJobStatus.failed, SmsJobStatus.cancelled):
|
||||
raise HTTPException(400, f"Cannot retry job with status '{job.status.value}'")
|
||||
job.status = SmsJobStatus.pending
|
||||
job.retry_count = 0
|
||||
job.error_message = None
|
||||
await db.commit()
|
||||
return {"id": job.id, "status": "pending", "message": "Job queued for retry"}
|
||||
|
||||
|
||||
# ── Manual queue trigger ──────────────────────────────────────────────────────
|
||||
|
||||
@router.post("/trigger-queue", status_code=202)
|
||||
async def trigger_sms_queue(
|
||||
_admin: HubUser = Depends(require_super_admin),
|
||||
):
|
||||
"""Manually trigger the SMS processing Celery task. Admin only."""
|
||||
try:
|
||||
from app.worker import celery_app
|
||||
celery_app.send_task("sms.process_queue")
|
||||
return {"message": "SMS queue processing triggered", "status": "queued"}
|
||||
except Exception as exc:
|
||||
raise HTTPException(500, f"Failed to trigger task: {exc}")
|
||||
|
||||
|
||||
# ── Semaphore health check ────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/health")
|
||||
async def semaphore_health(
|
||||
_admin: HubUser = Depends(require_super_admin),
|
||||
):
|
||||
"""
|
||||
Check Semaphore API connectivity and remaining message balance.
|
||||
Returns: reachable (bool), balance (int | null), error (str | null).
|
||||
"""
|
||||
import httpx
|
||||
from app.config import settings
|
||||
|
||||
if not settings.SEMAPHORE_API_KEY:
|
||||
return {"reachable": False, "balance": None, "error": "SEMAPHORE_API_KEY not configured"}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=8) as client:
|
||||
resp = await client.get(
|
||||
"https://api.semaphore.co/api/v4/account",
|
||||
params={"apikey": settings.SEMAPHORE_API_KEY},
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# Semaphore returns credits as a numeric field
|
||||
balance = data.get("credits") or data.get("balance") or data.get("credit_balance")
|
||||
return {"reachable": True, "balance": balance, "error": None, "account": data.get("name")}
|
||||
return {"reachable": False, "balance": None, "error": f"HTTP {resp.status_code}"}
|
||||
except httpx.TimeoutException:
|
||||
return {"reachable": False, "balance": None, "error": "Timeout connecting to Semaphore API"}
|
||||
except Exception as exc:
|
||||
return {"reachable": False, "balance": None, "error": str(exc)}
|
||||
|
||||
Reference in New Issue
Block a user