Video Content Analytics Pipeline

from collections import Counter

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="video-analytics")

# 1. A bucket for campaign videos, and a collection that cuts at scene changes,
#    transcribes, and has Gemini return topic and sentiment per scene
bucket = client.buckets.create(
    bucket_name="campaign-videos",
    bucket_schema={"properties": {"video": {"type": "video"}}},
)
collection = client.collections.create(
    collection_name="marketing-videos",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
        "parameters": {
            "split_method": "scene",
            "run_transcription": True,
            "run_video_description": True,
            "response_shape": {
                "type": "object",
                "properties": {
                    "topic": {"type": "string"},
                    "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
                },
            },
        },
    },
)

# 2. Upload and process
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "video", "type": "video", "data": "s3://your-bucket/campaign-videos/spot-01.mp4"}],
)
client.collections.trigger(collection["collection_id"])

# 3. Aggregate across scenes
docs = client.documents.list(collection["collection_id"], page_size=500)
sentiment = Counter((doc.get("json_output") or {}).get("sentiment") for doc in docs["results"])
seconds_by_topic = Counter()
for doc in docs["results"]:
    topic = (doc.get("json_output") or {}).get("topic", "unknown")
    seconds_by_topic[topic] += doc["end_time"] - doc["start_time"]
print(sentiment.most_common(), seconds_by_topic.most_common(5))