"""WhatsApp message template lifecycle against the customer's WABA.""" from __future__ import annotations import re from typing import Any import httpx from fastapi import HTTPException, status from sqlalchemy import func, or_, select from sqlalchemy.orm import Session, selectinload from .config import settings from .models import UserAccount, WhatsAppBusinessAccount, WhatsAppMessage, WhatsAppMessageTemplate, WhatsAppMetaConnection from .whatsapp_onboarding import ( _apply_phones, _debug_token, _graph, _graph_error_message, _load_phones, _load_waba, _owned_account, _quality_label, _token_app_id, _waba_ids_from_debug, ) _NAME = re.compile(r"^[a-z0-9_]{1,512}$") _LANG = re.compile(r"^[a-z]{2}(_[A-Z]{2})?$") _VAR = re.compile(r"\{\{([A-Za-z0-9_]+)\}\}") _MEDIA = { "image/jpeg": "IMAGE", "image/png": "IMAGE", "video/mp4": "VIDEO", "application/pdf": "DOCUMENT", } LANGUAGES = [ ("en", "English"), ("en_US", "English (US)"), ("en_GB", "English (UK)"), ("hi", "Hindi"), ("bn", "Bengali"), ("ta", "Tamil"), ("te", "Telugu"), ("mr", "Marathi"), ("gu", "Gujarati"), ("kn", "Kannada"), ("ml", "Malayalam"), ("pa", "Punjabi"), ("ur", "Urdu"), ("ar", "Arabic"), ("es", "Spanish"), ("es_MX", "Spanish (Mexico)"), ("pt_BR", "Portuguese (Brazil)"), ("fr", "French"), ("de", "German"), ("id", "Indonesian"), ("it", "Italian"), ("nl", "Dutch"), ("ru", "Russian"), ("tr", "Turkish"), ("ja", "Japanese"), ("ko", "Korean"), ("zh_CN", "Chinese (China)"), ("zh_HK", "Chinese (Hong Kong)"), ] def language_catalog() -> list[dict[str, str]]: return [{"code": code, "label": label} for code, label in LANGUAGES] def _language_ok(code: str) -> bool: return any(item[0] == code for item in LANGUAGES) def _connection_method(connection: WhatsAppMetaConnection | None) -> str: if connection is None: return "WhatsApp" if connection.granted_scopes == "manual": return "Saved token" if (connection.granted_scopes or "").strip(): return "Embedded signup" return "WhatsApp" def _stored_token(connection: WhatsAppMetaConnection | None) -> str | None: if connection is None or connection.status == "disconnected": return None token = (connection.access_token or "").strip() return token or None def _candidate_tokens(db: Session, account: WhatsAppBusinessAccount) -> list[str]: """Embedded signup and a saved token are both stored on the connection.""" tokens: list[str] = [] own = _stored_token(account.connection) if own: tokens.append(own) others = db.scalars( select(WhatsAppMetaConnection).where( WhatsAppMetaConnection.user_id == account.user_id, WhatsAppMetaConnection.status != "disconnected", ) ).all() for connection in others: token = _stored_token(connection) if token and token not in tokens: tokens.append(token) platform_waba = settings.whatsapp_business_account_id.strip() platform_token = settings.whatsapp_access_token.strip() if platform_token and account.waba_id == platform_waba and platform_token not in tokens: tokens.append(platform_token) return tokens def _token(db: Session, account: WhatsAppBusinessAccount) -> str: tokens = _candidate_tokens(db, account) if not tokens: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before managing templates.", ) return tokens[0] def _retry_with_another_token(exc: HTTPException) -> bool: detail = str(exc.detail).lower() return any(part in detail for part in ("access token", "oauth", "(#190)", "permission", "unsupported get", "(#10)")) def _variables(text: str) -> list[str]: found: list[str] = [] for match in _VAR.finditer(text or ""): name = match.group(1) if name not in found: found.append(name) return found def _examples_for(text: str, examples: list[str], named: bool) -> list[str]: names = _variables(text) if len(examples) < len(names): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Every variable needs a sample value.") if named and any(name.isdigit() for name in names): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Named variables cannot be only numbers. Use a name such as customer_name.") if not named: expected = [str(index) for index in range(1, len(names) + 1)] if names != expected: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Numbered variables must be {{1}}, {{2}}, and so on, with no gaps.") return examples[: len(names)] def build_components(payload: dict[str, Any]) -> list[dict[str, Any]]: category = str(payload.get("category") or "").upper() named = str(payload.get("parameter_format") or "POSITIONAL").upper() == "NAMED" if category == "AUTHENTICATION": auth = payload.get("auth") if isinstance(payload.get("auth"), dict) else {} minutes = int(auth.get("code_expiration_minutes") or 0) components: list[dict[str, Any]] = [{ "type": "BODY", "add_security_recommendation": bool(auth.get("add_security_recommendation", True)), }] if minutes: if minutes < 1 or minutes > 90: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Code expiry should be between 1 and 90 minutes.") components.append({"type": "FOOTER", "code_expiration_minutes": minutes}) components.append({ "type": "BUTTONS", "buttons": [{ "type": "OTP", "otp_type": "COPY_CODE", "text": str(auth.get("copy_label") or "Copy code")[:25] or "Copy code", }], }) return components header = payload.get("header") if isinstance(payload.get("header"), dict) else {"type": "NONE"} body = str(payload.get("body") or "").strip() footer = str(payload.get("footer") or "").strip() buttons = payload.get("buttons") if isinstance(payload.get("buttons"), list) else [] if not body: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Message body is required.") if len(body) > 1024: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Message body can be at most 1,024 characters.") if len(footer) > 60: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Footer can be at most 60 characters.") if _variables(footer): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Footer cannot contain variables.") components = [] header_type = str(header.get("type") or "NONE").upper() if header_type == "TEXT": text = str(header.get("text") or "").strip() if not text: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Header text is required.") if len(text) > 60: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Header text can be at most 60 characters.") samples = _examples_for(text, list(header.get("examples") or []), named) block: dict[str, Any] = {"type": "HEADER", "format": "TEXT", "text": text} if samples: names = _variables(text) if named: block["example"] = {"header_text_named_params": [ {"param_name": name, "example": sample} for name, sample in zip(names, samples, strict=False) ]} else: block["example"] = {"header_text": samples} components.append(block) elif header_type in {"IMAGE", "VIDEO", "DOCUMENT"}: handle = str(header.get("handle") or "").strip() if not handle: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Upload a header file before submitting.") components.append({ "type": "HEADER", "format": header_type, "example": {"header_handle": [handle]}, }) elif header_type == "LOCATION": components.append({"type": "HEADER", "format": "LOCATION"}) elif header_type != "NONE": raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Choose a supported header type.") body_samples = _examples_for(body, [str(item) for item in (payload.get("body_examples") or [])], named) body_block: dict[str, Any] = {"type": "BODY", "text": body} if body_samples: names = _variables(body) if named: body_block["example"] = {"body_text_named_params": [ {"param_name": name, "example": sample} for name, sample in zip(names, body_samples, strict=False) ]} else: body_block["example"] = {"body_text": [body_samples]} components.append(body_block) if footer: components.append({"type": "FOOTER", "text": footer}) if buttons: components.append({"type": "BUTTONS", "buttons": _buttons(buttons, category)}) return components def _buttons(buttons: list[Any], category: str) -> list[dict[str, Any]]: if len(buttons) > 10: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="A template can have at most 10 buttons.") built: list[dict[str, Any]] = [] urls = phones = copies = 0 for raw in buttons: if not isinstance(raw, dict): continue kind = str(raw.get("type") or "").upper() label = str(raw.get("text") or "").strip() if kind != "COPY_CODE" and (not label or len(label) > 25): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Each button label should be 1 to 25 characters.") if kind == "URL": urls += 1 url = str(raw.get("url") or "").strip() if not url.startswith("https://"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Website buttons need an https URL.") button: dict[str, Any] = {"type": "URL", "text": label, "url": url} names = _variables(url) if len(names) > 1 or (names and not url.endswith("{{" + names[0] + "}}")): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="A website button can use one variable, and it has to be at the end of the URL.") if names: example = str(raw.get("example") or "").strip() if not example.startswith("https://"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Sample URL should be a full https address with the variable filled in.") button["example"] = [example] built.append(button) elif kind == "PHONE_NUMBER": phones += 1 phone = re.sub(r"\D", "", str(raw.get("phone") or "")) if len(phone) < 8: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Enter a phone number with the country code.") built.append({"type": "PHONE_NUMBER", "text": label, "phone_number": phone}) elif kind == "QUICK_REPLY": built.append({"type": "QUICK_REPLY", "text": label}) elif kind == "COPY_CODE": if category != "MARKETING": raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Copy offer code is available on marketing templates.") copies += 1 example = str(raw.get("example") or "").strip() if not example or len(example) > 15: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Offer code sample should be 1 to 15 characters.") built.append({"type": "COPY_CODE", "example": example}) elif kind == "CATALOG": built.append({"type": "CATALOG", "text": label}) elif kind == "FLOW": flow_id = str(raw.get("flow_id") or "").strip() if not flow_id: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="A WhatsApp Flow button needs the Flow ID from Meta.") built.append({ "type": "FLOW", "text": label, "flow_id": flow_id, "flow_action": "navigate", "navigate_screen": str(raw.get("screen") or "MAIN"), }) else: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="That button type is not supported for this template.") if urls > 2 or phones > 1 or copies > 1: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Use at most two website buttons, one phone button, and one offer code.", ) kinds = [str(item.get("type") or "") for item in built] quick = [index for index, kind in enumerate(kinds) if kind == "QUICK_REPLY"] if quick and (quick != list(range(quick[0], quick[0] + len(quick))) or quick[0] not in {0, len(kinds) - len(quick)}): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Keep quick reply buttons together, either all at the start or all at the end.", ) return built def _definition_from_components(components: list[Any], category: str) -> dict[str, Any]: definition: dict[str, Any] = { "category": category, "subtype": "otp" if category == "AUTHENTICATION" else "default", "parameter_format": "POSITIONAL", "header": {"type": "NONE"}, "body": "", "body_examples": [], "footer": "", "buttons": [], "auth": { "otp_type": "COPY_CODE", "add_security_recommendation": True, "code_expiration_minutes": 0, "copy_label": "Copy code", }, } for component in components: if not isinstance(component, dict): continue kind = str(component.get("type") or "").upper() if kind == "HEADER": fmt = str(component.get("format") or "TEXT").upper() example = component.get("example") if isinstance(component.get("example"), dict) else {} if fmt == "TEXT": samples = example.get("header_text") if isinstance(example.get("header_text"), list) else [] definition["header"] = {"type": "TEXT", "text": component.get("text") or "", "examples": samples} elif fmt == "LOCATION": definition["header"] = {"type": "LOCATION"} else: handles = example.get("header_handle") if isinstance(example.get("header_handle"), list) else [] definition["header"] = {"type": fmt, "handle": handles[0] if handles else "", "preview_name": fmt.title()} elif kind == "BODY": definition["body"] = component.get("text") or "" example = component.get("example") if isinstance(component.get("example"), dict) else {} if isinstance(example.get("body_text"), list) and example["body_text"] and isinstance(example["body_text"][0], list): definition["body_examples"] = example["body_text"][0] named = example.get("body_text_named_params") if isinstance(named, list): definition["parameter_format"] = "NAMED" definition["body_examples"] = [item.get("example") for item in named if isinstance(item, dict)] if "add_security_recommendation" in component: definition["auth"]["add_security_recommendation"] = bool(component.get("add_security_recommendation")) elif kind == "FOOTER": definition["footer"] = component.get("text") or "" if component.get("code_expiration_minutes"): definition["auth"]["code_expiration_minutes"] = component.get("code_expiration_minutes") elif kind == "BUTTONS": for button in component.get("buttons") or []: if not isinstance(button, dict): continue button_type = str(button.get("type") or "").upper() if button_type == "OTP": definition["auth"]["copy_label"] = button.get("text") or "Copy code" continue definition["buttons"].append({ "type": button_type, "text": button.get("text") or "", "url": button.get("url") or "", "phone": button.get("phone_number") or "", "example": (button.get("example") or [""])[0] if isinstance(button.get("example"), list) else button.get("example") or "", "flow_id": button.get("flow_id") or "", "screen": button.get("navigate_screen") or "", }) return definition def _sent_counts(db: Session, user: UserAccount) -> dict[str, int]: rows = db.execute( select(WhatsAppMessage.body, func.count()) .where(WhatsAppMessage.user_id == user.id, WhatsAppMessage.message_type == "template") .group_by(WhatsAppMessage.body) ).all() counts: dict[str, int] = {} for body, count in rows: text = str(body or "") for prefix in ("Template: ", "Broadcast: "): if text.startswith(prefix): name = text[len(prefix):] counts[name] = counts.get(name, 0) + int(count) return counts def _serialize(row: WhatsAppMessageTemplate, account: WhatsAppBusinessAccount | None, sent: int) -> dict[str, Any]: phone = None phones: list[str] = [] connection = account.connection if account is not None else None if account is not None: phone = next((item for item in account.phones if item.phone_number_id == row.phone_number_id), None) if phone is None and account.phones: phone = account.phones[0] for item in account.phones: label = item.display_phone or item.verified_name or "" if label and label not in phones: phones.append(label) return { "id": str(row.id), "account_id": str(row.account_id), "account_name": account.name if account else "WhatsApp Business", "waba_name": account.name if account else "", "connection_name": (connection.business_name if connection and connection.business_name else (account.name if account else "WhatsApp")), "connection_method": _connection_method(connection), "phone_number_id": row.phone_number_id, "display_phone": phone.display_phone if phone else "", "phones": phones, "meta_template_id": row.meta_template_id, "name": row.name, "language": row.language, "category": row.category, "subtype": row.subtype, "status": row.status, "quality": row.quality, "rejected_reason": row.rejected_reason, "parameter_format": row.parameter_format, "definition": row.definition or {}, "components": row.components or [], "messages_sent": sent, "messages_sent_source": "urmi", "created_at": row.created_at, "updated_at": row.updated_at, } def _upsert_remote(db: Session, account: WhatsAppBusinessAccount, remote: dict[str, Any]) -> WhatsAppMessageTemplate: name = str(remote.get("name") or "") language = str(remote.get("language") or "en") row = db.scalar( select(WhatsAppMessageTemplate).where( WhatsAppMessageTemplate.waba_id == account.waba_id, WhatsAppMessageTemplate.name == name, WhatsAppMessageTemplate.language == language, ) ) components = remote.get("components") if isinstance(remote.get("components"), list) else [] category = str(remote.get("category") or "MARKETING").upper() quality = remote.get("quality_score") score = quality.get("score") if isinstance(quality, dict) else None reason = remote.get("rejected_reason") if isinstance(reason, dict): reason = reason.get("reason") or reason.get("message") if row is None: row = WhatsAppMessageTemplate( user_id=account.user_id, account_id=account.id, waba_id=account.waba_id, name=name, language=language, definition=_definition_from_components(components, category), ) db.add(row) row.meta_template_id = str(remote.get("id") or row.meta_template_id or "") or None row.category = category if components: definition = _definition_from_components(components, category) if row.subtype and row.subtype != "default": definition["subtype"] = row.subtype row.definition = definition row.parameter_format = str(definition.get("parameter_format") or row.parameter_format) row.subtype = str(definition.get("subtype") or row.subtype or "default")[:40] row.status = str(remote.get("status") or "PENDING").upper() row.quality = _quality_label(str(score) if score else None) row.rejected_reason = str(reason)[:2000] if reason and str(reason).upper() not in {"NONE", "NULL"} else None row.components = components if not row.definition: row.definition = _definition_from_components(components, category) return row async def _template_page(token: str, waba_id: str, params: dict[str, Any]) -> dict[str, Any]: last: HTTPException | None = None for fields in ( "id,name,language,status,category,components,quality_score,rejected_reason", "id,name,language,status,category,components", ): try: return await _graph("GET", f"{waba_id}/message_templates", token=token, params={**params, "fields": fields}) except HTTPException as exc: last = exc detail = str(exc.detail).lower() if "nonexisting field" in detail or "(#100)" in detail: continue raise if last is not None: raise last return {} async def _pull_templates(db: Session, account: WhatsAppBusinessAccount, token: str) -> set[tuple[str, str]]: seen: set[tuple[str, str]] = set() after = None for _ in range(5): params: dict[str, Any] = {"limit": "100"} if after: params["after"] = after data = await _template_page(token, account.waba_id, params) rows = data.get("data") if isinstance(data.get("data"), list) else [] for remote in rows: if isinstance(remote, dict) and remote.get("name"): _upsert_remote(db, account, remote) seen.add((str(remote.get("name")), str(remote.get("language") or ""))) paging = data.get("paging") if isinstance(data.get("paging"), dict) else {} cursors = paging.get("cursors") if isinstance(paging.get("cursors"), dict) else {} after = cursors.get("after") if paging.get("next") else None if not after: break return seen def _wrong_template_node(exc: HTTPException) -> bool: detail = str(exc.detail).lower() return "message_templates" in detail and "nonexisting field" in detail def _add_id(found: list[str], value: Any) -> None: text = str(value or "").strip() if text and text not in found: found.append(text) async def _discover_waba_ids(db: Session, token: str, business_id: str | None) -> list[str]: """WABA IDs this token can manage. Debug target_ids are often empty on embedded signup.""" found: list[str] = [] try: debug = await _debug_token(db, token) except HTTPException: debug = {} for waba in _waba_ids_from_debug(debug): _add_id(found, waba) lookups: list[tuple[str, dict[str, str]]] = [ ("me/assigned_whatsapp_business_accounts", {"fields": "id,name", "limit": "50"}), ] if business_id: for edge in ("owned_whatsapp_business_accounts", "client_whatsapp_business_accounts"): lookups.append((f"{business_id}/{edge}", {"fields": "id,name", "limit": "50"})) for path, params in lookups: try: data = await _graph("GET", path, token=token, params=params) except HTTPException: continue rows = data.get("data") if isinstance(data.get("data"), list) else [] for row in rows: if isinstance(row, dict): _add_id(found, row.get("id")) return found async def _waba_owning_phone(token: str, wabas: list[str], phone_id: str) -> str | None: for waba in wabas: if waba == phone_id: continue try: rows = await _load_phones(token, waba) except HTTPException: continue if phone_id in {str(row.get("id")) for row in rows if row.get("id")}: return waba return None async def _correct_waba_id(db: Session, account: WhatsAppBusinessAccount, token: str, *, allow_only_waba: bool) -> str: """Point a phone-number ID at its WABA. Returns updated, covered, or no.""" stored = str(account.waba_id) business_id = account.connection.meta_business_id if account.connection else None wabas = [waba for waba in await _discover_waba_ids(db, token, business_id) if waba != stored] real = await _waba_owning_phone(token, wabas, stored) if real is None and allow_only_waba and len(wabas) == 1: real = wabas[0] if not real: return "no" taken = db.scalar( select(WhatsAppBusinessAccount.id).where( WhatsAppBusinessAccount.waba_id == real, WhatsAppBusinessAccount.id != account.id, ) ) if taken is not None: owner = db.get(WhatsAppBusinessAccount, taken) if owner is not None and owner.user_id == account.user_id: return "covered" return "no" account.waba_id = real try: info = await _load_waba(token, real) name = str(info.get("name") or "").strip() if name: account.name = name[:200] except HTTPException: pass user = db.get(UserAccount, account.user_id) if user is not None: try: rows = await _load_phones(token, real) _apply_phones(db, user, account, rows, None) except HTTPException: pass db.flush() return "updated" async def sync_account_templates(db: Session, account: WhatsAppBusinessAccount) -> str | None: tokens = _candidate_tokens(db, account) if not tokens: if account.status == "disconnected": return None return "Reconnect this WhatsApp account to refresh its templates." own_token = _stored_token(account.connection) last_error: str | None = None for index, token in enumerate(tokens): try: seen = await _pull_templates(db, account, token) except HTTPException as exc: db.rollback() if not _wrong_template_node(exc): last_error = str(exc.detail) if _retry_with_another_token(exc) and index < len(tokens) - 1: continue return last_error try: outcome = await _correct_waba_id(db, account, token, allow_only_waba=token == own_token) except HTTPException: outcome = "no" if outcome == "covered": return None if outcome != "updated": last_error = str(exc.detail) if index < len(tokens) - 1: continue return last_error try: seen = await _pull_templates(db, account, token) except HTTPException as retry_exc: db.rollback() last_error = str(retry_exc.detail) if index < len(tokens) - 1: continue return last_error stored = db.scalars( select(WhatsAppMessageTemplate).where(WhatsAppMessageTemplate.waba_id == account.waba_id) ).all() for row in stored: if row.meta_template_id and (row.name, row.language) not in seen: row.status = "DISABLED" account.template_count = len(seen) account.templates_status = "active" db.commit() return None return last_error async def list_managed_templates(db: Session, user: UserAccount, account_id: str | None) -> dict[str, Any]: query = ( select(WhatsAppBusinessAccount) .options(selectinload(WhatsAppBusinessAccount.phones), selectinload(WhatsAppBusinessAccount.connection)) .where(WhatsAppBusinessAccount.user_id == user.id) ) accounts = db.scalars(query).all() if account_id: accounts = [account for account in accounts if str(account.id) == account_id] warnings = [] for account in accounts: warning = await sync_account_templates(db, account) if warning: warnings.append(f"{account.name}: {warning}") rows = db.scalars( select(WhatsAppMessageTemplate) .where( WhatsAppMessageTemplate.user_id == user.id, WhatsAppMessageTemplate.account_id.in_([account.id for account in accounts] or [user.id]), ) .order_by(WhatsAppMessageTemplate.updated_at.desc()) ).all() if accounts else [] by_id = {account.id: account for account in accounts} sent = _sent_counts(db, user) templates = [_serialize(row, by_id.get(row.account_id), sent.get(row.name, 0)) for row in rows] summary = { "total": len(templates), "approved": sum(1 for item in templates if item["status"] == "APPROVED"), "pending": sum(1 for item in templates if item["status"] in {"PENDING", "IN_APPEAL"}), "rejected": sum(1 for item in templates if item["status"] == "REJECTED"), "paused": sum(1 for item in templates if item["status"] == "PAUSED"), } return {"templates": templates, "summary": summary, "warnings": warnings} def _owned_template(db: Session, user: UserAccount, template_id: Any) -> tuple[WhatsAppMessageTemplate, WhatsAppBusinessAccount]: row = db.scalar( select(WhatsAppMessageTemplate).where( WhatsAppMessageTemplate.id == template_id, WhatsAppMessageTemplate.user_id == user.id, ) ) if row is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found") account = _owned_account(db, user, row.account_id) return row, account def get_managed_template(db: Session, user: UserAccount, template_id: Any) -> dict[str, Any]: row, account = _owned_template(db, user, template_id) sent = _sent_counts(db, user).get(row.name, 0) payload = _serialize(row, account, sent) bodies = [f"Template: {row.name}", f"Broadcast: {row.name}"] status_rows = db.execute( select(WhatsAppMessage.status, func.count()) .where( WhatsAppMessage.user_id == user.id, WhatsAppMessage.message_type == "template", or_(*[WhatsAppMessage.body == body for body in bodies]), ) .group_by(WhatsAppMessage.status) ).all() payload["analytics"] = { "source": "urmi", "note": "Counts are messages Urmi sent with this template name, plus delivery updates Urmi received. They are separate from Meta's template analytics.", "sent": sent, "by_status": {str(item or "unknown"): int(count) for item, count in status_rows}, } return payload def _validate_identity(payload: dict[str, Any]) -> tuple[str, str, str]: name = str(payload.get("name") or "").strip().lower() language = str(payload.get("language") or "en_US").strip() category = str(payload.get("category") or "").upper() if not _NAME.match(name): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Template name should use lowercase letters, numbers, and underscores.") if not _language_ok(language): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Choose a language from the list.") if category not in {"MARKETING", "UTILITY", "AUTHENTICATION"}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Choose Marketing, Utility, or Authentication.") return name, language, category async def create_managed_template(db: Session, user: UserAccount, payload: dict[str, Any]) -> dict[str, Any]: account = _owned_account(db, user, payload.get("account_id")) token = _token(db, account) name, language, category = _validate_identity(payload) existing = db.scalar( select(WhatsAppMessageTemplate).where( WhatsAppMessageTemplate.waba_id == account.waba_id, WhatsAppMessageTemplate.name == name, WhatsAppMessageTemplate.language == language, ) ) if existing is not None and existing.status != "DISABLED": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="This template name and language already exist on the WhatsApp account.") if existing is not None: db.delete(existing) db.flush() components = build_components(payload) body: dict[str, Any] = {"name": name, "language": language, "category": category, "components": components} if category != "AUTHENTICATION": body["parameter_format"] = "NAMED" if str(payload.get("parameter_format") or "").upper() == "NAMED" else "POSITIONAL" data = await _graph("POST", f"{account.waba_id}/message_templates", token=token, json_body=body) phone_id = str(payload.get("phone_number_id") or "").strip() or None if phone_id and phone_id not in {phone.phone_number_id for phone in account.phones}: phone_id = None row = WhatsAppMessageTemplate( user_id=user.id, account_id=account.id, waba_id=account.waba_id, phone_number_id=phone_id, meta_template_id=str(data.get("id") or "") or None, name=name, language=language, category=str(data.get("category") or category).upper(), subtype=str(payload.get("subtype") or "default")[:40], status=str(data.get("status") or "PENDING").upper(), parameter_format="NAMED" if str(payload.get("parameter_format") or "").upper() == "NAMED" else "POSITIONAL", definition=payload, components=components, ) db.add(row) db.commit() db.refresh(row) return _serialize(row, account, 0) async def resubmit_managed_template(db: Session, user: UserAccount, template_id: Any, payload: dict[str, Any]) -> dict[str, Any]: row, account = _owned_template(db, user, template_id) if row.status not in {"REJECTED", "PAUSED", "APPROVED"}: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="This template can be edited after Meta rejects it, pauses it, or approves it.") if not row.meta_template_id: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Meta has no id for this template yet.") _validate_identity({**payload, "name": row.name, "language": row.language, "category": payload.get("category") or row.category}) components = build_components({**payload, "name": row.name, "language": row.language}) data = await _graph("POST", row.meta_template_id, token=_token(db, account), json_body={"components": components, "category": row.category}) row.components = components row.definition = payload row.status = str(data.get("status") or "PENDING").upper() row.rejected_reason = None row.category = str(data.get("category") or row.category).upper() db.commit() db.refresh(row) return _serialize(row, account, _sent_counts(db, user).get(row.name, 0)) async def delete_managed_template(db: Session, user: UserAccount, template_id: Any) -> dict[str, str]: row, account = _owned_template(db, user, template_id) params: dict[str, Any] = {"name": row.name} if row.meta_template_id: params["hsm_id"] = row.meta_template_id await _graph("DELETE", f"{account.waba_id}/message_templates", token=_token(db, account), params=params) db.delete(row) db.commit() return {"status": "deleted"} async def upload_header_media(db: Session, user: UserAccount, account_id: Any, filename: str, content_type: str, data: bytes) -> dict[str, str]: account = _owned_account(db, user, account_id) token = _token(db, account) kind = _MEDIA.get(content_type) if kind is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Upload a JPEG, PNG, MP4, or PDF.") if len(data) > 16 * 1024 * 1024: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Header files can be at most 16 MB.") app_id = (account.connection.meta_app_id if account.connection else None) or settings.meta_app_id.strip() if not app_id: app_id = await _token_app_id(token) or "" if not app_id: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="This connection has no Meta app id, so the header file cannot be uploaded.") session = await _graph( "POST", f"{app_id}/uploads", token=token, params={"file_name": filename or "header", "file_length": str(len(data)), "file_type": content_type}, ) upload_id = str(session.get("id") or "") if not upload_id: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Meta did not start the header upload.") url = f"https://graph.facebook.com/{settings.whatsapp_api_version}/{upload_id}" async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post( url, content=data, headers={"Authorization": f"OAuth {token}", "file_offset": "0", "Content-Type": content_type}, ) try: body = response.json() except ValueError: body = {} if response.status_code >= 400: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=_graph_error_message(body, "Meta did not accept the header file.")) handle = str(body.get("h") or "") if not handle: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Meta did not return a header file handle.") return {"handle": handle, "format": kind} def apply_template_status_webhook(db: Session, payload: dict[str, Any]) -> None: changed = False for entry in payload.get("entry") or []: if not isinstance(entry, dict): continue for change in entry.get("changes") or []: if not isinstance(change, dict) or change.get("field") != "message_template_status_update": continue value = change.get("value") if isinstance(change.get("value"), dict) else {} name = str(value.get("message_template_name") or "") language = str(value.get("message_template_language") or "") event = str(value.get("event") or "").upper() if not name or not event: continue rows = db.scalars( select(WhatsAppMessageTemplate).where( WhatsAppMessageTemplate.name == name, or_(WhatsAppMessageTemplate.language == language, language == ""), ) ).all() reason = value.get("reason") for row in rows: row.status = event if reason and str(reason).upper() not in {"NONE", "NULL"}: row.rejected_reason = str(reason)[:2000] meta_id = str(value.get("message_template_id") or "") if meta_id: row.meta_template_id = meta_id changed = True if changed: db.commit()