122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, UploadFile
|
|
from fastapi.responses import Response
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .auth import current_user
|
|
from .database import get_db
|
|
from .models import UserAccount
|
|
from .whatsapp_template_service import (
|
|
attach_template_sample,
|
|
create_managed_template,
|
|
delete_managed_template,
|
|
get_managed_template,
|
|
language_catalog,
|
|
list_managed_templates,
|
|
read_template_sample,
|
|
resubmit_managed_template,
|
|
upload_header_media,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/v1/whatsapp/message-templates", tags=["whatsapp-templates"])
|
|
|
|
|
|
class TemplateWriteRequest(BaseModel):
|
|
account_id: str | None = None
|
|
phone_number_id: str | None = None
|
|
name: str = Field(min_length=1, max_length=512)
|
|
language: str = "en_US"
|
|
category: str
|
|
subtype: str = "default"
|
|
parameter_format: str = "POSITIONAL"
|
|
header: dict[str, Any] = Field(default_factory=lambda: {"type": "NONE"})
|
|
body: str = ""
|
|
body_examples: list[str] = Field(default_factory=list)
|
|
footer: str = ""
|
|
buttons: list[dict[str, Any]] = Field(default_factory=list)
|
|
auth: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
@router.get("/languages")
|
|
def whatsapp_template_languages(_: UserAccount = Depends(current_user)) -> list[dict[str, str]]:
|
|
return language_catalog()
|
|
|
|
|
|
@router.get("")
|
|
async def whatsapp_template_list(
|
|
account_id: str | None = None,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
return await list_managed_templates(db, user, account_id)
|
|
|
|
|
|
@router.post("/media")
|
|
async def whatsapp_template_media(
|
|
account_id: str = Form(),
|
|
file: UploadFile = File(),
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
data = await file.read()
|
|
return await upload_header_media(db, user, account_id, file.filename or "header", file.content_type or "", data)
|
|
|
|
|
|
@router.get("/{template_id}")
|
|
def whatsapp_template_detail(
|
|
template_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
return get_managed_template(db, user, template_id)
|
|
|
|
|
|
@router.post("")
|
|
async def whatsapp_template_create(
|
|
payload: TemplateWriteRequest,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
return await create_managed_template(db, user, payload.model_dump())
|
|
|
|
|
|
@router.post("/{template_id}/sample")
|
|
async def whatsapp_template_sample(
|
|
template_id: str,
|
|
file: UploadFile = File(),
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
data = await file.read()
|
|
return attach_template_sample(db, user, template_id, file.filename or "header", file.content_type or "", data)
|
|
|
|
|
|
@router.get("/{template_id}/sample")
|
|
def whatsapp_template_sample_file(
|
|
template_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> Response:
|
|
data, mime, _name = read_template_sample(db, user, template_id)
|
|
return Response(content=data, media_type=mime)
|
|
|
|
|
|
@router.post("/{template_id}/resubmit")
|
|
async def whatsapp_template_resubmit(
|
|
template_id: str,
|
|
payload: TemplateWriteRequest,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
return await resubmit_managed_template(db, user, template_id, payload.model_dump())
|
|
|
|
|
|
@router.delete("/{template_id}")
|
|
async def whatsapp_template_delete(
|
|
template_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: UserAccount = Depends(current_user),
|
|
) -> dict:
|
|
return await delete_managed_template(db, user, template_id)
|