368 lines
14 KiB
Python
368 lines
14 KiB
Python
"""CSV broadcasts paced under Meta's per-number throughput limit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import csv
|
|
import io
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .database import SessionLocal
|
|
from .models import UserAccount, WhatsAppBroadcastJob
|
|
from .whatsapp_graph_client import WhatsAppGraphClient, WhatsAppGraphError
|
|
from .whatsapp_messages import extract_wa_message_id, record_outbound_message
|
|
from .whatsapp_onboarding import _messaging_phone, _owned_account
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Meta Cloud API allows 80 messages per second on a Cloud API number, and 20 per
|
|
# second when that number is also signed into the WhatsApp Business app.
|
|
# Ten sequential sends, then a two second pause, stays under the lower limit.
|
|
BATCH_SIZE = 10
|
|
COOLDOWN_SECONDS = 2
|
|
RATE_LIMIT_PAUSE_SECONDS = 10
|
|
MAX_NUMBERS = 1000
|
|
|
|
COUNTRIES = (
|
|
{"code": "91", "label": "India"},
|
|
{"code": "1", "label": "United States"},
|
|
{"code": "44", "label": "United Kingdom"},
|
|
{"code": "971", "label": "United Arab Emirates"},
|
|
{"code": "65", "label": "Singapore"},
|
|
)
|
|
_COUNTRY_CODES = {item["code"] for item in COUNTRIES}
|
|
_running: set[UUID] = set()
|
|
|
|
|
|
def country_options() -> list[dict[str, str]]:
|
|
return [{"code": item["code"], "label": item["label"]} for item in COUNTRIES]
|
|
|
|
|
|
def prepare_phone_number(raw: str, country_code: str) -> tuple[str | None, str | None]:
|
|
"""Strip symbols, add the country code to a 10-digit number, then check the prefix.
|
|
|
|
Returns (number, None) when it can be sent, (None, None) for a blank cell,
|
|
or (None, reason) when the row should count as failed.
|
|
"""
|
|
digits = re.sub(r"\D", "", raw or "")
|
|
if not digits:
|
|
return None, None
|
|
full_length = 10 + len(country_code)
|
|
if len(digits) == 10:
|
|
digits = f"{country_code}{digits}"
|
|
elif len(digits) != full_length:
|
|
return None, "This number does not match the selected country code."
|
|
if not digits.startswith(country_code):
|
|
return None, f"After cleanup this number must start with {country_code}."
|
|
return digits, None
|
|
|
|
|
|
def _parse_csv(raw: bytes, country_code: str) -> tuple[list[str], int]:
|
|
try:
|
|
text = raw.decode("utf-8-sig")
|
|
except UnicodeDecodeError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="The CSV must be UTF-8.") from exc
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
if not reader.fieldnames:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="The CSV needs a header row.")
|
|
fields = [(name or "").strip().lower() for name in reader.fieldnames]
|
|
if "phone_number" not in fields:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="The CSV needs a phone_number column.")
|
|
reader.fieldnames = fields
|
|
rows = list(reader)
|
|
filled = [row for row in rows if str((row or {}).get("phone_number") or "").strip()]
|
|
if not filled:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Add at least one phone number.")
|
|
if len(filled) > MAX_NUMBERS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"A CSV can include at most {MAX_NUMBERS} numbers.",
|
|
)
|
|
ready: list[str] = []
|
|
seen: set[str] = set()
|
|
failed = 0
|
|
for row in filled:
|
|
number, reason = prepare_phone_number(str(row.get("phone_number") or ""), country_code)
|
|
if reason or not number:
|
|
failed += 1
|
|
continue
|
|
if number in seen:
|
|
failed += 1
|
|
continue
|
|
seen.add(number)
|
|
ready.append(number)
|
|
return ready, failed
|
|
|
|
|
|
def _serialize_job(job: WhatsAppBroadcastJob) -> dict[str, Any]:
|
|
return {
|
|
"id": str(job.id),
|
|
"account_id": str(job.account_id),
|
|
"filename": job.filename,
|
|
"country_code": job.country_code,
|
|
"template_name": job.template_name,
|
|
"template_language": job.template_language,
|
|
"status": job.status,
|
|
"total": job.total,
|
|
"sent_count": job.sent_count,
|
|
"failed_count": job.failed_count,
|
|
"last_error": job.last_error,
|
|
"created_at": job.created_at,
|
|
"completed_at": job.completed_at,
|
|
}
|
|
|
|
|
|
def list_broadcast_jobs(db: Session, user: UserAccount, account_id: Any) -> list[dict[str, Any]]:
|
|
account = _owned_account(db, user, account_id)
|
|
rows = db.scalars(
|
|
select(WhatsAppBroadcastJob)
|
|
.where(WhatsAppBroadcastJob.user_id == user.id, WhatsAppBroadcastJob.account_id == account.id)
|
|
.order_by(WhatsAppBroadcastJob.created_at.desc())
|
|
.limit(50)
|
|
).all()
|
|
return [_serialize_job(row) for row in rows]
|
|
|
|
|
|
def _finish_if_idle(db: Session, job: WhatsAppBroadcastJob) -> None:
|
|
recipients = job.recipients if isinstance(job.recipients, list) else []
|
|
if job.next_index < len(recipients):
|
|
return
|
|
job.status = "completed"
|
|
job.completed_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
|
|
|
|
async def create_csv_broadcast(
|
|
db: Session,
|
|
user: UserAccount,
|
|
account_id: Any,
|
|
filename: str,
|
|
raw: bytes,
|
|
*,
|
|
country_code: str,
|
|
template_name: str,
|
|
template_language: str,
|
|
body_parameters: list[str],
|
|
product_policy: str,
|
|
use_sample_header: bool,
|
|
) -> dict[str, Any]:
|
|
code = "".join(ch for ch in str(country_code or "") if ch.isdigit())
|
|
if code not in _COUNTRY_CODES:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Choose a country code from the list.")
|
|
name = template_name.strip()
|
|
if not name:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Choose a marketing template.")
|
|
account = _owned_account(db, user, account_id)
|
|
_messaging_phone(account)
|
|
if use_sample_header:
|
|
from .whatsapp_template_service import _find_sample, _header_dict, _template_for_send
|
|
|
|
template = _template_for_send(db, user, account, name, template_language.strip() or "en")
|
|
header = _header_dict(template.definition if template else None)
|
|
if template is None or _find_sample(user.id, str(header.get("sample_id") or "")) is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Add the sample file on the template page before this broadcast can attach it.",
|
|
)
|
|
ready, failed = _parse_csv(raw, code)
|
|
safe_name = Path(filename or "broadcast.csv").name[:200] or "broadcast.csv"
|
|
job = WhatsAppBroadcastJob(
|
|
user_id=user.id,
|
|
account_id=account.id,
|
|
filename=safe_name,
|
|
country_code=code,
|
|
template_name=name,
|
|
template_language=template_language.strip() or "en",
|
|
product_policy=product_policy if product_policy in {"STRICT", "CLOUD_API_FALLBACK"} else "STRICT",
|
|
use_sample_header=use_sample_header,
|
|
body_parameters=[str(item).strip() for item in body_parameters if str(item).strip()],
|
|
recipients=ready,
|
|
status="processing",
|
|
total=len(ready) + failed,
|
|
sent_count=0,
|
|
failed_count=failed,
|
|
next_index=0,
|
|
)
|
|
db.add(job)
|
|
db.commit()
|
|
db.refresh(job)
|
|
_finish_if_idle(db, job)
|
|
db.refresh(job)
|
|
if job.status == "processing":
|
|
schedule_broadcast_job(job.id)
|
|
return _serialize_job(job)
|
|
|
|
|
|
def schedule_broadcast_job(job_id: UUID) -> None:
|
|
if job_id in _running:
|
|
return
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
return
|
|
_running.add(job_id)
|
|
loop.create_task(run_broadcast_job(job_id))
|
|
|
|
|
|
def resume_broadcast_jobs() -> None:
|
|
with SessionLocal() as db:
|
|
ids = db.scalars(
|
|
select(WhatsAppBroadcastJob.id).where(WhatsAppBroadcastJob.status == "processing")
|
|
).all()
|
|
for job_id in ids:
|
|
schedule_broadcast_job(job_id)
|
|
|
|
|
|
def _rate_limited(detail: str) -> bool:
|
|
text = detail.lower()
|
|
return "130429" in text or "rate limit" in text or "throughput" in text
|
|
|
|
|
|
async def _send_one(client: WhatsAppGraphClient, number: str, name: str, language: str, components: list[dict[str, Any]] | None, policy: str) -> tuple[str, dict[str, Any] | None, str | None]:
|
|
try:
|
|
result = await client.send_marketing_template(
|
|
number,
|
|
name,
|
|
language_code=language,
|
|
components=components,
|
|
product_policy=policy,
|
|
)
|
|
return "sent", result if isinstance(result, dict) else None, None
|
|
except WhatsAppGraphError as exc:
|
|
if not _rate_limited(exc.detail):
|
|
return "failed", None, exc.detail
|
|
await asyncio.sleep(RATE_LIMIT_PAUSE_SECONDS)
|
|
try:
|
|
result = await client.send_marketing_template(
|
|
number,
|
|
name,
|
|
language_code=language,
|
|
components=components,
|
|
product_policy=policy,
|
|
)
|
|
return "sent", result if isinstance(result, dict) else None, None
|
|
except WhatsAppGraphError as retry_exc:
|
|
return "failed", None, retry_exc.detail
|
|
|
|
|
|
async def run_broadcast_job(job_id: UUID) -> None:
|
|
client: WhatsAppGraphClient | None = None
|
|
components: list[dict[str, Any]] | None = None
|
|
from_number = ""
|
|
phone_number_id = ""
|
|
try:
|
|
while True:
|
|
with SessionLocal() as db:
|
|
job = db.get(WhatsAppBroadcastJob, job_id)
|
|
if job is None or job.status != "processing":
|
|
return
|
|
recipients = list(job.recipients or [])
|
|
if job.next_index >= len(recipients):
|
|
_finish_if_idle(db, job)
|
|
return
|
|
batch_start = job.next_index
|
|
batch = recipients[batch_start:batch_start + BATCH_SIZE]
|
|
user = db.get(UserAccount, job.user_id)
|
|
if user is None:
|
|
job.status = "failed"
|
|
job.last_error = "This broadcast has no workspace."
|
|
job.completed_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return
|
|
if client is None:
|
|
account = _owned_account(db, user, job.account_id)
|
|
phone, token = _messaging_phone(account)
|
|
phone_number_id = phone.phone_number_id
|
|
from_number = phone.display_phone or phone.phone_number_id
|
|
pieces: list[dict[str, Any]] = []
|
|
if job.use_sample_header:
|
|
from .whatsapp_template_service import header_parameter_for_send
|
|
|
|
pieces.append(
|
|
await header_parameter_for_send(
|
|
db,
|
|
user,
|
|
account,
|
|
token,
|
|
phone.phone_number_id,
|
|
job.template_name,
|
|
job.template_language,
|
|
)
|
|
)
|
|
parameters = job.body_parameters if isinstance(job.body_parameters, list) else []
|
|
if parameters:
|
|
pieces.append(
|
|
{
|
|
"type": "body",
|
|
"parameters": [{"type": "text", "text": str(item)} for item in parameters],
|
|
}
|
|
)
|
|
components = pieces or None
|
|
client = WhatsAppGraphClient(
|
|
access_token=token,
|
|
phone_number_id=phone.phone_number_id,
|
|
business_account_id=account.waba_id,
|
|
)
|
|
template_name = job.template_name
|
|
template_language = job.template_language
|
|
policy = job.product_policy
|
|
user_id = job.user_id
|
|
if client is None:
|
|
return
|
|
for offset, number in enumerate(batch):
|
|
outcome, result, error = await _send_one(client, number, template_name, template_language, components, policy)
|
|
with SessionLocal() as db:
|
|
job = db.get(WhatsAppBroadcastJob, job_id)
|
|
if job is None or job.status != "processing":
|
|
return
|
|
expected = batch_start + offset
|
|
if job.next_index != expected:
|
|
return
|
|
job.next_index = expected + 1
|
|
if outcome == "sent":
|
|
job.sent_count += 1
|
|
record_outbound_message(
|
|
db,
|
|
to_number=number,
|
|
body=f"Broadcast: {template_name}",
|
|
message_type="template",
|
|
wa_message_id=extract_wa_message_id(result),
|
|
status="sent",
|
|
payload=result,
|
|
from_number=from_number,
|
|
user_id=user_id,
|
|
phone_number_id=phone_number_id,
|
|
)
|
|
else:
|
|
job.failed_count += 1
|
|
job.last_error = (error or "Meta did not accept this number.")[:500]
|
|
db.commit()
|
|
if len(batch) == BATCH_SIZE:
|
|
await asyncio.sleep(COOLDOWN_SECONDS)
|
|
except HTTPException as exc:
|
|
_mark_failed(job_id, str(exc.detail))
|
|
except Exception:
|
|
logger.exception("CSV broadcast %s stopped", job_id)
|
|
_mark_failed(job_id, "The broadcast stopped before it finished.")
|
|
finally:
|
|
_running.discard(job_id)
|
|
|
|
|
|
def _mark_failed(job_id: UUID, message: str) -> None:
|
|
with SessionLocal() as db:
|
|
job = db.get(WhatsAppBroadcastJob, job_id)
|
|
if job is None or job.status != "processing":
|
|
return
|
|
job.status = "failed"
|
|
job.last_error = message[:500]
|
|
job.completed_at = datetime.now(timezone.utc)
|
|
db.commit()
|