"""SMS gateway endpoints.""" from typing import Optional 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_, cast, Date from app.auth.dependencies import require_super_admin, require_school_admin, get_current_user from app.database import get_db from app.models.user import HubUser, UserRole from app.models.sms import SmsJob, SmsJobStatus, SmsCreditLedger from app.models.school import School router = APIRouter(prefix="/api/sms", tags=["sms"]) class SubmitSmsJob(BaseModel): """Called by on-prem TapTrack to submit SMS jobs to Hub.""" license_key: str jobs: list[dict] # [{ recipient_phone, message, trigger }] class ManualSmsRequest(BaseModel): school_id: str recipient_phone: str message: str @router.post("/submit", status_code=202) async def submit_sms_jobs( body: SubmitSmsJob, db: AsyncSession = Depends(get_db), ): """On-prem posts SMS jobs for Hub to process via Semaphore.""" from app.models.license import License lic = (await db.execute(select(License).where(License.key == body.license_key))).scalar_one_or_none() if not lic: raise HTTPException(403, "Invalid license key") school = (await db.execute(select(School).where(School.id == lic.school_id))).scalar_one_or_none() if not school or float(school.sms_credits) <= 0: raise HTTPException(402, "Insufficient SMS credits") created_ids = [] for job_data in body.jobs: job = SmsJob( school_id=school.id, recipient_phone=job_data.get("recipient_phone", ""), message=job_data.get("message", ""), sender_name=school.sms_sender_name, trigger=job_data.get("trigger"), ) db.add(job) created_ids.append(job.id) await db.commit() return {"queued": len(created_ids), "job_ids": created_ids} @router.get("/jobs") async def list_sms_jobs( school_id: Optional[str] = Query(None), status: Optional[SmsJobStatus] = Query(None), page: int = Query(1, ge=1), per_page: int = Query(50, ge=1, le=200), current_user: HubUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): stmt = select(SmsJob).order_by(desc(SmsJob.created_at)) if current_user.role != UserRole.super_admin: stmt = stmt.where(SmsJob.school_id == current_user.school_id) elif school_id: stmt = stmt.where(SmsJob.school_id == school_id) if status: stmt = stmt.where(SmsJob.status == status) total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one() jobs = (await db.execute(stmt.offset((page - 1) * per_page).limit(per_page))).scalars().all() return { "items": [ { "id": j.id, "school_id": j.school_id, "recipient_phone": j.recipient_phone, "message": j.message[:60] + "..." if len(j.message) > 60 else j.message, "sender_name": j.sender_name, "status": j.status.value, "trigger": j.trigger, "created_at": j.created_at.isoformat(), "sent_at": j.sent_at.isoformat() if j.sent_at else None, "retry_count": j.retry_count, "error_message": j.error_message, } for j in jobs ], "total": total, "page": page, "per_page": per_page, } @router.get("/credits/{school_id}") async def get_credit_ledger( school_id: str, page: int = Query(1, ge=1), per_page: int = Query(50), current_user: HubUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if current_user.role != UserRole.super_admin and current_user.school_id != school_id: raise HTTPException(403) stmt = select(SmsCreditLedger).where(SmsCreditLedger.school_id == school_id).order_by(desc(SmsCreditLedger.created_at)) total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one() rows = (await db.execute(stmt.offset((page - 1) * per_page).limit(per_page))).scalars().all() return { "items": [ {"id": r.id, "tx_type": r.tx_type.value, "amount": float(r.amount), "balance_after": float(r.balance_after), "description": r.description, "created_at": r.created_at.isoformat()} for r in rows ], "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)}