694 lines
26 KiB
Python
694 lines
26 KiB
Python
"""Customer WhatsApp Embedded Signup provisioning.
|
|
|
|
Tokens and the platform verify token never leave this module's responses.
|
|
Each WABA and phone number is stored against one Urmi user and cannot be
|
|
addressed by another tenant.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from .config import settings
|
|
from .models import (
|
|
UserAccount,
|
|
WhatsAppBusinessAccount,
|
|
WhatsAppConnectionAudit,
|
|
WhatsAppMetaConnection,
|
|
WhatsAppPhoneAsset,
|
|
WhatsAppWebhookEvent,
|
|
)
|
|
from .whatsapp_config import resolved_whatsapp_app_secret
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RECOMMENDED_WEBHOOK_FIELDS = [
|
|
"messages",
|
|
"message_template_status_update",
|
|
"message_template_quality_update",
|
|
"account_update",
|
|
"phone_number_name_update",
|
|
"phone_number_quality_update",
|
|
"business_capability_update",
|
|
]
|
|
|
|
|
|
def signup_public_config(db: Session) -> dict[str, Any]:
|
|
app_id = settings.meta_app_id.strip()
|
|
config_id = settings.meta_embedded_signup_config_id.strip()
|
|
secret = resolved_whatsapp_app_secret(db)
|
|
return {
|
|
"app_id": app_id or None,
|
|
"config_id": config_id or None,
|
|
"api_version": settings.whatsapp_api_version,
|
|
"sdk_version": "v26.0",
|
|
"ready": bool(app_id and config_id and secret),
|
|
}
|
|
|
|
|
|
def _audit(
|
|
db: Session,
|
|
user: UserAccount,
|
|
event: str,
|
|
detail: str,
|
|
*,
|
|
connection_id: Any = None,
|
|
waba_id: str | None = None,
|
|
) -> None:
|
|
db.add(
|
|
WhatsAppConnectionAudit(
|
|
user_id=user.id,
|
|
connection_id=connection_id,
|
|
waba_id=waba_id,
|
|
event=event,
|
|
detail=detail[:500],
|
|
)
|
|
)
|
|
|
|
|
|
def mask_identifier(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
tail = value[-4:]
|
|
return f"••••••••{tail}"
|
|
|
|
|
|
def _quality_label(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
normalized = value.strip().upper()
|
|
return {
|
|
"GREEN": "High",
|
|
"HIGH": "High",
|
|
"YELLOW": "Medium",
|
|
"MEDIUM": "Medium",
|
|
"RED": "Low",
|
|
"LOW": "Low",
|
|
"UNKNOWN": "Unknown",
|
|
}.get(normalized, value.title())
|
|
|
|
|
|
def _graph_error_message(data: Any, fallback: str) -> str:
|
|
if isinstance(data, dict):
|
|
error = data.get("error")
|
|
if isinstance(error, dict) and error.get("message"):
|
|
return str(error["message"])[:500]
|
|
return fallback
|
|
|
|
|
|
async def _graph(
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
token: str | None = None,
|
|
params: dict[str, Any] | None = None,
|
|
json_body: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
url = f"https://graph.facebook.com/{settings.whatsapp_api_version}/{path.lstrip('/')}"
|
|
headers = {"Authorization": f"Bearer {token}"} if token else {}
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
response = await client.request(method, url, params=params, json=json_body, headers=headers)
|
|
try:
|
|
data = response.json()
|
|
except ValueError:
|
|
data = {}
|
|
if response.status_code >= 400:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail=_graph_error_message(data, "Meta could not complete WhatsApp setup"),
|
|
)
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
async def _exchange_code(db: Session, code: str) -> str:
|
|
app_id = settings.meta_app_id.strip()
|
|
secret = resolved_whatsapp_app_secret(db)
|
|
if not app_id or not secret:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="WhatsApp Embedded Signup is not configured",
|
|
)
|
|
data = await _graph(
|
|
"GET",
|
|
"oauth/access_token",
|
|
params={"client_id": app_id, "client_secret": secret, "code": code},
|
|
)
|
|
token = str(data.get("access_token") or "")
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="Meta did not return an authorization token",
|
|
)
|
|
return token
|
|
|
|
|
|
async def _debug_token(db: Session, token: str) -> dict[str, Any]:
|
|
app_id = settings.meta_app_id.strip()
|
|
secret = resolved_whatsapp_app_secret(db)
|
|
data = await _graph(
|
|
"GET",
|
|
"debug_token",
|
|
params={"input_token": token, "access_token": f"{app_id}|{secret}"},
|
|
)
|
|
payload = data.get("data")
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
|
|
def _waba_ids_from_debug(debug: dict[str, Any]) -> list[str]:
|
|
found: list[str] = []
|
|
scopes = debug.get("granular_scopes")
|
|
if not isinstance(scopes, list):
|
|
return found
|
|
preferred = {"whatsapp_business_management", "whatsapp_business_messaging"}
|
|
for item in scopes:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if item.get("scope") not in preferred:
|
|
continue
|
|
for target in item.get("target_ids") or []:
|
|
value = str(target)
|
|
if value and value not in found:
|
|
found.append(value)
|
|
return found
|
|
|
|
|
|
def _as_uuid(account_id: Any) -> uuid.UUID:
|
|
try:
|
|
return account_id if isinstance(account_id, uuid.UUID) else uuid.UUID(str(account_id))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="WhatsApp account not found") from exc
|
|
|
|
|
|
def _owned_account(db: Session, user: UserAccount, account_id: Any) -> WhatsAppBusinessAccount:
|
|
account = db.scalar(
|
|
select(WhatsAppBusinessAccount)
|
|
.options(selectinload(WhatsAppBusinessAccount.phones), selectinload(WhatsAppBusinessAccount.connection))
|
|
.where(WhatsAppBusinessAccount.id == _as_uuid(account_id), WhatsAppBusinessAccount.user_id == user.id)
|
|
)
|
|
if account is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="WhatsApp account not found")
|
|
return account
|
|
|
|
|
|
def _serialize_phone(phone: WhatsAppPhoneAsset) -> dict[str, Any]:
|
|
return {
|
|
"id": str(phone.id),
|
|
"phone_number_id": phone.phone_number_id,
|
|
"display_phone": phone.display_phone,
|
|
"verified_name": phone.verified_name,
|
|
"quality": phone.quality,
|
|
"status": phone.status,
|
|
"code_verification_status": phone.code_verification_status,
|
|
}
|
|
|
|
|
|
def serialize_account(account: WhatsAppBusinessAccount) -> dict[str, Any]:
|
|
phones = [_serialize_phone(phone) for phone in account.phones]
|
|
primary = next((phone for phone in phones if phone["status"] == "connected"), phones[0] if phones else None)
|
|
connection = account.connection
|
|
usable = account.status == "connected"
|
|
return {
|
|
"id": str(account.id),
|
|
"waba_id_masked": mask_identifier(account.waba_id),
|
|
"name": account.name,
|
|
"business_name": connection.business_name if connection else None,
|
|
"business_id_masked": mask_identifier(connection.meta_business_id) if connection else None,
|
|
"status": account.status,
|
|
"usable": usable,
|
|
"messaging_status": account.messaging_status if usable else "unavailable",
|
|
"management_status": account.management_status if usable or account.status == "incomplete" else "unavailable",
|
|
"templates_status": account.templates_status if account.templates_status == "active" else "unavailable",
|
|
"webhooks_status": account.webhooks_status if account.webhooks_status == "active" else "unavailable",
|
|
"template_count": account.template_count,
|
|
"quality": primary["quality"] if primary else None,
|
|
"primary_phone": primary,
|
|
"phones": phones,
|
|
"last_error": account.last_error,
|
|
"updated_at": account.updated_at,
|
|
}
|
|
|
|
|
|
def list_accounts(db: Session, user: UserAccount) -> list[dict[str, Any]]:
|
|
rows = db.scalars(
|
|
select(WhatsAppBusinessAccount)
|
|
.options(selectinload(WhatsAppBusinessAccount.phones), selectinload(WhatsAppBusinessAccount.connection))
|
|
.where(WhatsAppBusinessAccount.user_id == user.id)
|
|
.order_by(WhatsAppBusinessAccount.created_at.desc())
|
|
).all()
|
|
return [serialize_account(row) for row in rows]
|
|
|
|
|
|
async def _load_waba(token: str, waba_id: str) -> dict[str, Any]:
|
|
return await _graph(
|
|
"GET",
|
|
waba_id,
|
|
token=token,
|
|
params={"fields": "id,name,account_review_status,owner_business_info"},
|
|
)
|
|
|
|
|
|
async def _load_phones(token: str, waba_id: str) -> list[dict[str, Any]]:
|
|
data = await _graph(
|
|
"GET",
|
|
f"{waba_id}/phone_numbers",
|
|
token=token,
|
|
params={
|
|
"fields": "id,display_phone_number,verified_name,quality_rating,code_verification_status,status",
|
|
"limit": "50",
|
|
},
|
|
)
|
|
rows = data.get("data")
|
|
return [row for row in rows if isinstance(row, dict)] if isinstance(rows, list) else []
|
|
|
|
|
|
async def _subscribe(token: str, waba_id: str) -> bool:
|
|
try:
|
|
await _graph("POST", f"{waba_id}/subscribed_apps", token=token)
|
|
return True
|
|
except HTTPException:
|
|
logger.warning("WhatsApp subscribed_apps failed for waba %s", waba_id[-4:])
|
|
return False
|
|
|
|
|
|
async def _template_count(token: str, waba_id: str) -> tuple[int, bool]:
|
|
try:
|
|
data = await _graph(
|
|
"GET",
|
|
f"{waba_id}/message_templates",
|
|
token=token,
|
|
params={"limit": "1", "summary": "total_count"},
|
|
)
|
|
except HTTPException:
|
|
return 0, False
|
|
summary = data.get("summary")
|
|
if isinstance(summary, dict) and isinstance(summary.get("total_count"), int):
|
|
return summary["total_count"], True
|
|
rows = data.get("data")
|
|
return (len(rows) if isinstance(rows, list) else 0), True
|
|
|
|
|
|
def _claim_waba(db: Session, user: UserAccount, waba_id: str) -> WhatsAppBusinessAccount | None:
|
|
existing = db.scalar(select(WhatsAppBusinessAccount).where(WhatsAppBusinessAccount.waba_id == waba_id))
|
|
if existing is not None and existing.user_id != user.id:
|
|
return None
|
|
return existing
|
|
|
|
|
|
def _apply_phones(
|
|
db: Session,
|
|
user: UserAccount,
|
|
account: WhatsAppBusinessAccount,
|
|
phones: list[dict[str, Any]],
|
|
session_phone_id: str | None,
|
|
) -> None:
|
|
seen = {str(item.get("id")) for item in phones if item.get("id")}
|
|
if session_phone_id and session_phone_id not in seen:
|
|
phones.append({"id": session_phone_id})
|
|
current = {phone.phone_number_id: phone for phone in account.phones}
|
|
for item in phones:
|
|
phone_id = str(item.get("id") or "")
|
|
if not phone_id:
|
|
continue
|
|
other = db.scalar(select(WhatsAppPhoneAsset).where(WhatsAppPhoneAsset.phone_number_id == phone_id))
|
|
if other is not None and other.user_id != user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="This phone number is already connected to another Urmi workspace",
|
|
)
|
|
raw_status = str(item.get("status") or "PENDING").upper()
|
|
phone_status = "connected" if raw_status == "CONNECTED" else raw_status.lower()
|
|
row = current.get(phone_id) or other
|
|
if row is None:
|
|
row = WhatsAppPhoneAsset(
|
|
user_id=user.id,
|
|
account_id=account.id,
|
|
phone_number_id=phone_id,
|
|
)
|
|
db.add(row)
|
|
if row not in account.phones:
|
|
account.phones.append(row)
|
|
row.user_id = user.id
|
|
row.account_id = account.id
|
|
row.display_phone = str(item.get("display_phone_number") or row.display_phone or "")
|
|
row.verified_name = item.get("verified_name") or row.verified_name
|
|
row.quality = _quality_label(item.get("quality_rating"))
|
|
row.status = phone_status
|
|
row.code_verification_status = item.get("code_verification_status")
|
|
|
|
|
|
async def _provision_waba(
|
|
db: Session,
|
|
user: UserAccount,
|
|
connection: WhatsAppMetaConnection,
|
|
token: str,
|
|
waba_id: str,
|
|
session_phone_id: str | None,
|
|
) -> WhatsAppBusinessAccount:
|
|
claimed = _claim_waba(db, user, waba_id)
|
|
if claimed is None and db.scalar(select(WhatsAppBusinessAccount.id).where(WhatsAppBusinessAccount.waba_id == waba_id)):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="This WhatsApp Business Account is already connected to another Urmi workspace",
|
|
)
|
|
try:
|
|
waba = await _load_waba(token, waba_id)
|
|
except HTTPException as exc:
|
|
account = claimed or WhatsAppBusinessAccount(
|
|
user_id=user.id,
|
|
connection_id=connection.id,
|
|
waba_id=waba_id,
|
|
name="WhatsApp Business",
|
|
)
|
|
if claimed is None:
|
|
db.add(account)
|
|
db.flush()
|
|
detail = str(exc.detail)[:500]
|
|
expired = "expired" in detail.lower() or "validating access token" in detail.lower()
|
|
account.status = "expired" if expired else "failed"
|
|
account.last_error = detail
|
|
account.messaging_status = "unavailable"
|
|
account.management_status = "unavailable"
|
|
connection.status = "failed"
|
|
connection.last_error = account.last_error
|
|
_audit(db, user, "provision_failed", account.last_error or "", connection_id=connection.id, waba_id=waba_id)
|
|
return account
|
|
|
|
owner = waba.get("owner_business_info") if isinstance(waba.get("owner_business_info"), dict) else {}
|
|
business_id = str(owner.get("id") or connection.meta_business_id or "")
|
|
business_name = str(owner.get("name") or connection.business_name or waba.get("name") or "WhatsApp Business")
|
|
connection.meta_business_id = business_id or connection.meta_business_id
|
|
connection.business_name = business_name
|
|
connection.status = "connected"
|
|
connection.last_error = None
|
|
connection.access_token = token
|
|
|
|
account = claimed or WhatsAppBusinessAccount(
|
|
user_id=user.id,
|
|
connection_id=connection.id,
|
|
waba_id=waba_id,
|
|
)
|
|
if claimed is None:
|
|
db.add(account)
|
|
db.flush()
|
|
previous_connection_id = account.connection_id if claimed is not None else None
|
|
account.user_id = user.id
|
|
account.connection_id = connection.id
|
|
if previous_connection_id and previous_connection_id != connection.id:
|
|
previous = db.get(WhatsAppMetaConnection, previous_connection_id)
|
|
if previous is not None and previous.user_id == user.id:
|
|
previous.access_token = None
|
|
account.name = str(waba.get("name") or business_name)
|
|
account.last_error = None
|
|
account.management_status = "active"
|
|
|
|
try:
|
|
phones = await _load_phones(token, waba_id)
|
|
except HTTPException as exc:
|
|
phones = []
|
|
account.last_error = str(exc.detail)[:500]
|
|
_apply_phones(db, user, account, phones, session_phone_id)
|
|
|
|
subscribed = await _subscribe(token, waba_id)
|
|
template_count, templates_ok = await _template_count(token, waba_id)
|
|
account.template_count = template_count
|
|
account.templates_status = "active" if templates_ok else "unavailable"
|
|
account.webhooks_status = "active" if subscribed else "unavailable"
|
|
|
|
has_connected_phone = any(phone.status == "connected" for phone in account.phones)
|
|
if has_connected_phone and account.management_status == "active":
|
|
account.status = "connected"
|
|
account.messaging_status = "active"
|
|
_audit(db, user, "connected", "WhatsApp Business connection verified", connection_id=connection.id, waba_id=waba_id)
|
|
else:
|
|
account.status = "incomplete"
|
|
account.messaging_status = "unavailable"
|
|
account.last_error = account.last_error or "Phone number verification is not complete"
|
|
_audit(
|
|
db,
|
|
user,
|
|
"provision_incomplete",
|
|
account.last_error,
|
|
connection_id=connection.id,
|
|
waba_id=waba_id,
|
|
)
|
|
return account
|
|
|
|
|
|
async def complete_signup(db: Session, user: UserAccount, payload: dict[str, Any]) -> dict[str, Any]:
|
|
event = str(payload.get("event") or "").upper()
|
|
error_message = str(payload.get("error_message") or "").strip()[:500]
|
|
if event == "CANCEL":
|
|
_audit(db, user, "signup_cancelled", "Customer closed Embedded Signup")
|
|
db.commit()
|
|
return {"status": "cancelled", "message": "WhatsApp setup was cancelled. Nothing was connected.", "accounts": []}
|
|
if event == "ERROR" or event == "DENIED":
|
|
_audit(db, user, "signup_denied", error_message or "Meta denied the requested permissions")
|
|
db.commit()
|
|
return {
|
|
"status": "denied",
|
|
"message": error_message or "Meta did not grant the permissions Urmi needs. You can try again.",
|
|
"accounts": [],
|
|
}
|
|
if not event.startswith("FINISH"):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unrecognized Embedded Signup result")
|
|
|
|
code = str(payload.get("code") or "").strip()
|
|
if not code:
|
|
_audit(db, user, "provision_failed", "Embedded Signup finished without an authorization code")
|
|
db.commit()
|
|
return {
|
|
"status": "failed",
|
|
"message": "Meta finished without an authorization code. Start setup again.",
|
|
"accounts": [],
|
|
}
|
|
|
|
token = await _exchange_code(db, code)
|
|
debug = await _debug_token(db, token)
|
|
if debug.get("is_valid") is False:
|
|
_audit(db, user, "authorization_expired", "Meta authorization was not valid")
|
|
db.commit()
|
|
return {
|
|
"status": "expired",
|
|
"message": "The Meta authorization expired before Urmi could finish setup. Start setup again.",
|
|
"accounts": [],
|
|
}
|
|
|
|
session_waba = str(payload.get("waba_id") or "").strip()
|
|
waba_ids = [session_waba] if session_waba else _waba_ids_from_debug(debug)
|
|
if not waba_ids:
|
|
_audit(db, user, "provision_failed", "Embedded Signup did not return a WhatsApp Business Account")
|
|
db.commit()
|
|
return {
|
|
"status": "failed",
|
|
"message": "Meta did not return a WhatsApp Business Account. Finish business and number setup in the Meta window, then try again.",
|
|
"accounts": [],
|
|
}
|
|
|
|
scopes = debug.get("scopes")
|
|
granted = ",".join(str(item) for item in scopes) if isinstance(scopes, list) else None
|
|
connection = WhatsAppMetaConnection(
|
|
user_id=user.id,
|
|
status="pending",
|
|
meta_business_id=str(payload.get("business_id") or "") or None,
|
|
granted_scopes=granted,
|
|
access_token=token,
|
|
)
|
|
db.add(connection)
|
|
db.flush()
|
|
_audit(db, user, "authorization_received", "Embedded Signup returned an authorization code", connection_id=connection.id)
|
|
|
|
session_phone = str(payload.get("phone_number_id") or "").strip() or None
|
|
accounts: list[WhatsAppBusinessAccount] = []
|
|
for waba_id in waba_ids:
|
|
accounts.append(await _provision_waba(db, user, connection, token, waba_id, session_phone))
|
|
db.commit()
|
|
for account in accounts:
|
|
db.refresh(account)
|
|
|
|
statuses = {account.status for account in accounts}
|
|
if "connected" in statuses and statuses <= {"connected"}:
|
|
overall = "connected"
|
|
message = "WhatsApp Business is connected."
|
|
elif "connected" in statuses:
|
|
overall = "connected"
|
|
message = "Some WhatsApp Business accounts are connected. Others still need attention."
|
|
elif "incomplete" in statuses:
|
|
overall = "incomplete"
|
|
message = "Meta authorization succeeded, but phone verification is not complete. Finish verification in Meta, then reconnect."
|
|
else:
|
|
overall = "failed"
|
|
message = accounts[0].last_error or "Urmi could not finish WhatsApp setup."
|
|
return {"status": overall, "message": message, "accounts": [serialize_account(account) for account in accounts]}
|
|
|
|
|
|
async def refresh_account(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
account = _owned_account(db, user, account_id)
|
|
connection = account.connection
|
|
token = connection.access_token if connection else None
|
|
if connection is None or not token or connection.status == "disconnected":
|
|
account.status = "expired"
|
|
account.messaging_status = "unavailable"
|
|
account.last_error = "Authorization expired. Reconnect this WhatsApp account."
|
|
_audit(db, user, "authorization_expired", account.last_error, connection_id=connection.id if connection else None, waba_id=account.waba_id)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return serialize_account(account)
|
|
await _provision_waba(db, user, connection, token, account.waba_id, None)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return serialize_account(account)
|
|
|
|
|
|
async def disconnect_account(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
account = _owned_account(db, user, account_id)
|
|
connection = account.connection
|
|
token = connection.access_token if connection else None
|
|
if token:
|
|
try:
|
|
await _graph("DELETE", f"{account.waba_id}/subscribed_apps", token=token)
|
|
except HTTPException:
|
|
logger.info("Could not unsubscribe WABA %s during disconnect", account.waba_id[-4:])
|
|
account.status = "disconnected"
|
|
account.messaging_status = "unavailable"
|
|
account.management_status = "unavailable"
|
|
account.templates_status = "unavailable"
|
|
account.webhooks_status = "unavailable"
|
|
account.last_error = None
|
|
for phone in account.phones:
|
|
phone.status = "disconnected"
|
|
if connection is not None:
|
|
siblings_remain = (
|
|
db.scalar(
|
|
select(WhatsAppBusinessAccount.id).where(
|
|
WhatsAppBusinessAccount.connection_id == connection.id,
|
|
WhatsAppBusinessAccount.id != account.id,
|
|
WhatsAppBusinessAccount.status != "disconnected",
|
|
)
|
|
)
|
|
is not None
|
|
)
|
|
if not siblings_remain:
|
|
connection.status = "disconnected"
|
|
connection.access_token = None
|
|
connection.disconnected_at = datetime.now(timezone.utc)
|
|
_audit(db, user, "disconnected", "Customer disconnected this WhatsApp Business account", connection_id=connection.id if connection else None, waba_id=account.waba_id)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return serialize_account(account)
|
|
|
|
|
|
def account_detail(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
return serialize_account(_owned_account(db, user, account_id))
|
|
|
|
|
|
async def list_templates(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
account = _owned_account(db, user, account_id)
|
|
if account.templates_status != "active" or account.connection is None or not account.connection.access_token:
|
|
return {"available": False, "templates": []}
|
|
data = await _graph(
|
|
"GET",
|
|
f"{account.waba_id}/message_templates",
|
|
token=account.connection.access_token,
|
|
params={"fields": "name,category,language,status,quality_score", "limit": "50"},
|
|
)
|
|
rows = data.get("data") if isinstance(data.get("data"), list) else []
|
|
templates = []
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
quality = row.get("quality_score")
|
|
score = quality.get("score") if isinstance(quality, dict) else None
|
|
templates.append(
|
|
{
|
|
"name": row.get("name"),
|
|
"category": row.get("category"),
|
|
"language": row.get("language"),
|
|
"status": row.get("status"),
|
|
"quality": _quality_label(str(score) if score else None),
|
|
}
|
|
)
|
|
account.template_count = len(templates)
|
|
db.commit()
|
|
return {"available": True, "templates": templates}
|
|
|
|
|
|
def account_logs(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
account = _owned_account(db, user, account_id)
|
|
audits = db.scalars(
|
|
select(WhatsAppConnectionAudit)
|
|
.where(
|
|
WhatsAppConnectionAudit.user_id == user.id,
|
|
or_(
|
|
WhatsAppConnectionAudit.waba_id == account.waba_id,
|
|
WhatsAppConnectionAudit.connection_id == account.connection_id,
|
|
),
|
|
)
|
|
.order_by(WhatsAppConnectionAudit.created_at.desc())
|
|
.limit(50)
|
|
).all()
|
|
phone_ids = [phone.phone_number_id for phone in account.phones]
|
|
event_filter = [WhatsAppWebhookEvent.waba_id == account.waba_id]
|
|
if phone_ids:
|
|
event_filter.append(WhatsAppWebhookEvent.phone_number_id.in_(phone_ids))
|
|
events = db.scalars(
|
|
select(WhatsAppWebhookEvent)
|
|
.where(or_(*event_filter))
|
|
.order_by(WhatsAppWebhookEvent.created_at.desc())
|
|
.limit(25)
|
|
).all()
|
|
return {
|
|
"audits": [
|
|
{
|
|
"id": str(item.id),
|
|
"event": item.event,
|
|
"detail": item.detail,
|
|
"created_at": item.created_at,
|
|
}
|
|
for item in audits
|
|
],
|
|
"events": [
|
|
{
|
|
"id": str(item.id),
|
|
"field": item.field,
|
|
"event_type": item.event_type,
|
|
"from_wa_id": item.from_wa_id,
|
|
"status": item.status,
|
|
"created_at": item.created_at,
|
|
}
|
|
for item in events
|
|
],
|
|
}
|
|
|
|
|
|
def developer_config(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
|
account = _owned_account(db, user, account_id)
|
|
return {
|
|
"webhook_url": settings.whatsapp_webhook_url,
|
|
"verify_token_configured": bool(settings.whatsapp_verify_token),
|
|
"verify_token_visible": False,
|
|
"recommended_fields": RECOMMENDED_WEBHOOK_FIELDS,
|
|
"webhooks_status": account.webhooks_status,
|
|
"docs_url": "https://developers.facebook.com/documentation/business-messaging/whatsapp/embedded-signup",
|
|
}
|
|
|
|
|
|
def customer_token_for_phone(db: Session, user: UserAccount, phone_number_id: str) -> tuple[str, str]:
|
|
phone = db.scalar(
|
|
select(WhatsAppPhoneAsset)
|
|
.options(selectinload(WhatsAppPhoneAsset.account).selectinload(WhatsAppBusinessAccount.connection))
|
|
.where(WhatsAppPhoneAsset.phone_number_id == phone_number_id, WhatsAppPhoneAsset.user_id == user.id)
|
|
)
|
|
if phone is None or phone.account.status != "connected" or phone.status != "connected":
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No connected WhatsApp number for this workspace")
|
|
token = phone.account.connection.access_token if phone.account.connection else None
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before sending")
|
|
return token, phone.phone_number_id
|