urmi-ai/backend/app/meta_service.py
2026-10-08 07:20:18 +05:30

1228 lines
50 KiB
Python

"""Meta Login and Marketing API access for the signed-in customer only.
User tokens stay on the server. Responses never include tokens, raw Graph
payloads, or unmasked internal identifiers except the opaque ids the console
needs to call a follow-up action.
"""
from __future__ import annotations
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
from urllib.parse import urlencode
import httpx
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from .auth import create_token, decode_token
from .config import settings
from .models import MetaAuditEvent, MetaConnection, UserAccount
from .whatsapp_onboarding import list_accounts
META_SCOPES = (
"public_profile",
"pages_show_list",
"pages_read_engagement",
"business_management",
"ads_read",
"ads_management",
)
SCOPE_LABELS = {
"public_profile": "Public profile",
"pages_show_list": "Page list",
"pages_read_engagement": "Page engagement",
"business_management": "Business portfolio",
"ads_read": "Ads reporting",
"ads_management": "Ads management",
}
PUBLISHABLE_OBJECTIVES = {"OUTCOME_TRAFFIC", "OUTCOME_ENGAGEMENT", "OUTCOME_AWARENESS"}
OBJECTIVE_LABELS = {
"OUTCOME_TRAFFIC": "Traffic",
"OUTCOME_ENGAGEMENT": "Engagement",
"OUTCOME_AWARENESS": "Awareness",
"OUTCOME_LEADS": "Leads",
"OUTCOME_SALES": "Sales",
"OUTCOME_APP_PROMOTION": "App promotion",
}
CTA_TYPES = {"LEARN_MORE", "SHOP_NOW", "SIGN_UP", "CONTACT_US", "WHATSAPP_MESSAGE", "SEND_MESSAGE"}
ACCOUNT_STATUS = {
1: "Active",
2: "Disabled",
3: "Unsettled",
7: "Pending review",
8: "Pending settlement",
9: "Grace period",
100: "Pending closure",
101: "Closed",
201: "Active",
202: "Closed",
}
ZERO_DECIMAL = {"BIF", "CLP", "DJF", "GNF", "JPY", "KMF", "KRW", "MGA", "PYG", "RWF", "VND", "VUV", "XAF", "XOF", "XPF"}
class MetaApiError(Exception):
def __init__(
self,
status_code: int,
code: str,
message: str,
permission: str | None = None,
graph_code: int | None = None,
) -> None:
self.status_code = status_code
self.code = code
self.message = message
self.permission = permission
self.graph_code = graph_code
super().__init__(message)
def http_meta(exc: MetaApiError) -> HTTPException:
detail: dict[str, Any] = {"code": exc.code, "message": exc.message}
if exc.permission:
detail["permission"] = exc.permission
if exc.graph_code is not None:
detail["graph_code"] = exc.graph_code
return HTTPException(status_code=exc.status_code, detail=detail)
def safe_text(value: Any, limit: int = 400) -> str:
text = str(value or "")
text = re.sub(r"(access_token|token|secret)=([^&\s]+)", r"\1=redacted", text, flags=re.I)
return text[:limit]
def mask_identifier(value: str) -> str:
raw = (value or "").strip()
if not raw:
return ""
prefix = "act_" if raw.startswith("act_") else ""
body = raw[4:] if prefix else raw
if len(body) <= 4:
return f"{prefix}••••"
return f"{prefix}••••{body[-4:]}"
def _same_account(left: Any, right: Any) -> bool:
return act_id(str(left or "")) == act_id(str(right or ""))
def act_id(value: str) -> str:
raw = (value or "").strip()
if not raw:
return ""
return raw if raw.startswith("act_") else f"act_{raw}"
def currency_offset(currency: str) -> int:
return 1 if (currency or "").upper() in ZERO_DECIMAL else 100
def minor_to_major(amount: Any, currency: str) -> float | None:
if amount is None or amount == "":
return None
try:
return float(amount) / currency_offset(currency)
except (TypeError, ValueError):
return None
def money(amount: float | None, currency: str) -> str | None:
if amount is None:
return None
code = (currency or "").upper()
if code == "INR":
return f"₹{amount:,.2f}"
if not code:
return f"{amount:,.2f}"
return f"{code} {amount:,.2f}"
def scopes_of(conn: MetaConnection) -> set[str]:
return {item.strip() for item in (conn.granted_scopes or "").split(",") if item.strip()}
def has_scope(conn: MetaConnection, scope: str) -> bool:
return scope in scopes_of(conn)
def capabilities(conn: MetaConnection | None) -> dict[str, bool]:
granted = scopes_of(conn) if conn and conn.status == "connected" else set()
return {
"profile": "public_profile" in granted or bool(conn and conn.meta_user_name and conn.status == "connected"),
"pages": "pages_show_list" in granted,
"page_engagement": "pages_read_engagement" in granted,
"business": "business_management" in granted,
"ads_read": "ads_read" in granted,
"ads_manage": "ads_management" in granted,
}
def load_connection(db: Session, user: UserAccount) -> MetaConnection | None:
return db.scalar(select(MetaConnection).where(MetaConnection.user_id == user.id))
def ensure_configured() -> None:
if not settings.meta_oauth_configured:
raise MetaApiError(
503,
"not_configured",
"Meta Login is not configured for this Urmi environment yet.",
)
def require_connection(db: Session, user: UserAccount) -> MetaConnection:
ensure_configured()
conn = load_connection(db, user)
if not conn or conn.status == "disconnected" or not conn.access_token:
raise MetaApiError(409, "not_connected", "Connect Meta before using this screen.")
expires = conn.token_expires_at
if expires and expires.tzinfo is None:
expires = expires.replace(tzinfo=timezone.utc)
if conn.status == "expired" or (expires and expires <= datetime.now(timezone.utc)):
conn.status = "expired"
conn.last_error = "The Meta connection has expired. Reconnect to continue."
db.commit()
raise MetaApiError(401, "token_expired", conn.last_error, graph_code=190)
return conn
def require_scope(conn: MetaConnection, scope: str) -> None:
if has_scope(conn, scope):
return
label = SCOPE_LABELS.get(scope, scope)
raise MetaApiError(
403,
"permission_denied",
f"Meta has not granted {label} for this connection.",
permission=scope,
)
def audit(
db: Session,
user: UserAccount,
action: str,
*,
asset_type: str = "",
asset_label: str = "",
asset_id: str = "",
detail: str = "",
status: str = "success",
error_message: str = "",
graph_code: int | None = None,
) -> None:
db.add(
MetaAuditEvent(
user_id=user.id,
action=action,
asset_type=asset_type,
asset_label=asset_label[:255],
asset_id=asset_id[:64],
detail=safe_text(detail, 1000),
status=status,
error_message=safe_text(error_message),
graph_code=graph_code,
)
)
db.commit()
def public_connection(conn: MetaConnection | None) -> dict[str, Any]:
if not conn or conn.status == "disconnected":
return {
"configured": settings.meta_oauth_configured,
"connected": False,
"status": "disconnected",
"user_name": "",
"connected_at": None,
"expires_at": None,
"scopes": [],
"capabilities": capabilities(None),
"selected_page_name": "",
"selected_ad_account_name": "",
"last_error": "",
}
return {
"configured": settings.meta_oauth_configured,
"connected": conn.status == "connected",
"status": conn.status,
"user_name": conn.meta_user_name,
"connected_at": conn.connected_at,
"expires_at": conn.token_expires_at,
"scopes": [
{"scope": scope, "label": SCOPE_LABELS.get(scope, scope)}
for scope in META_SCOPES
if scope in scopes_of(conn)
],
"capabilities": capabilities(conn),
"selected_page_name": conn.selected_page_name,
"selected_ad_account_name": conn.selected_ad_account_name,
"last_error": conn.last_error,
}
def authorization_url(user: UserAccount) -> str:
ensure_configured()
state = create_token(user.id, purpose="meta_oauth", minutes=10)
params = {
"client_id": settings.meta_app_id,
"redirect_uri": settings.meta_oauth_redirect_uri_resolved,
"state": state,
"response_type": "code",
"scope": ",".join(META_SCOPES),
}
if settings.meta_login_config_id:
params["config_id"] = settings.meta_login_config_id
version = settings.whatsapp_api_version or "v21.0"
return f"https://www.facebook.com/{version}/dialog/oauth?{urlencode(params)}"
async def finish_oauth(db: Session, code: str, state: str) -> None:
ensure_configured()
payload = decode_token(state, purpose="meta_oauth")
try:
user_id = uuid.UUID(payload["sub"])
except (KeyError, ValueError) as exc:
raise MetaApiError(401, "token_expired", "The Meta login attempt expired. Start it again.") from exc
user = db.scalar(select(UserAccount).where(UserAccount.id == user_id, UserAccount.status == "active"))
if not user:
raise MetaApiError(401, "token_expired", "The Meta login attempt expired. Start it again.")
short = await _exchange_code(code)
long_token = await _exchange_long_lived(short["access_token"])
token = long_token.get("access_token") or short["access_token"]
expires_in = long_token.get("expires_in") or short.get("expires_in")
profile = await graph("GET", "me", token, {"fields": "id,name"})
permission_rows = await graph("GET", "me/permissions", token)
granted = [
row.get("permission")
for row in permission_rows.get("data") or []
if isinstance(row, dict) and row.get("status") == "granted" and row.get("permission")
]
conn = load_connection(db, user) or MetaConnection(user_id=user.id)
conn.meta_user_id = str(profile.get("id") or "")
conn.meta_user_name = str(profile.get("name") or "")
conn.granted_scopes = ",".join(scope for scope in META_SCOPES if scope in set(granted))
conn.access_token = token
conn.status = "connected"
conn.last_error = ""
conn.last_graph_code = None
conn.connected_at = datetime.now(timezone.utc)
if expires_in:
conn.token_expires_at = datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))
else:
conn.token_expires_at = None
db.add(conn)
db.commit()
audit(db, user, "connection.authorized", asset_type="connection", asset_label=conn.meta_user_name, detail="Meta Login completed")
async def _exchange_code(code: str) -> dict[str, Any]:
return await graph(
"GET",
"oauth/access_token",
token=None,
params={
"client_id": settings.meta_app_id,
"client_secret": settings.meta_graph_app_secret,
"redirect_uri": settings.meta_oauth_redirect_uri_resolved,
"code": code,
},
)
async def _exchange_long_lived(token: str) -> dict[str, Any]:
try:
return await graph(
"GET",
"oauth/access_token",
token=None,
params={
"grant_type": "fb_exchange_token",
"client_id": settings.meta_app_id,
"client_secret": settings.meta_graph_app_secret,
"fb_exchange_token": token,
},
)
except MetaApiError:
return {}
async def graph(
method: str,
path: str,
token: str | None,
params: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
) -> dict[str, Any]:
version = settings.whatsapp_api_version or "v21.0"
url = f"https://graph.facebook.com/{version}/{path.lstrip('/')}"
headers = {"Authorization": f"Bearer {token}"} if token else {}
async with httpx.AsyncClient(timeout=25) as client:
response = await client.request(method, url, params=params, json=body, headers=headers)
payload: dict[str, Any] = {}
try:
parsed = response.json()
if isinstance(parsed, dict):
payload = parsed
except ValueError:
payload = {}
if response.status_code >= 400:
err = payload.get("error") if isinstance(payload.get("error"), dict) else {}
graph_code = err.get("code") if isinstance(err.get("code"), int) else None
message = safe_text(err.get("message") or "Meta could not complete this request.")
if graph_code == 190:
raise MetaApiError(401, "token_expired", "The Meta connection has expired. Reconnect to continue.", graph_code=190)
if graph_code in {10, 200, 294} or "permission" in message.lower() or "requires" in message.lower():
raise MetaApiError(
403,
"permission_denied",
"Meta did not allow this request with the permissions granted to Urmi.",
graph_code=graph_code,
)
raise MetaApiError(502, "graph_error", message, graph_code=graph_code)
return payload
def remember_graph_error(db: Session, conn: MetaConnection, exc: MetaApiError) -> None:
conn.last_error = exc.message
conn.last_graph_code = exc.graph_code
if exc.code == "token_expired":
conn.status = "expired"
db.commit()
async def graph_as_user(
db: Session,
conn: MetaConnection,
method: str,
path: str,
params: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
) -> dict[str, Any]:
try:
return await graph(method, path, conn.access_token, params, body)
except MetaApiError as exc:
remember_graph_error(db, conn, exc)
raise
def disconnect(db: Session, user: UserAccount) -> None:
conn = load_connection(db, user)
if not conn or conn.status == "disconnected":
raise MetaApiError(409, "not_connected", "Meta is not connected for this workspace.")
label = conn.meta_user_name
conn.access_token = ""
conn.status = "disconnected"
conn.selected_page_id = ""
conn.selected_page_name = ""
conn.selected_ad_account_id = ""
conn.selected_ad_account_name = ""
conn.last_error = ""
conn.last_graph_code = None
db.commit()
audit(db, user, "connection.disconnected", asset_type="connection", asset_label=label, detail="Customer disconnected Meta")
async def list_pages(db: Session, user: UserAccount) -> dict[str, Any]:
conn = require_connection(db, user)
require_scope(conn, "pages_show_list")
data = await graph_as_user(
db,
conn,
"GET",
"me/accounts",
{"fields": "id,name,username,link,fan_count,category,tasks", "limit": "50"},
)
pages = []
for row in data.get("data") or []:
if not isinstance(row, dict) or not row.get("id"):
continue
pages.append(_page_summary(row, selected=row.get("id") == conn.selected_page_id))
return {"pages": pages, "selected_name": conn.selected_page_name}
def _page_summary(row: dict[str, Any], selected: bool) -> dict[str, Any]:
tasks = row.get("tasks") if isinstance(row.get("tasks"), list) else []
return {
"id": str(row.get("id")),
"name": row.get("name") or "Untitled Page",
"username": row.get("username") or "",
"category": row.get("category") or "",
"link": row.get("link") or "",
"fan_count": row.get("fan_count") if isinstance(row.get("fan_count"), int) else None,
"tasks": [str(task) for task in tasks],
"selected": selected,
}
async def page_detail(db: Session, user: UserAccount, page_id: str) -> dict[str, Any]:
conn = require_connection(db, user)
require_scope(conn, "pages_show_list")
page, page_token = await _authorized_page(db, conn, page_id)
posts: list[dict[str, Any]] = []
engagement_error = None
if has_scope(conn, "pages_read_engagement"):
try:
payload = await graph(
"GET",
f"{page_id}/posts",
page_token or conn.access_token,
{
"fields": "message,created_time,permalink_url,full_picture,likes.summary(true),comments.summary(true)",
"limit": "15",
},
)
for row in payload.get("data") or []:
if isinstance(row, dict):
posts.append(_post_summary(row))
except MetaApiError as exc:
remember_graph_error(db, conn, exc)
engagement_error = {"code": exc.code, "message": exc.message, "permission": exc.permission}
else:
engagement_error = {
"code": "permission_denied",
"message": "Meta has not granted Page engagement for this connection.",
"permission": "pages_read_engagement",
}
return {
"page": page,
"posts": posts,
"engagement": {
"available": engagement_error is None,
"error": engagement_error,
"note": "Page insight charts need a separate Page insights permission, which this connection does not include.",
},
}
def _post_summary(row: dict[str, Any]) -> dict[str, Any]:
likes = row.get("likes") if isinstance(row.get("likes"), dict) else {}
comments = row.get("comments") if isinstance(row.get("comments"), dict) else {}
like_summary = likes.get("summary") if isinstance(likes.get("summary"), dict) else {}
comment_summary = comments.get("summary") if isinstance(comments.get("summary"), dict) else {}
return {
"message": row.get("message") or "",
"created_time": row.get("created_time"),
"permalink_url": row.get("permalink_url") or "",
"picture_url": row.get("full_picture") or "",
"reactions": like_summary.get("total_count") if isinstance(like_summary.get("total_count"), int) else 0,
"comments": comment_summary.get("total_count") if isinstance(comment_summary.get("total_count"), int) else 0,
}
async def _authorized_page(db: Session, conn: MetaConnection, page_id: str) -> tuple[dict[str, Any], str]:
data = await graph_as_user(
db,
conn,
"GET",
"me/accounts",
{"fields": "id,name,username,link,fan_count,category,tasks,access_token", "limit": "100"},
)
for row in data.get("data") or []:
if isinstance(row, dict) and str(row.get("id")) == page_id:
token = str(row.get("access_token") or "")
summary = _page_summary(row, selected=page_id == conn.selected_page_id)
return summary, token
raise MetaApiError(404, "not_found", "That Page is not available to this Meta connection.")
async def select_page(db: Session, user: UserAccount, page_id: str) -> dict[str, Any]:
conn = require_connection(db, user)
require_scope(conn, "pages_show_list")
page, _token = await _authorized_page(db, conn, page_id)
conn.selected_page_id = page_id
conn.selected_page_name = str(page["name"])[:255]
db.commit()
audit(db, user, "page.selected", asset_type="page", asset_label=conn.selected_page_name, asset_id=page_id)
return {"name": conn.selected_page_name}
async def list_businesses(db: Session, user: UserAccount) -> dict[str, Any]:
conn = require_connection(db, user)
require_scope(conn, "business_management")
data = await graph_as_user(db, conn, "GET", "me/businesses", {"fields": "id,name", "limit": "25"})
businesses = []
for row in data.get("data") or []:
if isinstance(row, dict) and row.get("name"):
businesses.append({"name": row.get("name")})
return {"businesses": businesses}
async def list_ad_accounts(db: Session, user: UserAccount) -> dict[str, Any]:
conn = require_connection(db, user)
if not (has_scope(conn, "ads_read") or has_scope(conn, "ads_management") or has_scope(conn, "business_management")):
raise MetaApiError(
403,
"permission_denied",
"Meta has not granted Ads reporting or Business portfolio for this connection.",
permission="ads_read",
)
rows: dict[str, dict[str, Any]] = {}
if has_scope(conn, "ads_read"):
data = await graph_as_user(
db,
conn,
"GET",
"me/adaccounts",
{"fields": "id,name,account_status,currency,amount_spent", "limit": "50"},
)
for row in data.get("data") or []:
if isinstance(row, dict) and row.get("id"):
rows[act_id(str(row["id"]))] = row
if has_scope(conn, "business_management"):
businesses = await graph_as_user(db, conn, "GET", "me/businesses", {"fields": "id,name", "limit": "25"})
for business in businesses.get("data") or []:
if not isinstance(business, dict) or not business.get("id"):
continue
business_name = str(business.get("name") or "")
for edge in ("owned_ad_accounts", "client_ad_accounts"):
try:
owned = await graph_as_user(
db,
conn,
"GET",
f"{business['id']}/{edge}",
{"fields": "id,name,account_status,currency,amount_spent", "limit": "50"},
)
except MetaApiError as exc:
if exc.code == "permission_denied":
continue
raise
for row in owned.get("data") or []:
if isinstance(row, dict) and row.get("id"):
current = rows.setdefault(act_id(str(row["id"])), row)
current["business_name"] = business_name
accounts = [
_account_summary(row, selected=act_id(str(row.get("id"))) == act_id(conn.selected_ad_account_id))
for row in rows.values()
]
accounts.sort(key=lambda item: item["name"].lower())
return {"accounts": accounts, "selected_name": conn.selected_ad_account_name}
def _account_summary(row: dict[str, Any], selected: bool) -> dict[str, Any]:
currency = str(row.get("currency") or "")
spent = minor_to_major(row.get("amount_spent"), currency)
status_code = row.get("account_status")
return {
"id": act_id(str(row.get("id"))),
"name": row.get("name") or "Ad account",
"business_name": row.get("business_name") or "",
"currency": currency,
"status": ACCOUNT_STATUS.get(status_code, "Unknown") if isinstance(status_code, int) else "Unknown",
"amount_spent": money(spent, currency),
"selected": selected,
"ref": mask_identifier(act_id(str(row.get("id")))),
}
async def select_ad_account(db: Session, user: UserAccount, account_id: str) -> dict[str, Any]:
conn = require_connection(db, user)
if not (has_scope(conn, "ads_read") or has_scope(conn, "ads_management") or has_scope(conn, "business_management")):
raise MetaApiError(403, "permission_denied", "Meta has not granted Ads reporting for this connection.", permission="ads_read")
accounts = await list_ad_accounts(db, user)
match = next((row for row in accounts["accounts"] if row["id"] == act_id(account_id)), None)
if not match:
raise MetaApiError(404, "not_found", "That ad account is not available to this Meta connection.")
conn.selected_ad_account_id = match["id"]
conn.selected_ad_account_name = str(match["name"])[:255]
db.commit()
audit(db, user, "ad_account.selected", asset_type="ad_account", asset_label=conn.selected_ad_account_name, asset_id=conn.selected_ad_account_id)
return {"name": conn.selected_ad_account_name}
async def _selected_account(db: Session, user: UserAccount, manage: bool) -> tuple[MetaConnection, str]:
conn = require_connection(db, user)
if manage:
require_scope(conn, "ads_management")
elif not (has_scope(conn, "ads_read") or has_scope(conn, "ads_management")):
require_scope(conn, "ads_read")
if not conn.selected_ad_account_id:
raise MetaApiError(409, "no_ad_account", "Select an ad account before opening campaigns.")
accounts = await list_ad_accounts(db, user)
match = next((row for row in accounts["accounts"] if row["id"] == act_id(conn.selected_ad_account_id)), None)
if not match:
raise MetaApiError(409, "no_ad_account", "The selected ad account is no longer available to this connection.")
return conn, match["id"]
async def list_campaigns(db: Session, user: UserAccount) -> dict[str, Any]:
conn, account_id = await _selected_account(db, user, manage=False)
data = await graph_as_user(
db,
conn,
"GET",
f"{account_id}/campaigns",
{
"fields": "id,name,status,effective_status,objective,created_time",
"limit": "50",
},
)
insights: dict[str, dict[str, Any]] = {}
insights_error = None
try:
report = await graph_as_user(
db,
conn,
"GET",
f"{account_id}/insights",
{
"level": "campaign",
"fields": "campaign_id,spend,impressions,clicks,ctr,reach",
"date_preset": "last_30d",
"limit": "50",
},
)
for row in report.get("data") or []:
if isinstance(row, dict) and row.get("campaign_id"):
insights[str(row["campaign_id"])] = row
except MetaApiError as exc:
insights_error = {"code": exc.code, "message": exc.message}
currency = await _account_currency(db, conn, account_id)
campaigns = []
for row in data.get("data") or []:
if not isinstance(row, dict) or not row.get("id"):
continue
metric = insights.get(str(row["id"]), {})
spend = _float(metric.get("spend"))
campaigns.append(
{
"id": str(row["id"]),
"name": row.get("name") or "Campaign",
"status": row.get("effective_status") or row.get("status") or "",
"objective": OBJECTIVE_LABELS.get(str(row.get("objective")), str(row.get("objective") or "").replace("OUTCOME_", "").title()),
"created_time": row.get("created_time"),
"spend": money(spend, currency) if metric else None,
"impressions": _int(metric.get("impressions")) if metric else None,
"clicks": _int(metric.get("clicks")) if metric else None,
"ctr": _float(metric.get("ctr")) if metric else None,
"reach": _int(metric.get("reach")) if metric else None,
}
)
return {
"account_name": conn.selected_ad_account_name,
"currency": currency,
"campaigns": campaigns,
"can_manage": has_scope(conn, "ads_management"),
"insights_error": insights_error,
}
async def _account_currency(db: Session, conn: MetaConnection, account_id: str) -> str:
try:
data = await graph_as_user(db, conn, "GET", account_id, {"fields": "currency"})
return str(data.get("currency") or "")
except MetaApiError:
return ""
def _float(value: Any) -> float | None:
try:
return float(value)
except (TypeError, ValueError):
return None
def _int(value: Any) -> int | None:
try:
return int(float(value))
except (TypeError, ValueError):
return None
async def set_campaign_status(db: Session, user: UserAccount, campaign_id: str, status: str) -> dict[str, Any]:
if status not in {"ACTIVE", "PAUSED"}:
raise MetaApiError(422, "validation", "Choose pause or resume.")
conn, account_id = await _selected_account(db, user, manage=True)
current = await graph_as_user(db, conn, "GET", campaign_id, {"fields": "id,name,account_id"})
if not _same_account(current.get("account_id"), account_id):
raise MetaApiError(404, "not_found", "That campaign is not in the selected ad account.")
name = str(current.get("name") or "Campaign")
action = "campaign.resume" if status == "ACTIVE" else "campaign.pause"
try:
await graph_as_user(db, conn, "POST", campaign_id, body={"status": status})
except MetaApiError as exc:
audit(db, user, action, asset_type="campaign", asset_label=name, asset_id=campaign_id, status="failed", error_message=exc.message, graph_code=exc.graph_code)
raise
audit(db, user, action, asset_type="campaign", asset_label=name, asset_id=campaign_id, detail=f"Status set to {status}")
return {"name": name, "status": status}
async def update_campaign(db: Session, user: UserAccount, campaign_id: str, name: str | None, daily_budget: float | None) -> dict[str, Any]:
conn, account_id = await _selected_account(db, user, manage=True)
current = await graph_as_user(db, conn, "GET", campaign_id, {"fields": "id,name,account_id"})
if not _same_account(current.get("account_id"), account_id):
raise MetaApiError(404, "not_found", "That campaign is not in the selected ad account.")
if not name and daily_budget is None:
raise MetaApiError(422, "validation", "Add a name or daily budget to update.")
label = name or str(current.get("name") or "Campaign")
try:
if name:
await graph_as_user(db, conn, "POST", campaign_id, body={"name": name[:120]})
if daily_budget is not None:
currency = await _account_currency(db, conn, account_id)
adsets = await graph_as_user(db, conn, "GET", f"{campaign_id}/adsets", {"fields": "id,name", "limit": "5"})
rows = [row for row in adsets.get("data") or [] if isinstance(row, dict) and row.get("id")]
if len(rows) != 1:
raise MetaApiError(422, "feature_unavailable", "Budget can be edited here only when the campaign has one ad set.")
minor = str(int(round(daily_budget * currency_offset(currency))))
await graph_as_user(db, conn, "POST", str(rows[0]["id"]), body={"daily_budget": minor})
except MetaApiError as exc:
audit(db, user, "campaign.edit", asset_type="campaign", asset_label=label, asset_id=campaign_id, status="failed", error_message=exc.message, graph_code=exc.graph_code)
raise
audit(db, user, "campaign.edit", asset_type="campaign", asset_label=label, asset_id=campaign_id, detail="Campaign details updated")
return {"name": label}
async def publish_campaign(db: Session, user: UserAccount, payload: dict[str, Any]) -> dict[str, Any]:
conn, account_id = await _selected_account(db, user, manage=True)
objective = str(payload.get("objective") or "")
destination = str(payload.get("destination") or "")
if objective not in PUBLISHABLE_OBJECTIVES:
raise MetaApiError(
422,
"feature_unavailable",
"Leads, sales, and app promotion stay unavailable until this connection includes the assets those objectives require.",
)
if destination not in {"website", "whatsapp", "page"}:
raise MetaApiError(422, "validation", "Choose a destination Urmi can send to Meta.")
if destination == "whatsapp" and objective == "OUTCOME_AWARENESS":
raise MetaApiError(422, "feature_unavailable", "Awareness cannot be published to WhatsApp from this connection.")
name = str(payload.get("name") or "").strip()[:120]
message = str(payload.get("message") or "").strip()
headline = str(payload.get("headline") or "").strip()[:80]
link = str(payload.get("link") or "").strip()
cta = str(payload.get("call_to_action") or "LEARN_MORE")
page_id = str(payload.get("page_id") or "")
countries = [str(code).upper() for code in payload.get("countries") or [] if re.fullmatch(r"[A-Za-z]{2}", str(code))]
age_min = int(payload.get("age_min") or 18)
age_max = int(payload.get("age_max") or 65)
daily_budget = float(payload.get("daily_budget") or 0)
activate = bool(payload.get("activate"))
if not name or not message or not page_id or not countries or daily_budget <= 0 or age_min < 18 or age_max < age_min or age_max > 65:
raise MetaApiError(422, "validation", "Add a name, Page, message, country, and daily budget before publishing.")
if cta not in CTA_TYPES:
raise MetaApiError(422, "validation", "Choose a supported call to action.")
if destination == "website" and not link.startswith(("http://", "https://")):
raise MetaApiError(422, "validation", "Website destination needs an http or https link.")
page, _page_token = await _authorized_page(db, conn, page_id)
page_link = str(page.get("link") or "")
if destination != "whatsapp" and not (link or page_link):
raise MetaApiError(422, "validation", "Meta did not return a link for this Page. Add a website link or choose a Page that has one.")
currency = await _account_currency(db, conn, account_id)
status = "ACTIVE" if activate else "PAUSED"
created_campaign = ""
try:
campaign = await graph_as_user(
db,
conn,
"POST",
f"{account_id}/campaigns",
body={
"name": name,
"objective": objective,
"status": "PAUSED",
"special_ad_categories": [],
"is_adset_budget_sharing_enabled": False,
},
)
created_campaign = str(campaign.get("id") or "")
optimization = "REACH" if objective == "OUTCOME_AWARENESS" else "POST_ENGAGEMENT" if objective == "OUTCOME_ENGAGEMENT" and destination != "whatsapp" else "CONVERSATIONS" if destination == "whatsapp" else "LINK_CLICKS"
adset_body: dict[str, Any] = {
"name": f"{name} ad set",
"campaign_id": created_campaign,
"daily_budget": str(int(round(daily_budget * currency_offset(currency)))),
"billing_event": "IMPRESSIONS",
"optimization_goal": optimization,
"bid_strategy": "LOWEST_COST_WITHOUT_CAP",
"status": "PAUSED",
"promoted_object": {"page_id": page_id},
"targeting": {
"geo_locations": {"countries": countries[:5]},
"age_min": age_min,
"age_max": age_max,
"targeting_automation": {"advantage_audience": 0},
},
}
if destination == "whatsapp":
adset_body["destination_type"] = "WHATSAPP"
elif destination == "website":
adset_body["destination_type"] = "WEBSITE"
adset = await graph_as_user(db, conn, "POST", f"{account_id}/adsets", body=adset_body)
story = _story_spec(page_id, destination, message, headline, link, cta, page_link)
creative = await graph_as_user(
db,
conn,
"POST",
f"{account_id}/adcreatives",
body={"name": f"{name} creative", "object_story_spec": story},
)
await graph_as_user(
db,
conn,
"POST",
f"{account_id}/ads",
body={
"name": f"{name} ad",
"adset_id": adset.get("id"),
"creative": {"creative_id": creative.get("id")},
"status": "PAUSED",
},
)
if activate and created_campaign:
await graph_as_user(db, conn, "POST", created_campaign, body={"status": "ACTIVE"})
await graph_as_user(db, conn, "POST", str(adset.get("id")), body={"status": "ACTIVE"})
except MetaApiError as exc:
detail = exc.message
if created_campaign:
detail = f"{exc.message} A paused campaign named {name} may already exist in the ad account."
audit(
db,
user,
"campaign.publish",
asset_type="campaign",
asset_label=name,
asset_id=created_campaign,
status="failed",
error_message=detail,
graph_code=exc.graph_code,
)
raise MetaApiError(exc.status_code, exc.code, detail, exc.permission, exc.graph_code) from exc
audit(
db,
user,
"campaign.publish",
asset_type="campaign",
asset_label=name,
asset_id=created_campaign,
detail="Published active" if activate else "Created paused",
)
return {"name": name, "status": status}
def _story_spec(page_id: str, destination: str, message: str, headline: str, link: str, cta: str, page_link: str) -> dict[str, Any]:
target = link or page_link
if destination == "whatsapp":
link_data: dict[str, Any] = {
"message": message,
"link": "https://api.whatsapp.com/send",
"call_to_action": {"type": "WHATSAPP_MESSAGE"},
}
else:
link_data = {
"message": message,
"link": target,
"call_to_action": {"type": cta, "value": {"link": target}},
}
if headline:
link_data["name"] = headline
return {"page_id": page_id, "link_data": link_data}
async def list_ads(db: Session, user: UserAccount) -> dict[str, Any]:
conn, account_id = await _selected_account(db, user, manage=False)
data = await graph_as_user(
db,
conn,
"GET",
f"{account_id}/ads",
{
"fields": "id,name,status,effective_status,creative{title,body,image_url,thumbnail_url,call_to_action_type,object_story_spec},campaign{name}",
"limit": "50",
},
)
ads = []
for row in data.get("data") or []:
if isinstance(row, dict) and row.get("id"):
ads.append(_ad_summary(row))
return {"account_name": conn.selected_ad_account_name, "ads": ads, "can_manage": has_scope(conn, "ads_management")}
def _ad_summary(row: dict[str, Any]) -> dict[str, Any]:
creative = row.get("creative") if isinstance(row.get("creative"), dict) else {}
story = creative.get("object_story_spec") if isinstance(creative.get("object_story_spec"), dict) else {}
link_data = story.get("link_data") if isinstance(story.get("link_data"), dict) else {}
video = story.get("video_data") if isinstance(story.get("video_data"), dict) else {}
cta = link_data.get("call_to_action") if isinstance(link_data.get("call_to_action"), dict) else {}
campaign = row.get("campaign") if isinstance(row.get("campaign"), dict) else {}
return {
"id": str(row.get("id")),
"name": row.get("name") or "Ad",
"campaign_name": campaign.get("name") or "",
"status": row.get("effective_status") or row.get("status") or "",
"headline": link_data.get("name") or video.get("title") or creative.get("title") or "",
"message": link_data.get("message") or video.get("message") or creative.get("body") or "",
"call_to_action": cta.get("type") or creative.get("call_to_action_type") or "",
"image_url": creative.get("image_url") or creative.get("thumbnail_url") or link_data.get("picture") or "",
"link": link_data.get("link") or "",
"editable": bool(story.get("page_id")),
}
async def update_ad(db: Session, user: UserAccount, ad_id: str, payload: dict[str, Any]) -> dict[str, Any]:
conn, account_id = await _selected_account(db, user, manage=True)
current = await graph_as_user(
db,
conn,
"GET",
ad_id,
{"fields": "id,name,account_id,creative{object_story_spec}"},
)
if not _same_account(current.get("account_id"), account_id):
raise MetaApiError(404, "not_found", "That ad is not in the selected ad account.")
label = str(payload.get("name") or current.get("name") or "Ad")
try:
if payload.get("name"):
await graph_as_user(db, conn, "POST", ad_id, body={"name": str(payload["name"])[:120]})
creative_fields = ("message", "headline", "call_to_action", "link")
if any(payload.get(field) for field in creative_fields):
creative = current.get("creative") if isinstance(current.get("creative"), dict) else {}
story = creative.get("object_story_spec") if isinstance(creative.get("object_story_spec"), dict) else {}
page_id = str(story.get("page_id") or "")
if not page_id:
raise MetaApiError(422, "feature_unavailable", "This ad's creative is not a Page link ad Urmi can edit.")
await _authorized_page(db, conn, page_id)
cta = str(payload.get("call_to_action") or "LEARN_MORE")
if cta not in CTA_TYPES:
raise MetaApiError(422, "validation", "Choose a supported call to action.")
link = str(payload.get("link") or "")
if link and not link.startswith(("http://", "https://")):
raise MetaApiError(422, "validation", "Creative links need to start with http or https.")
replacement = await graph_as_user(
db,
conn,
"POST",
f"{account_id}/adcreatives",
body={
"name": f"{label} creative",
"object_story_spec": _story_spec(
page_id,
"whatsapp" if cta == "WHATSAPP_MESSAGE" else "website",
str(payload.get("message") or ""),
str(payload.get("headline") or ""),
link,
cta,
"",
),
},
)
await graph_as_user(db, conn, "POST", ad_id, body={"creative": {"creative_id": replacement.get("id")}})
except MetaApiError as exc:
audit(db, user, "ad.edit", asset_type="ad", asset_label=label, asset_id=ad_id, status="failed", error_message=exc.message, graph_code=exc.graph_code)
raise
audit(db, user, "ad.edit", asset_type="ad", asset_label=label, asset_id=ad_id, detail="Ad details updated")
return {"name": label}
async def set_ad_status(db: Session, user: UserAccount, ad_id: str, status: str) -> dict[str, Any]:
if status not in {"ACTIVE", "PAUSED"}:
raise MetaApiError(422, "validation", "Choose pause or resume.")
conn, account_id = await _selected_account(db, user, manage=True)
current = await graph_as_user(db, conn, "GET", ad_id, {"fields": "id,name,account_id"})
if not _same_account(current.get("account_id"), account_id):
raise MetaApiError(404, "not_found", "That ad is not in the selected ad account.")
name = str(current.get("name") or "Ad")
action = "ad.resume" if status == "ACTIVE" else "ad.pause"
try:
await graph_as_user(db, conn, "POST", ad_id, body={"status": status})
except MetaApiError as exc:
audit(db, user, action, asset_type="ad", asset_label=name, asset_id=ad_id, status="failed", error_message=exc.message, graph_code=exc.graph_code)
raise
audit(db, user, action, asset_type="ad", asset_label=name, asset_id=ad_id, detail=f"Status set to {status}")
return {"name": name, "status": status}
async def analytics(db: Session, user: UserAccount, date_preset: str) -> dict[str, Any]:
if date_preset not in {"last_7d", "last_14d", "last_30d", "last_90d", "maximum"}:
raise MetaApiError(422, "validation", "Choose a supported date range.")
conn, account_id = await _selected_account(db, user, manage=False)
currency = await _account_currency(db, conn, account_id)
summary = await graph_as_user(
db,
conn,
"GET",
f"{account_id}/insights",
{"fields": "spend,impressions,reach,clicks,ctr,cpc,cpm", "date_preset": date_preset},
)
daily = await graph_as_user(
db,
conn,
"GET",
f"{account_id}/insights",
{
"fields": "spend,impressions,clicks,date_start",
"date_preset": date_preset,
"time_increment": "1",
"limit": "100",
},
)
row = (summary.get("data") or [{}])[0] if isinstance(summary.get("data"), list) else {}
if not isinstance(row, dict):
row = {}
series = []
for item in daily.get("data") or []:
if not isinstance(item, dict):
continue
series.append(
{
"date": item.get("date_start"),
"spend": _float(item.get("spend")) or 0,
"impressions": _int(item.get("impressions")) or 0,
"clicks": _int(item.get("clicks")) or 0,
}
)
spend = _float(row.get("spend"))
return {
"account_name": conn.selected_ad_account_name,
"currency": currency,
"date_preset": date_preset,
"has_delivery": bool(summary.get("data")),
"totals": {
"spend": money(spend, currency) if row else None,
"impressions": _int(row.get("impressions")),
"reach": _int(row.get("reach")),
"clicks": _int(row.get("clicks")),
"ctr": _float(row.get("ctr")),
"cpc": money(_float(row.get("cpc")), currency) if row.get("cpc") is not None else None,
"cpm": money(_float(row.get("cpm")), currency) if row.get("cpm") is not None else None,
},
"series": series,
}
async def overview(db: Session, user: UserAccount) -> dict[str, Any]:
conn = load_connection(db, user)
whatsapp = []
for account in list_accounts(db, user):
phone = account.get("primary_phone") if isinstance(account.get("primary_phone"), dict) else {}
whatsapp.append(
{
"name": account.get("name") or "WhatsApp Business",
"phone": phone.get("display_phone") or "",
"status": account.get("status") or "",
"webhooks": account.get("webhooks_status") or "",
"templates": account.get("template_count") or 0,
}
)
result: dict[str, Any] = {
"connection": public_connection(conn if conn and conn.status != "disconnected" else None),
"whatsapp": whatsapp,
"businesses": _section_empty(),
"pages": _section_empty(),
"ad_accounts": _section_empty(),
"performance": _section_empty(),
"activity": recent_audit(db, user, limit=6),
}
if not settings.meta_oauth_configured:
result["connection"]["configured"] = False
return result
if not conn or conn.status != "connected":
return result
try:
require_connection(db, user)
except MetaApiError as exc:
result["connection"] = public_connection(conn)
result["connection"]["status"] = "expired" if exc.code == "token_expired" else conn.status
result["connection"]["last_error"] = exc.message
return result
result["businesses"] = await _section(lambda: list_businesses(db, user), "businesses")
result["pages"] = await _section(lambda: list_pages(db, user), "pages")
if has_scope(conn, "ads_read") or has_scope(conn, "business_management"):
result["ad_accounts"] = await _section(lambda: list_ad_accounts(db, user), "accounts")
else:
result["ad_accounts"] = {
"state": "permission_denied",
"message": "Meta has not granted Ads reporting for this connection.",
"permission": "ads_read",
"items": [],
}
if has_scope(conn, "ads_read") and conn.selected_ad_account_id:
try:
report = await analytics(db, user, "last_30d")
result["performance"] = {"state": "ok", "message": "", "report": report}
except MetaApiError as exc:
result["performance"] = {"state": exc.code, "message": exc.message, "permission": exc.permission, "report": None}
elif has_scope(conn, "ads_read"):
result["performance"] = {"state": "no_ad_account", "message": "Select an ad account to load performance.", "report": None}
else:
result["performance"] = {
"state": "permission_denied",
"message": "Meta has not granted Ads reporting for this connection.",
"permission": "ads_read",
"report": None,
}
result["connection"] = public_connection(load_connection(db, user))
result["activity"] = recent_audit(db, user, limit=6)
return result
def _section_empty() -> dict[str, Any]:
return {"state": "not_connected", "message": "Connect Meta to load this.", "items": []}
async def _section(loader, key: str) -> dict[str, Any]:
try:
payload = await loader()
items = payload.get(key) or []
return {"state": "ok" if items else "empty", "message": "", "items": items}
except MetaApiError as exc:
return {"state": exc.code, "message": exc.message, "permission": exc.permission, "items": []}
def recent_audit(db: Session, user: UserAccount, limit: int = 20) -> list[dict[str, Any]]:
rows = db.scalars(
select(MetaAuditEvent)
.where(MetaAuditEvent.user_id == user.id)
.order_by(MetaAuditEvent.created_at.desc())
.limit(limit)
).all()
return [
{
"action": row.action,
"asset_type": row.asset_type,
"asset_label": row.asset_label,
"detail": row.detail,
"status": row.status,
"error_message": row.error_message,
"created_at": row.created_at,
}
for row in rows
]
def settings_view(db: Session, user: UserAccount) -> dict[str, Any]:
conn = load_connection(db, user)
whatsapp = []
for account in list_accounts(db, user):
whatsapp.append(
{
"name": account.get("name") or "WhatsApp Business",
"status": account.get("status") or "",
"webhooks": account.get("webhooks_status") or "unavailable",
}
)
debug = {
"app_configured": settings.meta_oauth_configured,
"redirect_configured": bool(settings.meta_oauth_redirect_uri_resolved),
"login_config_set": bool(settings.meta_login_config_id),
"granted_scopes": sorted(scopes_of(conn)) if conn else [],
"last_error": conn.last_error if conn else "",
"last_graph_code": conn.last_graph_code if conn else None,
"meta_user_ref": mask_identifier(conn.meta_user_id) if conn else "",
"page_ref": mask_identifier(conn.selected_page_id) if conn else "",
"ad_account_ref": mask_identifier(conn.selected_ad_account_id) if conn else "",
}
return {
"connection": public_connection(conn if conn and conn.status != "disconnected" else None),
"whatsapp": whatsapp,
"ads_webhooks": "not_subscribed",
"audit": recent_audit(db, user, limit=30),
"debug": debug,
}