Whatsapp

backend/app/routes_meta.py

@@ -13,11 +13,14 @@ from .meta_service import (
     analytics,
     authorization_url,
     disconnect,
+    connect_with_user_token,
     finish_oauth,
     http_meta,
     list_ad_accounts,
     list_ads,
+    list_adsets,
     list_businesses,
+    campaign_detail,
     list_campaigns,
     list_pages,
     overview,
@@ -43,6 +46,10 @@ class Confirm(BaseModel):
     confirm: bool = False
 
 
+class ManualToken(BaseModel):
+    access_token: str = Field(min_length=1, max_length=4096)
+
+
 class IdBody(Confirm):
     id: str = Field(min_length=1, max_length=64)
 
@@ -118,6 +125,14 @@ async def oauth_callback(code: str = "", state: str = "", error: str = "", db: S
     return _redirect("connected=1")
 
 
+@router.post("/token")
+async def save_token(body: ManualToken, user: UserAccount = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+    try:
+        return await connect_with_user_token(db, user, body.access_token)
+    except MetaApiError as exc:
+        raise http_meta(exc) from exc
+
+
 @router.post("/disconnect")
 def disconnect_meta(body: Confirm, user: UserAccount = Depends(current_user), db: Session = Depends(get_db)) -> dict:
     _confirmed(body)
@@ -191,6 +206,26 @@ async def campaigns(user: UserAccount = Depends(current_user), db: Session = Dep
         raise http_meta(exc) from exc
 
 
+@router.get("/campaigns/{campaign_id}")
+async def campaign(campaign_id: str, user: UserAccount = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+    try:
+        return await campaign_detail(db, user, campaign_id)
+    except MetaApiError as exc:
+        raise http_meta(exc) from exc
+
+
+@router.get("/adsets")
+async def adsets(
+    campaign_id: str | None = None,
+    user: UserAccount = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    try:
+        return await list_adsets(db, user, campaign_id or None)
+    except MetaApiError as exc:
+        raise http_meta(exc) from exc
+
+
 @router.post("/campaigns")
 async def create_campaign(body: CampaignPublish, user: UserAccount = Depends(current_user), db: Session = Depends(get_db)) -> dict:
     _confirmed(body)