1511 lines
60 KiB
Python
1511 lines
60 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 json
|
|
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 _equal_filter(field: str, value: str) -> str:
|
|
return json.dumps([{"field": field, "operator": "EQUAL", "value": value}])
|
|
|
|
|
|
def _in_filter(field: str, values: list[str]) -> str:
|
|
return json.dumps([{"field": field, "operator": "IN", "value": values}])
|
|
|
|
|
|
def _is_ads_api_path(path: str) -> bool:
|
|
"""ads_read / ads_management live on Marketing API paths, not Page/Business Graph."""
|
|
cleaned = path.lstrip("/")
|
|
if cleaned == "me/adaccounts" or cleaned.startswith("me/adaccounts?"):
|
|
return True
|
|
if cleaned.startswith("act_"):
|
|
return True
|
|
return bool(re.fullmatch(r"\d+(?:/[A-Za-z_]+)?", cleaned))
|
|
|
|
|
|
async def ads_api(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
method: str,
|
|
path: str,
|
|
params: dict[str, Any] | None = None,
|
|
body: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Marketing API for ads_read and ads_management (campaigns, ad sets, ads, insights)."""
|
|
if not _is_ads_api_path(path):
|
|
raise MetaApiError(500, "graph_error", "Ads operations must use the Marketing API on an ad account.")
|
|
return await graph_as_user(db, conn, method, path.lstrip("/"), params, body)
|
|
|
|
|
|
async def marketing_get(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
account_id: str,
|
|
edge: str,
|
|
params: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""GET /act_{id}/{edge} — Ads Insights / Ads Management read."""
|
|
return await ads_api(db, conn, "GET", f"{act_id(account_id)}/{edge.lstrip('/')}", params)
|
|
|
|
|
|
async def marketing_post(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
account_id: str,
|
|
edge: str,
|
|
body: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""POST /act_{id}/{edge} — Ads Management write."""
|
|
return await ads_api(db, conn, "POST", f"{act_id(account_id)}/{edge.lstrip('/')}", body=body)
|
|
|
|
|
|
async def ads_object_post(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
object_id: str,
|
|
body: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""POST /{campaign|adset|ad id} status and field updates (Marketing API object)."""
|
|
if not re.fullmatch(r"\d+", str(object_id)):
|
|
raise MetaApiError(422, "validation", "That Ads object is not valid for this ad account.")
|
|
return await ads_api(db, conn, "POST", str(object_id), body=body)
|
|
|
|
|
|
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:
|
|
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]:
|
|
redirect_uri = settings.meta_oauth_redirect_uri_resolved
|
|
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": "",
|
|
"oauth_redirect_uri": redirect_uri,
|
|
}
|
|
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,
|
|
"oauth_redirect_uri": redirect_uri,
|
|
}
|
|
|
|
|
|
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 connect_with_user_token(db: Session, user: UserAccount, access_token: str) -> dict[str, Any]:
|
|
"""Store a customer-supplied user token without the Facebook login redirect."""
|
|
token = "".join(str(access_token or "").split())
|
|
if token.lower().startswith("bearer"):
|
|
token = token[6:]
|
|
if len(token) < 20:
|
|
raise MetaApiError(400, "validation", "Paste the access token from Meta.")
|
|
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")
|
|
]
|
|
scopes = [scope for scope in META_SCOPES if scope in set(granted)]
|
|
if not any(scope != "public_profile" for scope in scopes):
|
|
raise MetaApiError(
|
|
403,
|
|
"permission_denied",
|
|
"This token cannot see Pages or ad accounts. Use a token that includes those permissions.",
|
|
)
|
|
stored = token
|
|
expires_at = None
|
|
if settings.meta_app_id and settings.meta_graph_app_secret:
|
|
long_token = await _exchange_long_lived(token)
|
|
if long_token.get("access_token"):
|
|
stored = str(long_token["access_token"])
|
|
expires_in = long_token.get("expires_in")
|
|
if expires_in:
|
|
expires_at = datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))
|
|
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(scopes)
|
|
conn.access_token = stored
|
|
conn.status = "connected"
|
|
conn.last_error = ""
|
|
conn.last_graph_code = None
|
|
conn.connected_at = datetime.now(timezone.utc)
|
|
conn.token_expires_at = expires_at
|
|
db.add(conn)
|
|
db.commit()
|
|
audit(db, user, "connection.authorized", asset_type="connection", asset_label=conn.meta_user_name, detail="Access token saved")
|
|
return public_connection(conn)
|
|
|
|
|
|
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")):
|
|
raise MetaApiError(
|
|
403,
|
|
"permission_denied",
|
|
"Meta has not granted Ads reporting or Ads management for this connection.",
|
|
permission="ads_read",
|
|
)
|
|
data = await ads_api(
|
|
db,
|
|
conn,
|
|
"GET",
|
|
"me/adaccounts",
|
|
{"fields": "id,name,account_status,currency,amount_spent,business{name}", "limit": "50"},
|
|
)
|
|
rows: dict[str, dict[str, Any]] = {}
|
|
for row in data.get("data") or []:
|
|
if not isinstance(row, dict) or not row.get("id"):
|
|
continue
|
|
business = row.get("business") if isinstance(row.get("business"), dict) else {}
|
|
row["business_name"] = business.get("name") or ""
|
|
rows[act_id(str(row["id"]))] = row
|
|
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")):
|
|
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.")
|
|
return conn, act_id(conn.selected_ad_account_id)
|
|
|
|
|
|
async def _account_object(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
account_id: str,
|
|
edge: str,
|
|
object_id: str,
|
|
fields: str,
|
|
) -> dict[str, Any]:
|
|
data = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
edge,
|
|
{"fields": fields, "filtering": _equal_filter("id", object_id), "limit": "1"},
|
|
)
|
|
rows = [row for row in data.get("data") or [] if isinstance(row, dict) and str(row.get("id")) == object_id]
|
|
if not rows:
|
|
raise MetaApiError(404, "not_found", "That object is not in the selected ad account.")
|
|
return rows[0]
|
|
|
|
|
|
def _adset_summary(row: dict[str, Any], currency: str) -> dict[str, Any]:
|
|
campaign = row.get("campaign") if isinstance(row.get("campaign"), dict) else {}
|
|
daily = minor_to_major(row.get("daily_budget"), currency)
|
|
lifetime = minor_to_major(row.get("lifetime_budget"), currency)
|
|
budget = money(daily, currency) if daily is not None else money(lifetime, currency) if lifetime is not None else None
|
|
goal = str(row.get("optimization_goal") or "").replace("_", " ").title()
|
|
return {
|
|
"id": str(row.get("id")),
|
|
"name": row.get("name") or "Ad set",
|
|
"campaign_id": str(row.get("campaign_id") or campaign.get("id") or ""),
|
|
"campaign_name": campaign.get("name") or "",
|
|
"status": row.get("effective_status") or row.get("status") or "",
|
|
"optimization_goal": goal or "—",
|
|
"billing_event": str(row.get("billing_event") or "").replace("_", " ").title() or "—",
|
|
"budget": budget,
|
|
"start_time": row.get("start_time"),
|
|
"end_time": row.get("end_time"),
|
|
}
|
|
|
|
|
|
async def list_adsets(db: Session, user: UserAccount, campaign_id: str | None = None) -> dict[str, Any]:
|
|
conn, account_id = await _selected_account(db, user, manage=False)
|
|
currency = await _account_currency(db, conn, account_id)
|
|
params: dict[str, Any] = {
|
|
"fields": "id,name,status,effective_status,daily_budget,lifetime_budget,optimization_goal,billing_event,campaign_id,start_time,end_time,campaign{name}",
|
|
"limit": "50",
|
|
}
|
|
if campaign_id:
|
|
params["filtering"] = _equal_filter("campaign.id", campaign_id)
|
|
data = await marketing_get(db, conn, account_id, "adsets", params)
|
|
adsets = [_adset_summary(row, currency) for row in data.get("data") or [] if isinstance(row, dict) and row.get("id")]
|
|
return {
|
|
"account_name": conn.selected_ad_account_name,
|
|
"currency": currency,
|
|
"adsets": adsets,
|
|
}
|
|
|
|
|
|
async def campaign_detail(db: Session, user: UserAccount, campaign_id: str) -> dict[str, Any]:
|
|
conn, account_id = await _selected_account(db, user, manage=False)
|
|
current = await _account_object(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"campaigns",
|
|
campaign_id,
|
|
"id,name,status,effective_status,objective,created_time",
|
|
)
|
|
currency = await _account_currency(db, conn, account_id)
|
|
adsets_payload = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"adsets",
|
|
{
|
|
"fields": "id,name,status,effective_status,daily_budget,lifetime_budget,optimization_goal,billing_event,campaign_id,start_time,end_time",
|
|
"filtering": _equal_filter("campaign.id", campaign_id),
|
|
"limit": "50",
|
|
},
|
|
)
|
|
adsets = [_adset_summary(row, currency) for row in adsets_payload.get("data") or [] if isinstance(row, dict) and row.get("id")]
|
|
insights_error = None
|
|
metrics: dict[str, Any] = {}
|
|
try:
|
|
report = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"insights",
|
|
{
|
|
"fields": "campaign_id,spend,impressions,clicks,reach,ctr",
|
|
"date_preset": "last_30d",
|
|
"level": "campaign",
|
|
"filtering": _in_filter("campaign.id", [campaign_id]),
|
|
},
|
|
)
|
|
row = (report.get("data") or [{}])[0] if isinstance(report.get("data"), list) else {}
|
|
if isinstance(row, dict):
|
|
spend = _float(row.get("spend"))
|
|
metrics = {
|
|
"spend": money(spend, currency) if row else None,
|
|
"impressions": _int(row.get("impressions")),
|
|
"clicks": _int(row.get("clicks")),
|
|
"reach": _int(row.get("reach")),
|
|
"ctr": _float(row.get("ctr")),
|
|
}
|
|
except MetaApiError as exc:
|
|
insights_error = {"code": exc.code, "message": exc.message}
|
|
return {
|
|
"account_name": conn.selected_ad_account_name,
|
|
"currency": currency,
|
|
"can_manage": has_scope(conn, "ads_management"),
|
|
"campaign": {
|
|
"id": str(current.get("id")),
|
|
"name": current.get("name") or "Campaign",
|
|
"status": current.get("effective_status") or current.get("status") or "",
|
|
"objective": OBJECTIVE_LABELS.get(str(current.get("objective")), str(current.get("objective") or "").replace("OUTCOME_", "").title()),
|
|
"created_time": current.get("created_time"),
|
|
},
|
|
"adsets": adsets,
|
|
"metrics": metrics,
|
|
"insights_error": insights_error,
|
|
}
|
|
|
|
|
|
async def list_campaigns(db: Session, user: UserAccount) -> dict[str, Any]:
|
|
conn, account_id = await _selected_account(db, user, manage=False)
|
|
data = await marketing_get(
|
|
db,
|
|
conn,
|
|
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 marketing_get(
|
|
db,
|
|
conn,
|
|
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 ads_api(db, conn, "GET", account_id, {"fields": "currency,account_status,name"})
|
|
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 _account_object(db, conn, account_id, "campaigns", campaign_id, "id,name")
|
|
name = str(current.get("name") or "Campaign")
|
|
action = "campaign.resume" if status == "ACTIVE" else "campaign.pause"
|
|
try:
|
|
await ads_object_post(db, conn, campaign_id, {"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 _account_object(db, conn, account_id, "campaigns", campaign_id, "id,name")
|
|
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 ads_object_post(db, conn, campaign_id, {"name": name[:120]})
|
|
if daily_budget is not None:
|
|
currency = await _account_currency(db, conn, account_id)
|
|
adsets = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"adsets",
|
|
{"fields": "id,name", "filtering": _equal_filter("campaign.id", campaign_id), "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 ads_object_post(db, conn, str(rows[0]["id"]), {"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 marketing_post(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"campaigns",
|
|
{
|
|
"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 marketing_post(db, conn, account_id, "adsets", adset_body)
|
|
story = _story_spec(page_id, destination, message, headline, link, cta, page_link)
|
|
creative = await marketing_post(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"adcreatives",
|
|
{"name": f"{name} creative", "object_story_spec": story},
|
|
)
|
|
await marketing_post(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"ads",
|
|
{
|
|
"name": f"{name} ad",
|
|
"adset_id": adset.get("id"),
|
|
"creative": {"creative_id": creative.get("id")},
|
|
"status": "PAUSED",
|
|
},
|
|
)
|
|
if activate and created_campaign:
|
|
await ads_object_post(db, conn, created_campaign, {"status": "ACTIVE"})
|
|
await ads_object_post(db, conn, str(adset.get("id")), {"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 marketing_get(
|
|
db,
|
|
conn,
|
|
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},adset{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 {}
|
|
adset = row.get("adset") if isinstance(row.get("adset"), dict) else {}
|
|
return {
|
|
"id": str(row.get("id")),
|
|
"name": row.get("name") or "Ad",
|
|
"campaign_name": campaign.get("name") or "",
|
|
"adset_name": adset.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 _account_object(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"ads",
|
|
ad_id,
|
|
"id,name,creative{object_story_spec}",
|
|
)
|
|
label = str(payload.get("name") or current.get("name") or "Ad")
|
|
try:
|
|
if payload.get("name"):
|
|
await ads_object_post(db, conn, ad_id, {"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 marketing_post(
|
|
db,
|
|
conn,
|
|
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 ads_object_post(db, conn, ad_id, {"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 _account_object(db, conn, account_id, "ads", ad_id, "id,name")
|
|
name = str(current.get("name") or "Ad")
|
|
action = "ad.resume" if status == "ACTIVE" else "ad.pause"
|
|
try:
|
|
await ads_object_post(db, conn, ad_id, {"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, breakdowns: bool = True) -> 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 marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"insights",
|
|
{"fields": "spend,impressions,reach,clicks,ctr,cpc,cpm", "date_preset": date_preset},
|
|
)
|
|
daily = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"insights",
|
|
{
|
|
"fields": "spend,impressions,clicks,date_start",
|
|
"date_preset": date_preset,
|
|
"time_increment": "1",
|
|
"limit": "100",
|
|
},
|
|
)
|
|
campaign_rows: list[dict[str, Any]] = []
|
|
adset_rows: list[dict[str, Any]] = []
|
|
ad_rows: list[dict[str, Any]] = []
|
|
if breakdowns:
|
|
campaign_rows = await _insights_rows(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
date_preset,
|
|
"campaign",
|
|
"campaign_id,campaign_name,spend,impressions,clicks,ctr,reach",
|
|
)
|
|
adset_rows = await _insights_rows(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
date_preset,
|
|
"adset",
|
|
"adset_id,adset_name,spend,impressions,clicks,ctr",
|
|
)
|
|
ad_rows = await _insights_rows(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
date_preset,
|
|
"ad",
|
|
"ad_id,ad_name,spend,impressions,clicks,ctr",
|
|
)
|
|
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,
|
|
"campaigns": [_insight_line(item, "campaign", currency) for item in campaign_rows],
|
|
"adsets": [_insight_line(item, "adset", currency) for item in adset_rows],
|
|
"ads": [_insight_line(item, "ad", currency) for item in ad_rows],
|
|
}
|
|
|
|
|
|
async def _insights_rows(
|
|
db: Session,
|
|
conn: MetaConnection,
|
|
account_id: str,
|
|
date_preset: str,
|
|
level: str,
|
|
fields: str,
|
|
) -> list[dict[str, Any]]:
|
|
try:
|
|
payload = await marketing_get(
|
|
db,
|
|
conn,
|
|
account_id,
|
|
"insights",
|
|
{"fields": fields, "date_preset": date_preset, "level": level, "limit": "50"},
|
|
)
|
|
except MetaApiError:
|
|
return []
|
|
return [row for row in payload.get("data") or [] if isinstance(row, dict)]
|
|
|
|
|
|
def _insight_line(row: dict[str, Any], level: str, currency: str) -> dict[str, Any]:
|
|
name = row.get(f"{level}_name") or row.get("campaign_name") or "Untitled"
|
|
spend = _float(row.get("spend"))
|
|
return {
|
|
"name": name,
|
|
"spend": money(spend, currency),
|
|
"impressions": _int(row.get("impressions")),
|
|
"clicks": _int(row.get("clicks")),
|
|
"ctr": _float(row.get("ctr")),
|
|
"reach": _int(row.get("reach")),
|
|
}
|
|
|
|
|
|
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, "ads_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", breakdowns=False)
|
|
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),
|
|
"oauth_redirect_uri": 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,
|
|
}
|