"""Meta Marketing API client — publish only (SPEC_NEW §7.6) — built in Phase 3. Ad-metrics *ingest* is NOT here: it runs via Windsor.ai in ingest.py. This module is the Meta publish path only. Contract: raw Graph API via httpx (no SDK). API version from META_API_VERSION. - Publish: chunked /advideos upload → /adcreatives (object_story_spec.video_data) → /ads with status=PAUSED always (v1 human approval gate). - Every call logged with the token redacted. Honors DRY_RUN: log, don't POST. The ingest() stub below is kept as a thin shim that defers to pipeline.ingest (Windsor.ai). """ from __future__ import annotations from pathlib import Path from typing import Any import httpx from . import db from .config import get_settings # Meta Graph base. Version is interpolated from META_API_VERSION (e.g. v23.0). # VERIFY-AT-BUILD-TIME: confirm the current documented Graph version + that the # /advideos, /adcreatives, /ads paths below match it (same convention as # VEO_MODEL_ID / META_API_VERSION elsewhere in the repo). GRAPH_BASE = "https://graph.facebook.com" # The default landing link when a product carries none. VERIFY-AT-BUILD-TIME. DEFAULT_LINK = "https://example.com" # Default call-to-action button enum on the creative. VERIFY-AT-BUILD-TIME. DEFAULT_CTA_TYPE = "LEARN_MORE" def ingest(days: int = 14) -> None: # Ad-metrics ingest moved to Windsor.ai — see pipeline.ingest.run(). from pipeline import ingest as _ingest return _ingest.run(days) # ── helpers ────────────────────────────────────────────────────────────────── def _redact(token: str | None) -> str: """Show only the last 4 chars of a token so logs never leak credentials.""" if not token: return "" return f"…{token[-4:]}" if len(token) > 4 else "…" def _log(msg: str) -> None: print(f"[meta] {msg}") def _ad_name(video_id: str, avatar_name: str, angle: str) -> str: """AUTO___ (SPEC_NEW §7.6).""" short = str(video_id).replace("-", "")[:8] return f"AUTO_{short}_{avatar_name}_{angle}" def _load_context(video_id: str) -> dict[str, Any]: """Load the video + its avatar + script in one go. Raises if anything required is missing (no assembled mp4 → cannot publish).""" row = db.fetch_one( """ select v.id as video_id, v.final_path as final_path, v.tags as tags, v.product_id as product_id, a.name as avatar_name, a.ref_image_path as startframe, s.cta as cta from videos v left join avatars a on a.id = v.avatar_id left join scripts s on s.id = v.script_id where v.id = %s """, (str(video_id),), ) if row is None: raise ValueError(f"no video {video_id}") if not row.get("final_path"): raise RuntimeError( f"video {video_id} has no final_path — assemble the mp4 before publishing" ) tags = row.get("tags") or {} row["angle"] = (tags.get("angle") if isinstance(tags, dict) else None) or "unknown" row["avatar_name"] = row.get("avatar_name") or "avatar" return row def _landing_link(product_id: str | None) -> str: """The ad's destination URL — from the product's knowledge if present, else a placeholder. VERIFY-AT-BUILD-TIME: confirm where the owner stores the link.""" if not product_id: return DEFAULT_LINK product = db.fetch_one("select knowledge from products where id = %s", (str(product_id),)) knowledge = (product or {}).get("knowledge") or {} if isinstance(knowledge, dict): link = knowledge.get("link") or knowledge.get("url") if link: return str(link) return DEFAULT_LINK def _insert_publish_row( *, video_id: str, meta_video_id: str, creative_id: str, ad_id: str, adset_id: str, ad_name: str, ) -> dict[str, Any]: """Insert the publishes row (published_at=now()), flip videos.status=published, and return the inserted row with uuids stringified (SPEC_NEW §7.6).""" with db.connect() as conn: row = conn.execute( """ insert into publishes (video_id, platform, meta_video_id, creative_id, ad_id, adset_id, ad_name, published_at) values (%s, 'meta', %s, %s, %s, %s, %s, now()) returning id, video_id, platform, meta_video_id, creative_id, ad_id, adset_id, ad_name, published_at """, (str(video_id), meta_video_id, creative_id, ad_id, adset_id, ad_name), ).fetchone() assert row is not None conn.execute("update videos set status = 'published' where id = %s", (str(video_id),)) row["id"] = str(row["id"]) row["video_id"] = str(row["video_id"]) return dict(row) # ── live Graph calls (only reached when not dry-run) ───────────────────────── def _upload_video(account: str, version: str, token: str, final_path: str) -> str: """(1) chunked upload POST /act_{id}/advideos → meta_video_id. VERIFY-AT-BUILD-TIME: Graph path + the start/transfer/finish chunked-upload handshake for /advideos under {version}.""" url = f"{GRAPH_BASE}/{version}/act_{account}/advideos" _log(f"POST {url} (token={_redact(token)}) chunked-upload {final_path}") size = Path(final_path).stat().st_size with httpx.Client(timeout=600) as client: start = client.post( url, data={"access_token": token, "upload_phase": "start", "file_size": size}, ) if start.status_code >= 400: raise RuntimeError(f"advideos start {start.status_code}: {start.text[:300]}") info = start.json() session_id = info["upload_session_id"] meta_video_id = info["video_id"] start_offset = int(info["start_offset"]) end_offset = int(info["end_offset"]) with open(final_path, "rb") as fh: while start_offset < end_offset: fh.seek(start_offset) chunk = fh.read(end_offset - start_offset) r = client.post( url, data={ "access_token": token, "upload_phase": "transfer", "upload_session_id": session_id, "start_offset": start_offset, }, files={"video_file_chunk": ("chunk", chunk, "application/octet-stream")}, ) if r.status_code >= 400: raise RuntimeError(f"advideos transfer {r.status_code}: {r.text[:300]}") nxt = r.json() start_offset = int(nxt["start_offset"]) end_offset = int(nxt["end_offset"]) finish = client.post( url, data={ "access_token": token, "upload_phase": "finish", "upload_session_id": session_id, }, ) if finish.status_code >= 400: raise RuntimeError(f"advideos finish {finish.status_code}: {finish.text[:300]}") _log(f"advideos → meta_video_id={meta_video_id}") return str(meta_video_id) def _create_creative( account: str, version: str, token: str, *, page_id: str, meta_video_id: str, thumbnail: str | None, link: str, cta_text: str | None, ad_name: str, ) -> str: """(2) POST /act_{id}/adcreatives with object_story_spec.video_data. VERIFY-AT-BUILD-TIME: Graph path + object_story_spec.video_data shape under {version} (page_id, image_url thumbnail, link + call_to_action).""" url = f"{GRAPH_BASE}/{version}/act_{account}/adcreatives" video_data: dict[str, Any] = { "video_id": meta_video_id, "call_to_action": {"type": DEFAULT_CTA_TYPE, "value": {"link": link}}, } if thumbnail: video_data["image_url"] = thumbnail if cta_text: video_data["message"] = cta_text story_spec = {"page_id": page_id, "video_data": video_data} _log(f"POST {url} (token={_redact(token)}) object_story_spec.video_data page={page_id}") with httpx.Client(timeout=120) as client: r = client.post( url, data={ "access_token": token, "name": ad_name, "object_story_spec": db.to_json(story_spec), }, ) if r.status_code >= 400: raise RuntimeError(f"adcreatives {r.status_code}: {r.text[:300]}") creative_id = str(r.json()["id"]) _log(f"adcreatives → creative_id={creative_id}") return creative_id def _resolve_adset(account: str, version: str, token: str, campaign_id: str) -> str: """Find the test campaign's ad set to drop the ad into. VERIFY-AT-BUILD-TIME: Graph path GET /{campaign_id}/adsets under {version}.""" url = f"{GRAPH_BASE}/{version}/{campaign_id}/adsets" _log(f"GET {url} (token={_redact(token)})") with httpx.Client(timeout=60) as client: r = client.get(url, params={"access_token": token, "fields": "id", "limit": 1}) if r.status_code >= 400: raise RuntimeError(f"campaign adsets {r.status_code}: {r.text[:300]}") data = r.json().get("data") or [] if not data: raise RuntimeError(f"test campaign {campaign_id} has no ad set to publish into") return str(data[0]["id"]) def _create_ad( account: str, version: str, token: str, *, ad_name: str, adset_id: str, creative_id: str, ) -> str: """(3) POST /act_{id}/ads into the test ad set with status=PAUSED ALWAYS (§7.6). VERIFY-AT-BUILD-TIME: Graph path /act_{id}/ads under {version}.""" url = f"{GRAPH_BASE}/{version}/act_{account}/ads" _log(f"POST {url} (token={_redact(token)}) adset={adset_id} status=PAUSED") with httpx.Client(timeout=120) as client: r = client.post( url, data={ "access_token": token, "name": ad_name, "adset_id": adset_id, "creative": db.to_json({"creative_id": creative_id}), "status": "PAUSED", # ALWAYS paused in v1 — human activates in Ads Manager }, ) if r.status_code >= 400: raise RuntimeError(f"ads {r.status_code}: {r.text[:300]}") ad_id = str(r.json()["id"]) _log(f"ads → ad_id={ad_id} (PAUSED)") return ad_id # ── public entrypoint ──────────────────────────────────────────────────────── def publish(video_id: str) -> dict: """Publish an assembled video as a PAUSED Meta ad (SPEC_NEW §7.6, Phase 3). Loads the video + avatar + script, requires videos.final_path, then either (DRY_RUN) logs the three steps with synthetic ids, or (live) chunk-uploads the mp4 → creates the creative → creates the ad PAUSED in the test campaign's ad set. Inserts a publishes row, sets videos.status='published', and returns that row (uuids stringified). The ad is ALWAYS created PAUSED — the human activates it in Ads Manager. """ s = get_settings() ctx = _load_context(video_id) ad_name = _ad_name(video_id, ctx["avatar_name"], ctx["angle"]) if s.dry_run: # DRY_RUN: never hit the live Graph API. Log the three steps + use synthetic # ids so the whole path is testable without credentials (SPEC_NEW §5). short = str(video_id).replace("-", "")[:8] meta_video_id = f"dryrun_video_{short}" creative_id = f"dryrun_creative_{short}" ad_id = f"dryrun_ad_{short}" adset_id = f"dryrun_adset_{short}" _log(f"DRY_RUN publish {video_id} → ad_name={ad_name}") _log(f"DRY_RUN (1) advideos chunked-upload {ctx['final_path']} → {meta_video_id}") _log(f"DRY_RUN (2) adcreatives object_story_spec.video_data → {creative_id}") _log(f"DRY_RUN (3) ads into test ad set, status=PAUSED → {ad_id}") return _insert_publish_row( video_id=video_id, meta_video_id=meta_video_id, creative_id=creative_id, ad_id=ad_id, adset_id=adset_id, ad_name=ad_name, ) # Live: require every Meta credential up front (fails loudly, names the var). token = s.require("META_ACCESS_TOKEN") account = s.require("META_AD_ACCOUNT_ID").replace("act_", "") version = s.require("META_API_VERSION") page_id = s.require("META_PAGE_ID") campaign_id = s.require("META_TEST_CAMPAIGN_ID") _log(f"LIVE publish {video_id} → ad_name={ad_name} (token={_redact(token)})") meta_video_id = _upload_video(account, version, token, ctx["final_path"]) creative_id = _create_creative( account, version, token, page_id=page_id, meta_video_id=meta_video_id, thumbnail=ctx.get("startframe"), link=_landing_link(ctx.get("product_id")), cta_text=ctx.get("cta"), ad_name=ad_name, ) adset_id = _resolve_adset(account, version, token, campaign_id) ad_id = _create_ad( account, version, token, ad_name=ad_name, adset_id=adset_id, creative_id=creative_id, ) return _insert_publish_row( video_id=video_id, meta_video_id=meta_video_id, creative_id=creative_id, ad_id=ad_id, adset_id=adset_id, ad_name=ad_name, )