"""Celery task: process pending SMS jobs via Semaphore.""" import logging import os from datetime import datetime, timezone import httpx from sqlalchemy import create_engine, select, update, and_ from sqlalchemy.orm import sessionmaker from app.worker import celery_app from app.config import settings logger = logging.getLogger(__name__) def _make_sync_engine(): db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub") return create_engine(db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True, pool_size=2) _engine = _make_sync_engine() _Session = sessionmaker(bind=_engine) @celery_app.task(name="sms.process_queue") def process_sms_queue(): """Process up to 20 pending SMS jobs per run via Semaphore API.""" from app.models.sms import SmsJob, SmsJobStatus from app.models.school import School db = _Session() try: jobs = db.execute( select(SmsJob).where(SmsJob.status == SmsJobStatus.pending).limit(20) ).scalars().all() for job in jobs: school = db.get(School, job.school_id) if not school or float(school.sms_credits) <= 0: job.status = SmsJobStatus.cancelled job.error_message = "Insufficient credits" db.commit() continue result = _send_semaphore(job.recipient_phone, job.message, job.sender_name) if result["success"]: job.status = SmsJobStatus.sent job.sent_at = datetime.now(timezone.utc) job.semaphore_message_id = result.get("message_id") # Deduct credit from app.models.sms import SmsCreditLedger, SmsCreditTx school.sms_credits = float(school.sms_credits) - 1.0 db.add(SmsCreditLedger( school_id=school.id, tx_type=SmsCreditTx.deduct, amount=-1.0, balance_after=float(school.sms_credits), description=f"SMS sent to {job.recipient_phone}", reference_id=job.id, )) # Low credit alert if float(school.sms_credits) <= school.sms_credit_low_threshold: send_low_credit_alert.delay(school.id) else: job.retry_count += 1 if job.retry_count >= 5: job.status = SmsJobStatus.failed job.error_message = result.get("error") db.commit() finally: db.close() def _send_semaphore(phone: str, message: str, sender: str) -> dict: try: with httpx.Client(timeout=15) as client: resp = client.post(settings.SEMAPHORE_URL, data={ "apikey": settings.SEMAPHORE_API_KEY, "number": phone, "message": message, "sendername": sender, }) if resp.status_code == 200: data = resp.json() msg_id = str(data[0].get("message_id", "")) if isinstance(data, list) and data else None return {"success": True, "message_id": msg_id} return {"success": False, "error": f"HTTP {resp.status_code}"} except Exception as e: return {"success": False, "error": str(e)} @celery_app.task(name="sms.send_low_credit_alert") def send_low_credit_alert(school_id: str): """Send low credit warning email to school billing contact.""" from app.services.email import send_email from app.models.school import School db = _Session() try: school = db.get(School, school_id) if school and school.billing_email: send_email( to=school.billing_email, subject=f"[TapTrack Hub] Low SMS Credits — {school.name}", body=f"Your SMS credit balance for {school.name} is low ({float(school.sms_credits):.0f} remaining). Please top up to continue sending SMS notifications.", ) finally: db.close()