pipeline/meta_client.py

"""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 "<none>"
    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_<first8 of video_id>_<avatar_name>_<angle> (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,
    )