Brand Safety & Ad Verification Pipeline

import time

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="brand-safety")

# 1. A bucket for ad creatives, and a collection whose Gemini step scores each
#    scene against GARM categories as JSON
bucket = client.buckets.create(
    bucket_name="ad-creatives",
    bucket_schema={"properties": {"creative": {"type": "video"}}},
)
collection = client.collections.create(
    collection_name="ad-creatives",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
        "parameters": {
            "split_method": "scene",
            "run_video_description": True,
            "description_prompt": "Assess this scene for brand safety.",
            "response_shape": {
                "type": "object",
                "properties": {
                    "violence": {"type": "number"},
                    "adult": {"type": "number"},
                    "hate": {"type": "number"},
                    "drugs": {"type": "number"},
                },
            },
        },
    },
)


# 2. Check one creative: upload it, process just that object, wait, read the scores
def check_creative(creative_url):
    obj = client.buckets.upload(
        bucket["bucket_id"],
        blobs=[{"property": "creative", "type": "video", "data": creative_url}],
    )
    run = client.collections.trigger(collection["collection_id"], object_ids=[obj["object_id"]])
    while client.tasks.get(run["task_id"])["status"] not in ("COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED"):
        time.sleep(10)
    scenes = client.documents.list(
        collection["collection_id"],
        filters={"AND": [{"field": "root_object_id", "operator": "eq", "value": obj["object_id"]}]},
    )
    scores = [doc.get("json_output") or {} for doc in scenes["results"]]
    worst = max((max(s.values(), default=0) for s in scores), default=0)
    return {"safe": worst < 0.15, "scene_scores": scores}


print(check_creative("https://cdn.adnetwork.com/creative/12345.mp4"))