Frequently Asked Questions

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Process multimodal content (text + image + video)
response = client.embed.multimodal(
    content={
        "text": "Product description text",
        "image_url": "https://example.com/product.jpg",
        "video_url": "https://example.com/demo.mp4"
    }
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Step 1: Retrieve relevant context
search_results = client.search(
    collection_id="your_collection",
    query="How do I process videos?",
    top_k=5
)

# Step 2: Generate answer using retrieved context
context = "\n".join([r["content"] for r in search_results])
answer = f"Based on our documentation: {context}"

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Create embeddings for your content
embedding = client.embed.text(
    text="Find videos about machine learning"
)

# Search using vector similarity
results = client.search(
    collection_id="videos",
    vector=embedding.vector,
    top_k=10,
    filters={"modality": "video"}
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Text embedding
text_embed = client.embed.text(text="Product description")
# Returns: [0.123, -0.456, 0.789, ...] (768 dimensions)

# Image embedding
image_embed = client.embed.image(image_url="https://example.com/product.jpg")
# Returns similar format, comparable with text embeddings

# Video embedding (per frame or scene)
video_embed = client.embed.video(video_url="https://example.com/demo.mp4")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Hybrid search combines keyword + semantic
results = client.search(
    collection_id="products",
    query="wireless bluetooth headphones with noise cancellation",
    hybrid_search=True,
    hybrid_config={
        "keyword_weight": 0.3,  # 30% keyword matching
        "vector_weight": 0.7,   # 70% semantic similarity
        "rrf_k": 60  # Reciprocal Rank Fusion parameter
    },
    top_k=20
)

from mixpeek import Mixpeek
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

client = Mixpeek(token="your_token")

# Define classes with descriptions
classes = {
    "billing": "Questions about invoices, payments, refunds, or pricing",
    "technical": "Problems with product features, bugs, or performance",
    "sales": "Inquiries about purchasing, demos, or product capabilities"
}

# Embed the customer message
message = "My payment didn't go through and I was charged twice"
message_embedding = client.embed.text(text=message)

# Embed class descriptions
class_embeddings = {
    name: client.embed.text(text=desc)
    for name, desc in classes.items()
}

# Find most similar class
similarities = {
    name: cosine_similarity(
        [message_embedding.vector],
        [class_embedding.vector]
    )[0][0]
    for name, class_embedding in class_embeddings.items()
}

predicted_class = max(similarities, key=similarities.get)
print(f"Class: {predicted_class}, Confidence: {similarities[predicted_class]:.2%}")
# Output: Class: billing, Confidence: 0.87

from mixpeek import Mixpeek

# Initialize client
client = Mixpeek(token="your_api_token")

# Create a collection for your content
collection = client.collections.create(
    name="my_media_library",
    feature_extractors=["video-embedding", "scene-splitting", "transcription"]
)

# Index content from S3, URLs, or local files
client.ingest(
    collection_id=collection.id,
    source="s3://my-bucket/videos/",
    process_async=True
)

# Search semantically across all content
results = client.search(
    collection_id=collection.id,
    query="people playing basketball outdoors",
    top_k=10
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Create collection with multiple feature extractors
collection = client.collections.create(
    name="video_analysis",
    feature_extractors=[
        "scene-splitting",      # Detect scene changes
        "video-embedding",      # Generate semantic embeddings
        "transcription",        # Speech-to-text
        "object-detection",     # Detect objects in frames
        "face-detection"        # Detect and track faces
    ],
    config={
        "transcription": {"language": "en"},
        "object_detection": {"confidence_threshold": 0.7}
    }
)

# Features are automatically extracted during ingestion
client.index(
    collection_id=collection.id,
    video_url="https://example.com/video.mp4"
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# CLIP-style cross-modal search
# Search images using text
image_results = client.search(
    collection_id="product_images",
    query="red leather handbag with gold hardware",  # Text query
    modality_filter="image",
    top_k=10
)

# Search text using an image
text_results = client.search(
    collection_id="product_descriptions",
    query_image_url="https://example.com/handbag.jpg",  # Image query
    modality_filter="text",
    top_k=10
)

# Zero-shot image classification
categories = ["laptop", "phone", "tablet", "watch"]
image_embedding = client.embed.image(image_url="https://example.com/device.jpg")

category_embeddings = [
    client.embed.text(text=f"a photo of a {cat}")
    for cat in categories
]

# Compare and find best match (CLIP-style zero-shot)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Set up content moderation pipeline
collection = client.collections.create(
    name="user_generated_content",
    feature_extractors=[
        "nsfw-detection",        # Detects adult content
        "violence-detection",    # Detects violent imagery
        "hate-speech-detection", # Detects hate speech in text
        "video-embedding"        # For similarity to known violations
    ]
)

# Index user upload
result = client.index(
    collection_id=collection.id,
    video_url="https://example.com/user-video.mp4",
    metadata={"user_id": "user123", "upload_time": "2024-01-15"}
)

# Check moderation flags
if result.features.get("nsfw_score", 0) > 0.8:
    # High confidence NSFW, auto-reject
    action = "reject"
elif result.features.get("violence_score", 0) > 0.6:
    # Medium confidence violence, queue for human review
    action = "review"
else:
    # Passed automated checks
    action = "approve"

print(f"Moderation decision: {action}")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Enable object detection
collection = client.collections.create(
    name="video_objects",
    feature_extractors=[
        "object-detection",  # Detect objects in frames
        "video-embedding"    # For semantic search
    ],
    config={
        "object_detection": {
            "taxonomy": "coco",  # 80 common objects
            "confidence_threshold": 0.6,
            "frame_interval": 30  # Process every 30th frame
        }
    }
)

# Index video
result = client.index(
    collection_id=collection.id,
    video_url="https://example.com/video.mp4"
)

# Query: Find videos with specific objects
results = client.search(
    collection_id=collection.id,
    filters={
        "detected_objects": {"$in": ["person", "bicycle", "car"]}
    },
    top_k=20
)

# Each result includes object detections
for match in results:
    print(f"Video: {match.video_id}")
    for detection in match.objects:
        print(f"  {detection.label} at {detection.timestamp}s "
              f"(confidence: {detection.confidence:.2%})")

from mixpeek import Mixpeek
from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
client = Mixpeek(token="your_token")

# Set up ingestion with webhook
def ingest_with_webhook(video_url, collection_id):
    job = client.index(
        collection_id=collection_id,
        video_url=video_url,
        process_async=True,
        webhook_url="https://myapp.com/webhooks/mixpeek"
    )
    return job.id

# Webhook endpoint
@app.route('/webhooks/mixpeek', methods=['POST'])
def handle_mixpeek_webhook():
    # Step 1: Verify signature
    signature = request.headers.get('X-Mixpeek-Signature')
    webhook_secret = "your_webhook_secret"

    computed_sig = hmac.new(
        webhook_secret.encode(),
        request.data,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, computed_sig):
        return jsonify({"error": "Invalid signature"}), 401

    # Step 2: Parse payload
    payload = request.json
    event_type = payload["event"]
    data = payload["data"]

    # Step 3: Handle event
    if event_type == "ingestion.completed":
        # Process completed ingestion
        print(f"File processed: {data['file_id']}")
        print(f"Features: {data['features']}")

        # Trigger downstream actions (e.g., notify user, update database)
        notify_user(data['file_id'], "processing_complete")

    elif event_type == "ingestion.failed":
        # Handle failure
        print(f"Processing failed: {data['error']}")
        log_error(data)

    # Step 4: Respond quickly (200 OK)
    return jsonify({"status": "received"}), 200

def notify_user(file_id, status):
    # Async notification logic
    pass

if __name__ == '__main__':
    app.run(port=8080)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Step 1: Index your video library
collection = client.collections.create(
    name="video_library",
    feature_extractors=["scene-splitting", "video-embedding"]
)

# Step 2: Add videos to collection
client.ingest(
    collection_id=collection.id,
    source="s3://my-videos/",
    process_async=True
)

# Step 3: Search by uploading a query video
results = client.search(
    collection_id=collection.id,
    query_video_url="https://example.com/query-video.mp4",
    top_k=10
)

# Results contain similar videos with timestamps
for result in results:
    print(f"Video: {result.video_id}, Timestamp: {result.timestamp}, Score: {result.score}")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Step 1: Create collection with embeddings
collection = client.collections.create(
    name="product_search",
    feature_extractors=["text-embedding", "image-embedding"]
)

# Step 2: Index your products
for product in products:
    client.index(
        collection_id=collection.id,
        content={
            "text": product.description,
            "image_url": product.image,
            "metadata": {"category": product.category, "price": product.price}
        }
    )

# Step 3: Search semantically
results = client.search(
    collection_id=collection.id,
    query="comfortable running shoes for marathon",
    filters={"price": {"$lt": 200}},
    top_k=20
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Zero-shot video classification
categories = ["tutorial", "product demo", "customer testimonial", "advertisement"]

# Process video
video_embedding = client.embed.video(
    video_url="https://example.com/video.mp4",
    scene_detection=True
)

# Compare against category descriptions
category_embeddings = [
    client.embed.text(f"A video showing a {cat}")
    for cat in categories
]

# Find best matching category
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

similarities = cosine_similarity(
    [video_embedding.vector],
    category_embeddings
)[0]

predicted_category = categories[np.argmax(similarities)]
confidence = np.max(similarities)

print(f"Category: {predicted_category} (confidence: {confidence:.2%})") 

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Use multilingual embedding model
collection = client.collections.create(
    name="multilingual_docs",
    feature_extractors=["multilingual-text-embedding"]
)

# Index content in multiple languages
client.index(collection_id=collection.id, content={"text": "Hello world", "lang": "en"})
client.index(collection_id=collection.id, content={"text": "Hola mundo", "lang": "es"})
client.index(collection_id=collection.id, content={"text": "Bonjour le monde", "lang": "fr"})

# Search in English, finds results in all languages
results = client.search(
    collection_id=collection.id,
    query="greetings and salutations",  # English query
    top_k=10
)
# Returns: Hello world, Hola mundo, Bonjour le monde...

from mixpeek import Mixpeek
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

client = Mixpeek(token="your_token")

# Index video library with embeddings
collection = client.collections.create(
    name="video_dedup",
    feature_extractors=["scene-splitting", "video-embedding"]
)

# Get all video embeddings
videos = client.list(collection_id=collection.id)

# Compare each video against others
duplicates = []
threshold = 0.90  # 90% similarity = likely duplicate

for i, video1 in enumerate(videos):
    for j, video2 in enumerate(videos[i+1:], i+1):
        similarity = cosine_similarity(
            [video1.embedding],
            [video2.embedding]
        )[0][0]

        if similarity > threshold:
            duplicates.append({
                "video1": video1.id,
                "video2": video2.id,
                "similarity": similarity
            })

print(f"Found {len(duplicates)} potential duplicate pairs")

# Step 1: Install SDK
# pip install mixpeek

# Step 2: Import and authenticate
from mixpeek import Mixpeek
client = Mixpeek(token="your_api_token")

# Step 3: Create collection
collection = client.collections.create(
    name="my_first_collection",
    feature_extractors=["image-embedding", "text-embedding"]
)

# Step 4: Index some content
client.index(
    collection_id=collection.id,
    content={
        "image_url": "https://example.com/image.jpg",
        "text": "A beautiful sunset over mountains",
        "metadata": {"category": "landscape"}
    }
)

# Step 5: Search!
results = client.search(
    collection_id=collection.id,
    query="mountain scenery at dusk",
    top_k=5
)

for result in results:
    print(f"Match: {result.metadata}, Score: {result.score}")

import os
from mixpeek import Mixpeek
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

# Production configuration
client = Mixpeek(
    token=os.environ["MIXPEEK_API_TOKEN"],  # From environment
    timeout=30,  # Request timeout
    max_retries=3
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Retry logic with exponential backoff
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def search_with_retry(collection_id, query):
    try:
        results = client.search(
            collection_id=collection_id,
            query=query,
            top_k=10
        )
        logger.info(f"Search successful: {len(results)} results")
        return results
    except Exception as e:
        logger.error(f"Search failed: {str(e)}")
        raise

# Async ingestion with webhook callback
def ingest_async(collection_id, source):
    job = client.ingest(
        collection_id=collection_id,
        source=source,
        process_async=True,
        webhook_url="https://myapp.com/webhooks/mixpeek/complete"
    )
    logger.info(f"Ingestion job started: {job.id}")
    return job.id

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Example: E-commerce visual search
# Need: image similarity + cross-modal text-to-image search
collection_visual = client.collections.create(
    name="products_visual_search",
    feature_extractors=[
        "clip-image-embedding",  # For cross-modal (text→image search)
        "object-detection"       # For attributes (color, category)
    ]
)

# Example: Document search
# Need: text-only, multilingual, high quality
collection_docs = client.collections.create(
    name="documents_multilingual",
    feature_extractors=[
        "multilingual-e5-large"  # Supports 100+ languages, 1024d
    ]
)

# Example: Video content moderation
# Need: fast processing, good enough quality
collection_moderation = client.collections.create(
    name="video_moderation",
    feature_extractors=[
        "clip-video-embedding-small",  # Faster, 512d
        "nsfw-detection",               # Specialized classifier
        "transcription"                 # Speech content
    ]
)

# Test and compare model quality
# Run evaluation set through different models, measure recall

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Set up face recognition pipeline
collection = client.collections.create(
    name="video_faces",
    feature_extractors=[
        "face-detection",  # Detect faces in frames
        "face-embedding",  # Generate face embeddings
        "face-tracking"    # Track faces across frames
    ]
)

# Index video library
client.ingest(
    collection_id=collection.id,
    source="s3://security-footage/",
    process_async=True
)

# Search for a specific person
# Provide reference image of the person
reference_face = client.embed.face(
    image_url="https://example.com/person.jpg"
)

# Find all appearances of this person
results = client.search(
    collection_id=collection.id,
    vector=reference_face.vector,
    filters={"has_face": True},
    top_k=100,
    min_score=0.85  # High threshold for face matching
)

# Group by video
from collections import defaultdict
appearances = defaultdict(list)
for result in results:
    appearances[result.video_id].append({
        "timestamp": result.timestamp,
        "confidence": result.score
    })

print(f"Found person in {len(appearances)} videos")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Set up scene-based video search
collection = client.collections.create(
    name="timestamped_videos",
    feature_extractors=[
        "scene-splitting",   # Detect scene boundaries
        "video-embedding",   # Embed each scene
        "transcription"      # For spoken content search
    ]
)

# Index video - automatically split into scenes with timestamps
client.index(
    collection_id=collection.id,
    video_url="https://example.com/lecture.mp4",
    metadata={"title": "ML Lecture Series Episode 1"}
)

# Search within videos
results = client.search(
    collection_id=collection.id,
    query="explanation of gradient descent",
    top_k=5
)

# Results include precise timestamps
for result in results:
    print(f"Video: {result.metadata['title']}")
    print(f"  Jump to: {result.start_time}s - {result.end_time}s")
    print(f"  Scene content: {result.transcript_snippet}")
    print(f"  Relevance: {result.score:.2%}")
    print(f"  Direct link: https://example.com/player?v={result.video_id}&t={result.start_time}")
    print()

# Advanced: Multi-moment search (find multiple relevant segments in same video)
all_moments = client.search(
    collection_id=collection.id,
    query="gradient descent",
    top_k=50,  # Get more results
    group_by="video_id",  # Group by video
    max_per_group=5  # Up to 5 moments per video
)

from mixpeek import Mixpeek
import numpy as np
from sklearn.metrics import ndcg_score

client = Mixpeek(token="your_token")

# Step 1: Prepare test set
test_queries = [
    {"query": "red leather handbag", "relevant_ids": ["prod_123", "prod_456"]},
    {"query": "wireless headphones", "relevant_ids": ["prod_789", "prod_012"]},
    # ... more test cases
]

# Step 2: Run searches and collect results
def evaluate_model(collection_id, test_set):
    recall_at_10 = []
    mrr_scores = []

    for test in test_set:
        results = client.search(
            collection_id=collection_id,
            query=test["query"],
            top_k=10
        )

        # Calculate Recall@10
        retrieved_ids = [r.id for r in results]
        relevant_retrieved = set(retrieved_ids) & set(test["relevant_ids"])
        recall = len(relevant_retrieved) / len(test["relevant_ids"])
        recall_at_10.append(recall)

        # Calculate MRR
        for i, result in enumerate(results, 1):
            if result.id in test["relevant_ids"]:
                mrr_scores.append(1.0 / i)
                break
        else:
            mrr_scores.append(0.0)

    return {
        "recall@10": np.mean(recall_at_10),
        "mrr": np.mean(mrr_scores)
    }

# Step 3: Compare models
model_a_metrics = evaluate_model("collection_model_a", test_queries)
model_b_metrics = evaluate_model("collection_model_b", test_queries)

print(f"Model A - Recall@10: {model_a_metrics['recall@10']:.2%}, MRR: {model_a_metrics['mrr']:.3f}")
print(f"Model B - Recall@10: {model_b_metrics['recall@10']:.2%}, MRR: {model_b_metrics['mrr']:.3f}")

from mixpeek import Mixpeek
import psycopg2

client = Mixpeek(token="your_token")
db = psycopg2.connect("postgresql://localhost/mydb")

# When creating a product in your database
def create_product(product_data):
    # Step 1: Insert into your database
    cursor = db.cursor()
    cursor.execute(
        "INSERT INTO products (title, description, image_url, price) VALUES (%s, %s, %s, %s) RETURNING id",
        (product_data["title"], product_data["description"], product_data["image_url"], product_data["price"])
    )
    product_id = cursor.fetchone()[0]
    db.commit()

    # Step 2: Index in Mixpeek for search
    mixpeek_result = client.index(
        collection_id="products",
        content={
            "text": f"{product_data['title']} {product_data['description']}",
            "image_url": product_data["image_url"],
            "metadata": {
                "db_id": product_id,  # Reference back to database
                "price": product_data["price"]
            }
        }
    )

    # Step 3: Store Mixpeek ID in database
    cursor.execute(
        "UPDATE products SET mixpeek_id = %s WHERE id = %s",
        (mixpeek_result.id, product_id)
    )
    db.commit()

    return product_id

# When searching
def search_products(query, max_price=None):
    # Step 1: Search in Mixpeek
    filters = {"price": {"$lt": max_price}} if max_price else {}
    search_results = client.search(
        collection_id="products",
        query=query,
        filters=filters,
        top_k=20
    )

    # Step 2: Fetch full records from database
    db_ids = [r.metadata["db_id"] for r in search_results]
    cursor = db.cursor()
    cursor.execute(
        "SELECT * FROM products WHERE id = ANY(%s)",
        (db_ids,)
    )
    products = cursor.fetchall()

    # Step 3: Combine search relevance with database data
    return products

import os
from mixpeek import Mixpeek

# GOOD: Load from environment variable
client = Mixpeek(token=os.environ["MIXPEEK_API_TOKEN"])

# BAD: Never hardcode tokens!
# client = Mixpeek(token="sk_live_abc123...")

# For production: Use secrets manager
import boto3

def get_mixpeek_client():
    # Fetch token from AWS Secrets Manager
    secrets = boto3.client('secretsmanager')
    response = secrets.get_secret_value(SecretId='prod/mixpeek/api_token')
    token = response['SecretString']

    return Mixpeek(token=token)

# Manual API requests (without SDK)
import requests

headers = {
    "Authorization": f"Bearer {os.environ['MIXPEEK_API_TOKEN']}",
    "Content-Type": "application/json"
}

response = requests.post(
    "https://api.mixpeek.com/v1/retrievers/my_retriever/execute",
    headers=headers,
    json={
        "inputs": {"query": "search query", "top_k": 10}
    }
)

# Token rotation strategy
# 1. Create new token in dashboard
# 2. Deploy new token to all services
# 3. Monitor for errors
# 4. Revoke old token after 24h grace period

from mixpeek import Mixpeek
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

client = Mixpeek(token="your_token")

# Batch ingest from S3
def batch_ingest_s3(bucket_path, collection_id, batch_size=100):
    """
    Process videos from S3 in batches
    """
    job = client.ingest(
        collection_id=collection_id,
        source=f"s3://{bucket_path}",
        process_async=True,
        batch_size=batch_size,
        webhook_url="https://myapp.com/webhooks/batch-complete"
    )

    print(f"Started batch job: {job.id}")
    return job.id

# Monitor batch progress
def monitor_batch_job(job_id, timeout=3600):
    """
    Poll job status until complete or timeout
    """
    start_time = time.time()

    while time.time() - start_time < timeout:
        status = client.jobs.get(job_id)

        print(f"Progress: {status.processed}/{status.total} files")
        print(f"Failed: {status.failed} files")

        if status.status == "completed":
            return status
        elif status.status == "failed":
            raise Exception(f"Batch job failed: {status.error}")

        time.sleep(30)  # Check every 30 seconds

    raise TimeoutError(f"Job {job_id} didn't complete in {timeout}s")

# Process multiple batches in parallel
def process_multiple_buckets(bucket_paths, collection_id):
    """
    Process multiple S3 paths concurrently
    """
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {
            executor.submit(batch_ingest_s3, path, collection_id): path
            for path in bucket_paths
        }

        for future in as_completed(futures):
            path = futures[future]
            try:
                job_id = future.result()
                print(f"Successfully started job for {path}: {job_id}")
            except Exception as e:
                print(f"Failed to start job for {path}: {e}")

# Usage
bucket_paths = [
    "my-videos/2024/january/",
    "my-videos/2024/february/",
    "my-videos/2024/march/"
]

process_multiple_buckets(bucket_paths, "my_collection")

from mixpeek import Mixpeek
import boto3
from botocore.exceptions import ClientError

client = Mixpeek(token="your_token")
s3 = boto3.client('s3')

# Method 1: Bulk ingest from S3 (using IAM role)
def ingest_s3_bucket(bucket_name, prefix, collection_id):
    """
    Ingest all videos from S3 bucket/prefix
    Requires: Mixpeek has IAM role with GetObject permission
    """
    job = client.ingest(
        collection_id=collection_id,
        source=f"s3://{bucket_name}/{prefix}",
        process_async=True,
        filters={
            "extensions": [".mp4", ".mov", ".avi"]  # Only process videos
        }
    )
    return job.id

# Method 2: Individual files with pre-signed URLs
def ingest_with_presigned_url(bucket_name, key, collection_id):
    """
    Generate pre-signed URL and ingest single file
    No IAM configuration needed
    """
    try:
        # Generate pre-signed URL (valid for 1 hour)
        presigned_url = s3.generate_presigned_url(
            'get_object',
            Params={'Bucket': bucket_name, 'Key': key},
            ExpiresIn=3600
        )

        # Ingest using pre-signed URL
        result = client.index(
            collection_id=collection_id,
            video_url=presigned_url,
            metadata={
                "s3_bucket": bucket_name,
                "s3_key": key
            }
        )
        return result.id

    except ClientError as e:
        print(f"Error generating pre-signed URL: {e}")
        return None

# Method 3: S3 Event-driven ingestion
# Configure S3 bucket notification to trigger Lambda function
# Lambda function calls Mixpeek to index new uploads
def lambda_handler(event, context):
    """
    AWS Lambda function triggered by S3 PutObject events
    """
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Only process video files
        if key.endswith(('.mp4', '.mov', '.avi')):
            print(f"New video uploaded: {bucket}/{key}")

            # Ingest into Mixpeek
            result = client.index(
                collection_id="my_collection",
                video_url=f"s3://{bucket}/{key}",
                metadata={"upload_time": record['eventTime']}
            )

            print(f"Indexed: {result.id}")

    return {"statusCode": 200}

# Usage
ingest_s3_bucket("my-video-bucket", "2024/videos/", "my_collection")

from mixpeek import Mixpeek
from datetime import datetime, timedelta

client = Mixpeek(token="your_token")

# Strategy 1: Time-based partitioning
def get_collection_for_date(date):
    """
    Route queries to monthly collections
    """
    return f"videos_{date.strftime('%Y_%m')}"

def search_across_time_partitions(query, start_date, end_date):
    """
    Search across multiple time-based collections
    """
    results = []
    current_date = start_date

    while current_date <= end_date:
        collection_id = get_collection_for_date(current_date)

        try:
            monthly_results = client.search(
                collection_id=collection_id,
                query=query,
                top_k=20
            )
            results.extend(monthly_results)
        except Exception as e:
            print(f"No collection for {current_date}: {e}")

        # Move to next month
        current_date += timedelta(days=32)
        current_date = current_date.replace(day=1)

    # Re-rank combined results
    results.sort(key=lambda x: x.score, reverse=True)
    return results[:20]  # Top 20 overall

# Strategy 2: Category-based partitioning
CATEGORY_COLLECTIONS = {
    "electronics": "products_electronics",
    "clothing": "products_clothing",
    "books": "products_books",
    "home": "products_home"
}

def search_by_category(query, category):
    """
    Route to category-specific collection
    """
    collection_id = CATEGORY_COLLECTIONS.get(category)

    if not collection_id:
        raise ValueError(f"Unknown category: {category}")

    return client.search(
        collection_id=collection_id,
        query=query,
        top_k=20
    )

# Strategy 3: Multi-tenant with shared collection
def search_tenant_data(query, tenant_id, top_k=10):
    """
    Search within tenant's data in shared collection
    """
    results = client.search(
        collection_id="shared_multi_tenant",
        query=query,
        filters={"tenant_id": tenant_id},  # Critical: isolate tenant data
        top_k=top_k
    )
    return results

# Strategy 4: Data archival
def archive_old_data(collection_id, days_old=90):
    """
    Move old items to archive collection
    """
    cutoff_date = datetime.now() - timedelta(days=days_old)

    # Find old items
    old_items = client.list(
        collection_id=collection_id,
        filters={
            "created_at": {"$lt": cutoff_date.isoformat()}
        }
    )

    # Move to archive
    for item in old_items:
        client.index(
            collection_id=f"{collection_id}_archive",
            content=item.content,
            metadata=item.metadata
        )

        # Delete from main collection
        client.delete(collection_id=collection_id, item_id=item.id)

    print(f"Archived {len(old_items)} items")

from mixpeek import Mixpeek
import hashlib
import redis
import pickle
from typing import Optional

client = Mixpeek(token="your_token")
cache = redis.Redis(host='localhost', port=6379, db=1)

MODEL_VERSION = "v1"  # Update when embedding model changes

def compute_content_hash(content: str) -> str:
    """
    Create deterministic hash for content
    """
    # Normalize content (lowercase, trim, etc.)
    normalized = content.lower().strip()
    return hashlib.sha256(normalized.encode()).hexdigest()

def get_cached_embedding(content: str, model: str = "text") -> Optional[list]:
    """
    Check cache for existing embedding
    """
    content_hash = compute_content_hash(content)
    cache_key = f"embedding:{model}:{MODEL_VERSION}:{content_hash}"

    cached = cache.get(cache_key)
    if cached:
        print(f"Cache hit for {content_hash[:8]}")
        return pickle.loads(cached)

    return None

def embed_with_cache(content: str, model: str = "text") -> list:
    """
    Get embedding with caching
    """
    # Check cache first
    cached_embedding = get_cached_embedding(content, model)
    if cached_embedding:
        return cached_embedding

    # Cache miss - compute embedding
    print(f"Cache miss - embedding content")
    if model == "text":
        result = client.embed.text(text=content)
    elif model == "image":
        result = client.embed.image(image_url=content)
    else:
        raise ValueError(f"Unknown model: {model}")

    embedding = result.vector

    # Store in cache (TTL: 30 days)
    content_hash = compute_content_hash(content)
    cache_key = f"embedding:{model}:{MODEL_VERSION}:{content_hash}"
    cache.setex(
        cache_key,
        30 * 24 * 3600,  # 30 days
        pickle.dumps(embedding)
    )

    return embedding

# Advanced: Deduplication with embedding cache
def index_with_dedup(collection_id: str, items: list):
    """
    Index items, reusing embeddings for duplicates
    """
    embeddings_computed = 0
    embeddings_cached = 0

    for item in items:
        # Check if we've seen this content before
        embedding = get_cached_embedding(item["text"])

        if embedding:
            embeddings_cached += 1
        else:
            # New content - compute and cache
            embedding = embed_with_cache(item["text"])
            embeddings_computed += 1

        # Index with pre-computed embedding
        client.index(
            collection_id=collection_id,
            vector=embedding,  # Pass embedding directly
            metadata=item["metadata"]
        )

    print(f"Computed: {embeddings_computed}, Cached: {embeddings_cached}")
    print(f"Cost savings: {embeddings_cached / (embeddings_computed + embeddings_cached):.1%}")

# Cache migration when model updates
def migrate_embeddings(old_version: str, new_version: str):
    """
    Gradually migrate to new embedding model
    """
    # Get all cached keys for old model
    old_pattern = f"embedding:*:{old_version}:*"
    old_keys = cache.keys(old_pattern)

    print(f"Found {len(old_keys)} old embeddings to migrate")

    # Option: Delete old embeddings (force recompute)
    # for key in old_keys:
    #     cache.delete(key)

    # Option: Keep old embeddings during transition period
    # (both versions coexist, gradually delete old as they expire)

from mixpeek import Mixpeek
import time
import logging
from datadog import statsd
from functools import wraps

client = Mixpeek(token="your_token")
logger = logging.getLogger(__name__)

def monitor_search(func):
    """
    Decorator to monitor search performance
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        query = kwargs.get('query', 'unknown')

        try:
            # Execute search
            results = func(*args, **kwargs)

            # Measure latency
            latency_ms = (time.time() - start_time) * 1000

            # Log metrics
            statsd.histogram('search.latency', latency_ms)
            statsd.increment('search.requests')
            statsd.histogram('search.results_count', len(results))

            # Log query details
            logger.info({
                "event": "search_completed",
                "query": query,
                "latency_ms": latency_ms,
                "result_count": len(results),
                "top_k": kwargs.get('top_k', 10),
                "filters": kwargs.get('filters')
            })

            # Alert on slow queries
            if latency_ms > 1000:
                logger.warning(f"Slow query detected: {query} took {latency_ms}ms")
                statsd.increment('search.slow_queries')

            return results

        except Exception as e:
            # Track errors
            statsd.increment('search.errors')
            logger.error(f"Search failed: {query}, error: {e}")
            raise

    return wrapper

@monitor_search
def search_products(collection_id, query, filters=None, top_k=10):
    """
    Monitored search function
    """
    return client.search(
        collection_id=collection_id,
        query=query,
        filters=filters,
        top_k=top_k
    )

# Track user interactions for quality metrics
def track_search_interaction(query, results, clicked_result_id=None):
    """
    Log user interaction with search results
    """
    event = {
        "event": "search_interaction",
        "query": query,
        "result_count": len(results),
        "clicked": clicked_result_id is not None
    }

    if clicked_result_id:
        # Find rank of clicked result
        for i, result in enumerate(results):
            if result.id == clicked_result_id:
                event["clicked_rank"] = i + 1
                statsd.histogram('search.clicked_rank', i + 1)
                break

        # Calculate CTR
        statsd.increment('search.with_clicks')
    else:
        statsd.increment('search.without_clicks')

    logger.info(event)

# Analytics query for slow searches
def find_slow_queries(log_file, threshold_ms=500):
    """
    Analyze logs to find slow queries
    """
    import json

    slow_queries = []

    with open(log_file, 'r') as f:
        for line in f:
            try:
                log = json.loads(line)
                if log.get('event') == 'search_completed':
                    if log.get('latency_ms', 0) > threshold_ms:
                        slow_queries.append({
                            "query": log['query'],
                            "latency": log['latency_ms'],
                            "filters": log.get('filters')
                        })
            except:
                continue

    # Sort by latency
    slow_queries.sort(key=lambda x: x['latency'], reverse=True)

    print(f"Top 10 slowest queries:")
    for q in slow_queries[:10]:
        print(f"  {q['query']}: {q['latency']}ms")

    return slow_queries

# Usage
results = search_products(
    collection_id="products",
    query="wireless headphones",
    filters={"price": {"$lt": 200}},
    top_k=20
)

# User clicked result #3
track_search_interaction(
    query="wireless headphones",
    results=results,
    clicked_result_id=results[2].id
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Step 1: Create product search collection
collection = client.collections.create(
    name="product_visual_search",
    feature_extractors=[
        "clip-image-embedding",  # For cross-modal search
        "text-embedding",         # For product descriptions
        "object-detection"        # Extract product attributes
    ]
)

# Step 2: Index product catalog
def index_product_catalog(products):
    for product in products:
        client.index(
            collection_id=collection.id,
            content={
                "image_url": product["image_url"],
                "text": f"{product['title']} {product['description']}",
                "metadata": {
                    "product_id": product["id"],
                    "category": product["category"],
                    "price": product["price"],
                    "brand": product["brand"],
                    "colors": product["colors"],
                    "in_stock": product["in_stock"]
                }
            }
        )

# Step 3: Search by text query
def search_by_text(query, category=None, max_price=None):
    filters = {"in_stock": True}

    if category:
        filters["category"] = category
    if max_price:
        filters["price"] = {"$lt": max_price}

    return client.search(
        collection_id=collection.id,
        query=query,
        filters=filters,
        top_k=20
    )

# Step 4: Search by image upload (find similar)
def search_by_image(uploaded_image_url, category=None):
    filters = {"in_stock": True}

    if category:
        filters["category"] = category

    # Search using image instead of text
    return client.search(
        collection_id=collection.id,
        query_image_url=uploaded_image_url,
        filters=filters,
        top_k=20
    )

# Step 5: Hybrid search (text + filters + visual)
def advanced_product_search(text_query, color=None, price_range=None):
    filters = {"in_stock": True}

    if color:
        filters["colors"] = {"$contains": color}
    if price_range:
        filters["price"] = {
            "$gte": price_range["min"],
            "$lte": price_range["max"]
        }

    results = client.search(
        collection_id=collection.id,
        query=text_query,
        filters=filters,
        hybrid_search=True,  # Combine keyword + semantic
        top_k=50
    )

    return results

# Usage examples
# Text search: "summer dresses"
text_results = search_by_text(
    query="summer dresses",
    category="womens-clothing",
    max_price=100
)

# Image search: User uploads a photo
image_results = search_by_image(
    uploaded_image_url="https://user-upload.com/dress.jpg",
    category="womens-clothing"
)

# Advanced: "red leather handbags under $200"
hybrid_results = advanced_product_search(
    text_query="leather handbag",
    color="red",
    price_range={"min": 50, "max": 200}
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Set up compliance checking pipeline
compliance_collection = client.collections.create(
    name="content_compliance",
    feature_extractors=[
        "nsfw-detection",          # Adult content
        "violence-detection",      # Violent imagery
        "hate-speech-detection",   # Hate speech
        "medical-claims-detection",# Health misinformation
        "logo-detection",          # Brand/trademark
        "transcription"            # Audio content analysis
    ]
)

def check_compliance(content_url, content_type="video"):
    """
    Run compliance checks on content
    """
    # Process content
    result = client.index(
        collection_id=compliance_collection.id,
        video_url=content_url if content_type == "video" else None,
        image_url=content_url if content_type == "image" else None,
        metadata={
            "upload_time": "2024-01-15",
            "uploader_id": "user123"
        }
    )

    # Evaluate compliance scores
    compliance_status = {
        "approved": True,
        "flags": [],
        "requires_review": False,
        "violations": []
    }

    # NSFW check
    if result.features.get("nsfw_score", 0) > 0.7:
        compliance_status["approved"] = False
        compliance_status["violations"].append("NSFW content detected")
    elif result.features.get("nsfw_score", 0) > 0.4:
        compliance_status["requires_review"] = True
        compliance_status["flags"].append("Potential NSFW content")

    # Violence check
    if result.features.get("violence_score", 0) > 0.6:
        compliance_status["approved"] = False
        compliance_status["violations"].append("Violent content detected")

    # Hate speech check
    if result.features.get("hate_speech_score", 0) > 0.5:
        compliance_status["approved"] = False
        compliance_status["violations"].append("Hate speech detected")

    # Medical claims check (for advertising compliance)
    if result.features.get("medical_claims", []):
        compliance_status["requires_review"] = True
        compliance_status["flags"].append("Medical claims require verification")

    # Brand safety check
    detected_logos = result.features.get("logos", [])
    for logo in detected_logos:
        if logo["brand"] in RESTRICTED_BRANDS:
            compliance_status["requires_review"] = True
            compliance_status["flags"].append(f"Restricted brand detected: {logo['brand']}")

    # Log compliance check
    log_compliance_check(
        content_id=result.id,
        status=compliance_status,
        features=result.features
    )

    return compliance_status

def log_compliance_check(content_id, status, features):
    """
    Create audit trail for compliance
    """
    audit_log = {
        "timestamp": "2024-01-15T10:30:00Z",
        "content_id": content_id,
        "decision": "approved" if status["approved"] else "rejected",
        "automated_flags": status["flags"],
        "violations": status["violations"],
        "feature_scores": {
            "nsfw": features.get("nsfw_score"),
            "violence": features.get("violence_score"),
            "hate_speech": features.get("hate_speech_score")
        }
    }

    # Store in compliance database
    # save_to_audit_log(audit_log)
    print(f"Compliance check logged: {audit_log}")

# Batch compliance audit
def audit_content_library(collection_id):
    """
    Review existing content for compliance
    """
    # Get all items
    items = client.list(collection_id=collection_id)

    violations = []
    for item in items:
        status = check_compliance(item.content_url)

        if not status["approved"]:
            violations.append({
                "item_id": item.id,
                "violations": status["violations"]
            })

    print(f"Audit complete: {len(violations)} violations found")
    return violations

RESTRICTED_BRANDS = ["competitor-brand-1", "competitor-brand-2"]

# Usage
compliance_result = check_compliance(
    content_url="https://example.com/user-video.mp4",
    content_type="video"
)

if compliance_result["approved"]:
    print("Content approved for publication")
elif compliance_result["requires_review"]:
    print(f"Content flagged for review: {compliance_result['flags']}")
else:
    print(f"Content rejected: {compliance_result['violations']}")

from mixpeek import Mixpeek
import numpy as np

client = Mixpeek(token="your_token")

# Create deepfake detection pipeline
deepfake_collection = client.collections.create(
    name="deepfake_detection",
    feature_extractors=[
        "video-embedding",           # Overall video similarity
        "face-embedding",            # Facial characteristics
        "deepfake-detection",        # Specialized deepfake model
        "audio-embedding",           # Voice characteristics
        "lip-sync-analysis",         # Audio-visual sync
        "frame-consistency-check"    # Temporal anomalies
    ]
)

def detect_deepfake(video_url, known_authentic_videos=None):
    """
    Analyze video for signs of manipulation
    """
    # Process suspect video
    result = client.index(
        collection_id=deepfake_collection.id,
        video_url=video_url,
        process_async=False  # Wait for immediate results
    )

    deepfake_indicators = {
        "is_deepfake": False,
        "confidence": 0.0,
        "flags": [],
        "authenticity_score": 1.0
    }

    # Check deepfake detection model score
    deepfake_score = result.features.get("deepfake_probability", 0)
    if deepfake_score > 0.7:
        deepfake_indicators["is_deepfake"] = True
        deepfake_indicators["confidence"] = deepfake_score
        deepfake_indicators["flags"].append("High deepfake probability")

    # Check lip-sync analysis
    lip_sync_score = result.features.get("lip_sync_accuracy", 1.0)
    if lip_sync_score < 0.6:
        deepfake_indicators["flags"].append("Poor audio-visual synchronization")
        deepfake_indicators["authenticity_score"] *= 0.7

    # Check frame consistency
    frame_consistency = result.features.get("frame_consistency_score", 1.0)
    if frame_consistency < 0.7:
        deepfake_indicators["flags"].append("Temporal inconsistencies detected")
        deepfake_indicators["authenticity_score"] *= 0.8

    # Compare against known authentic videos
    if known_authentic_videos:
        for auth_video in known_authentic_videos:
            # Search for similarity
            similarity_results = client.search(
                collection_id=deepfake_collection.id,
                query_video_url=auth_video,
                top_k=1
            )

            if similarity_results and similarity_results[0].score > 0.95:
                # Very similar to authentic - likely authentic
                deepfake_indicators["authenticity_score"] = max(
                    deepfake_indicators["authenticity_score"],
                    similarity_results[0].score
                )

    # Check for manipulation artifacts
    artifacts = result.features.get("manipulation_artifacts", [])
    if artifacts:
        deepfake_indicators["flags"].extend([
            f"Artifact detected: {a['type']}" for a in artifacts
        ])
        deepfake_indicators["authenticity_score"] *= 0.6

    # Final determination
    if deepfake_indicators["authenticity_score"] < 0.5:
        deepfake_indicators["is_deepfake"] = True
        deepfake_indicators["confidence"] = 1 - deepfake_indicators["authenticity_score"]

    return deepfake_indicators

# Batch verification of video library
def verify_video_library(video_urls, authentic_reference_urls):
    """
    Check multiple videos for authenticity
    """
    results = []

    for url in video_urls:
        detection = detect_deepfake(url, authentic_reference_urls)
        results.append({
            "video_url": url,
            "is_authentic": not detection["is_deepfake"],
            "confidence": detection["confidence"],
            "flags": detection["flags"]
        })

    # Summary statistics
    deepfakes_detected = sum(1 for r in results if not r["is_authentic"])
    print(f"Analyzed {len(results)} videos")
    print(f"Suspected deepfakes: {deepfakes_detected}")

    return results

# Content provenance tracking
def register_authentic_video(video_url, metadata):
    """
    Register video as authentic with provenance data
    """
    result = client.index(
        collection_id=deepfake_collection.id,
        video_url=video_url,
        metadata={
            "is_authentic": True,
            "source": metadata["source"],
            "capture_device": metadata.get("device"),
            "capture_time": metadata.get("timestamp"),
            "creator": metadata.get("creator"),
            "signature": metadata.get("cryptographic_signature")
        }
    )

    print(f"Registered authentic video: {result.id}")
    return result.id

# Usage
suspect_video = "https://example.com/suspect-video.mp4"
authentic_refs = [
    "https://example.com/known-authentic-1.mp4",
    "https://example.com/known-authentic-2.mp4"
]

detection_result = detect_deepfake(suspect_video, authentic_refs)

if detection_result["is_deepfake"]:
    print(f"⚠️ Deepfake detected (confidence: {detection_result['confidence']:.1%})")
    print(f"Flags: {', '.join(detection_result['flags'])}")
else:
    print(f"✓ Video appears authentic (score: {detection_result['authenticity_score']:.1%})")

from mixpeek import Mixpeek
from collections import defaultdict
import numpy as np

client = Mixpeek(token="your_token")

# Index video catalog
video_collection = client.collections.create(
    name="video_recommendations",
    feature_extractors=[
        "video-embedding",      # Content similarity
        "scene-splitting",      # Temporal features
        "transcription"         # Semantic content
    ]
)

# User viewing history storage
user_viewing_history = defaultdict(list)  # user_id -> [video_ids]

def track_video_view(user_id, video_id, watch_time_seconds, total_duration):
    """
    Record user viewing behavior
    """
    completion_rate = watch_time_seconds / total_duration

    user_viewing_history[user_id].append({
        "video_id": video_id,
        "watch_time": watch_time_seconds,
        "completion_rate": completion_rate,
        "timestamp": "2024-01-15T10:30:00Z",
        "engagement_score": completion_rate  # Higher if fully watched
    })

def get_content_recommendations(user_id, num_recommendations=10):
    """
    Generate recommendations using content-based filtering
    """
    # Get user's viewing history
    history = user_viewing_history.get(user_id, [])

    if not history:
        # Cold start: return trending/popular videos
        return get_trending_videos(num_recommendations)

    # Find videos user engaged with most
    high_engagement = [
        v for v in history
        if v["engagement_score"] > 0.7
    ]

    # Sort by recency and engagement
    high_engagement.sort(
        key=lambda x: x["engagement_score"],
        reverse=True
    )

    # Get embeddings of top watched videos
    seed_videos = high_engagement[:5]

    recommendations = []
    for seed in seed_videos:
        # Find similar videos
        similar = client.search(
            collection_id=video_collection.id,
            query=f"video_id:{seed['video_id']}",  # Use video as query
            top_k=20
        )

        # Filter out already watched
        watched_ids = set(v["video_id"] for v in history)
        new_videos = [
            s for s in similar
            if s.metadata["video_id"] not in watched_ids
        ]

        recommendations.extend(new_videos)

    # Deduplicate and rank by similarity score
    seen = set()
    unique_recommendations = []
    for rec in recommendations:
        if rec.id not in seen:
            seen.add(rec.id)
            unique_recommendations.append(rec)

    unique_recommendations.sort(key=lambda x: x.score, reverse=True)

    return unique_recommendations[:num_recommendations]

def get_collaborative_recommendations(user_id, num_recommendations=10):
    """
    Find similar users and recommend their favorites
    """
    # Find users with similar viewing patterns
    current_user_videos = set(
        v["video_id"] for v in user_viewing_history[user_id]
    )

    similar_users = []
    for other_user_id, other_history in user_viewing_history.items():
        if other_user_id == user_id:
            continue

        other_videos = set(v["video_id"] for v in other_history)

        # Calculate Jaccard similarity
        intersection = len(current_user_videos & other_videos)
        union = len(current_user_videos | other_videos)
        similarity = intersection / union if union > 0 else 0

        if similarity > 0.2:  # At least 20% overlap
            similar_users.append({
                "user_id": other_user_id,
                "similarity": similarity
            })

    # Get videos from similar users
    recommendations = defaultdict(float)
    for similar_user in sorted(similar_users, key=lambda x: x["similarity"], reverse=True)[:10]:
        for video in user_viewing_history[similar_user["user_id"]]:
            if video["video_id"] not in current_user_videos:
                # Weight by user similarity and engagement
                score = similar_user["similarity"] * video["engagement_score"]
                recommendations[video["video_id"]] += score

    # Sort by aggregated score
    sorted_recs = sorted(
        recommendations.items(),
        key=lambda x: x[1],
        reverse=True
    )

    return [video_id for video_id, score in sorted_recs[:num_recommendations]]

def get_hybrid_recommendations(user_id, num_recommendations=10):
    """
    Combine content-based and collaborative filtering
    """
    # Get both types of recommendations
    content_recs = get_content_recommendations(user_id, num_recommendations * 2)
    collab_recs = get_collaborative_recommendations(user_id, num_recommendations)

    # Merge and deduplicate
    # Weighted: 70% content-based, 30% collaborative
    final_recs = []

    # Add top content-based recs (70%)
    final_recs.extend(content_recs[:int(num_recommendations * 0.7)])

    # Add collaborative recs (30%)
    seen_ids = set(r.metadata["video_id"] for r in final_recs)
    collab_count = 0
    for video_id in collab_recs:
        if video_id not in seen_ids:
            # Fetch video details
            final_recs.append({"video_id": video_id})
            collab_count += 1
            if collab_count >= int(num_recommendations * 0.3):
                break

    return final_recs[:num_recommendations]

def get_trending_videos(num_videos=10):
    """
    Cold start: Get popular videos
    """
    # Query for most viewed videos in last 7 days
    return client.search(
        collection_id=video_collection.id,
        filters={
            "upload_date": {"$gte": "2024-01-08"},
            "view_count": {"$gte": 1000}
        },
        sort_by="view_count",
        top_k=num_videos
    )

# Usage
user_id = "user_123"

# Track user behavior
track_video_view(
    user_id=user_id,
    video_id="video_456",
    watch_time_seconds=180,
    total_duration=200
)

# Get personalized recommendations
recommendations = get_hybrid_recommendations(user_id, num_recommendations=10)

print(f"Top recommendations for {user_id}:")
for i, rec in enumerate(recommendations, 1):
    print(f"{i}. Video: {rec.get('video_id', 'N/A')}")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Create collection with transcription
transcript_collection = client.collections.create(
    name="searchable_video_transcripts",
    feature_extractors=[
        "transcription",        # Speech-to-text with timestamps
        "video-embedding",      # Visual content
        "scene-splitting"       # Scene boundaries
    ],
    config={
        "transcription": {
            "language": "en",                    # Specify language
            "word_timestamps": True,             # Get word-level timestamps
            "speaker_diarization": True,         # Identify different speakers
            "punctuation": True,                 # Add punctuation
            "paragraph_segmentation": True       # Break into paragraphs
        }
    }
)

# Index video with transcription
def index_video_with_transcript(video_url, metadata):
    result = client.index(
        collection_id=transcript_collection.id,
        video_url=video_url,
        metadata=metadata,
        process_async=False  # Wait for transcription
    )

    # Transcript is automatically generated and indexed
    transcript = result.features.get("transcript", {})

    print(f"Indexed video: {result.id}")
    print(f"Transcript length: {len(transcript.get('text', ''))} characters")
    print(f"Duration: {transcript.get('duration', 0)} seconds")

    return result

# Search within transcripts
def search_transcripts(query, collection_id=None):
    """
    Search for keywords or concepts in video transcripts
    """
    results = client.search(
        collection_id=collection_id or transcript_collection.id,
        query=query,
        search_fields=["transcript"],  # Only search transcript text
        top_k=20
    )

    # Process results
    matches = []
    for result in results:
        transcript = result.features.get("transcript", {})

        # Find matching segments with timestamps
        for segment in transcript.get("segments", []):
            if query.lower() in segment["text"].lower():
                matches.append({
                    "video_id": result.metadata.get("video_id"),
                    "timestamp": segment["start"],
                    "end_time": segment["end"],
                    "text": segment["text"],
                    "speaker": segment.get("speaker"),
                    "relevance_score": result.score
                })

    return matches

# Get full transcript with timestamps
def get_video_transcript(video_id):
    """
    Retrieve complete transcript for a video
    """
    # Fetch video item
    video = client.get(
        collection_id=transcript_collection.id,
        item_id=video_id
    )

    transcript = video.features.get("transcript", {})

    formatted_transcript = []
    for segment in transcript.get("segments", []):
        formatted_transcript.append({
            "timestamp": format_timestamp(segment["start"]),
            "text": segment["text"],
            "speaker": segment.get("speaker", "Unknown")
        })

    return formatted_transcript

def format_timestamp(seconds):
    """
    Convert seconds to MM:SS format
    """
    minutes = int(seconds // 60)
    secs = int(seconds % 60)
    return f"{minutes:02d}:{secs:02d}"

# Semantic search in transcripts
def search_by_topic(topic, collection_id=None):
    """
    Find videos discussing a topic (semantic, not keyword match)
    """
    results = client.search(
        collection_id=collection_id or transcript_collection.id,
        query=f"discussion about {topic}",
        search_fields=["transcript"],
        hybrid_search=True,  # Combine semantic + keyword
        top_k=10
    )

    return [
        {
            "video_id": r.metadata.get("video_id"),
            "title": r.metadata.get("title"),
            "relevance": r.score,
            "transcript_preview": r.features.get("transcript", {}).get("text", "")[:200] + "..."
        }
        for r in results
    ]

# Multi-video transcript search
def search_across_library(query):
    """
    Search transcripts across entire video library
    """
    matches = search_transcripts(query, transcript_collection.id)

    # Group by video
    by_video = {}
    for match in matches:
        video_id = match["video_id"]
        if video_id not in by_video:
            by_video[video_id] = []
        by_video[video_id].append(match)

    # Format results
    results = []
    for video_id, video_matches in by_video.items():
        results.append({
            "video_id": video_id,
            "match_count": len(video_matches),
            "timestamps": [m["timestamp"] for m in video_matches],
            "snippets": [m["text"] for m in video_matches]
        })

    # Sort by match count
    results.sort(key=lambda x: x["match_count"], reverse=True)

    return results

# Usage
# Index a video
video_result = index_video_with_transcript(
    video_url="https://example.com/lecture.mp4",
    metadata={
        "video_id": "lecture_001",
        "title": "Introduction to Machine Learning",
        "course": "CS101"
    }
)

# Search for specific keyword
keyword_matches = search_transcripts("gradient descent")

print(f"Found {len(keyword_matches)} mentions of 'gradient descent':")
for match in keyword_matches[:5]:
    print(f"  At {format_timestamp(match['timestamp'])}: {match['text']}")

# Semantic topic search
topic_results = search_by_topic("neural networks and deep learning")

print(f"\nVideos discussing neural networks:")
for result in topic_results[:3]:
    print(f"  {result['title']} (relevance: {result['relevance']:.2%})")

# Get full transcript
full_transcript = get_video_transcript("lecture_001")

print(f"\nFull transcript (first 3 segments):")
for segment in full_transcript[:3]:
    print(f"[{segment['timestamp']}] {segment['speaker']}: {segment['text']}")

from mixpeek import Mixpeek
from collections import defaultdict
import numpy as np

client = Mixpeek(token="your_token")

# Create image similarity collection
product_images = client.collections.create(
    name="product_image_similarity",
    feature_extractors=[
        "image-embedding",      # Visual similarity
        "object-detection",     # Product attributes
        "color-extraction",     # Dominant colors
        "image-quality-check"   # Quality metrics
    ]
)

# Index product catalog
def index_product_images(products):
    for product in products:
        client.index(
            collection_id=product_images.id,
            content={
                "image_url": product["image_url"],
                "metadata": {
                    "sku": product["sku"],
                    "category": product["category"],
                    "price": product["price"],
                    "supplier_id": product.get("supplier_id")
                }
            }
        )

# Find duplicate product images
def find_duplicate_images(similarity_threshold=0.95):
    """
    Detect near-duplicate product images
    """
    all_products = client.list(collection_id=product_images.id)

    duplicates = []
    checked = set()

    for product in all_products:
        if product.id in checked:
            continue

        # Search for visually similar images
        similar = client.search(
            collection_id=product_images.id,
            query_image_url=product.content["image_url"],
            top_k=10
        )

        # Find near-duplicates
        for match in similar:
            if match.id == product.id:
                continue  # Skip self

            if match.score >= similarity_threshold:
                duplicates.append({
                    "product_1": product.metadata["sku"],
                    "product_2": match.metadata["sku"],
                    "similarity": match.score,
                    "action": "review" if match.score < 0.99 else "likely_duplicate"
                })
                checked.add(match.id)

    print(f"Found {len(duplicates)} potential duplicates")
    return duplicates

# Cluster similar products
def cluster_similar_products(category, num_clusters=10):
    """
    Group products into visual style clusters
    """
    # Get all products in category
    category_products = client.search(
        collection_id=product_images.id,
        filters={"category": category},
        top_k=1000
    )

    # Extract embeddings
    embeddings = [p.embedding for p in category_products]

    # Cluster using k-means (simplified)
    from sklearn.cluster import KMeans

    kmeans = KMeans(n_clusters=num_clusters, random_state=42)
    cluster_labels = kmeans.fit_predict(embeddings)

    # Group products by cluster
    clusters = defaultdict(list)
    for product, label in zip(category_products, cluster_labels):
        clusters[label].append({
            "sku": product.metadata["sku"],
            "image_url": product.content["image_url"],
            "price": product.metadata["price"]
        })

    # Analyze clusters
    for cluster_id, products in clusters.items():
        print(f"Cluster {cluster_id}: {len(products)} products")
        # Get representative (centroid)
        # ...

    return clusters

# Quality control check
def check_image_quality(image_url):
    """
    Verify product image meets quality standards
    """
    result = client.index(
        collection_id=product_images.id,
        content={"image_url": image_url},
        process_async=False
    )

    quality_issues = []

    # Check quality metrics
    quality_score = result.features.get("quality_score", 1.0)
    if quality_score < 0.7:
        quality_issues.append("Low image quality")

    # Check resolution
    resolution = result.features.get("resolution", {})
    if resolution.get("width", 0) < 800 or resolution.get("height", 0) < 800:
        quality_issues.append("Resolution too low (< 800px)")

    # Check background
    background = result.features.get("background_type", "")
    if background != "white" and background != "transparent":
        quality_issues.append(f"Non-standard background: {background}")

    # Check for watermarks
    if result.features.get("has_watermark", False):
        quality_issues.append("Watermark detected")

    return {
        "passes": len(quality_issues) == 0,
        "issues": quality_issues,
        "quality_score": quality_score
    }

# Find visually similar alternatives
def find_similar_products(sku, max_price_difference=50):
    """
    Find products that look similar for substitution/upsell
    """
    # Get product
    product = client.search(
        collection_id=product_images.id,
        filters={"sku": sku},
        top_k=1
    )[0]

    # Find similar visually
    similar = client.search(
        collection_id=product_images.id,
        query_image_url=product.content["image_url"],
        filters={
            "category": product.metadata["category"],
            "price": {
                "$gte": product.metadata["price"] - max_price_difference,
                "$lte": product.metadata["price"] + max_price_difference
            }
        },
        top_k=10
    )

    alternatives = []
    for match in similar:
        if match.metadata["sku"] == sku:
            continue  # Skip self

        alternatives.append({
            "sku": match.metadata["sku"],
            "similarity": match.score,
            "price": match.metadata["price"],
            "price_diff": match.metadata["price"] - product.metadata["price"]
        })

    return alternatives

# Usage
# Find duplicates
duplicates = find_duplicate_images(similarity_threshold=0.95)

print("Duplicate product images:")
for dup in duplicates[:5]:
    print(f"  {dup['product_1']} ≈ {dup['product_2']} ({dup['similarity']:.1%} similar)")

# Cluster products by visual style
clusters = cluster_similar_products(category="dresses", num_clusters=5)

print(f"\nFound {len(clusters)} visual style clusters")

# Quality check
quality_result = check_image_quality("https://example.com/product-image.jpg")

if not quality_result["passes"]:
    print(f"\nQuality issues: {', '.join(quality_result['issues'])}")

# Find alternatives
alternatives = find_similar_products(sku="SKU-12345", max_price_difference=30)

print(f"\nSimilar products for upsell:")
for alt in alternatives[:3]:
    price_diff = f"+{alt['price_diff']:.2f}" if alt['price_diff'] >= 0 else f"{alt['price_diff']:.2f}"
    print(f"  {alt['sku']}: {alt['similarity']:.1%} similar, \${price_diff}")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Step 1: Create a collection with video extractors
collection = client.collections.create(
    name="video_search_engine",
    feature_extractors=[
        "scene-splitting",
        "video-embedding",
        "transcription"
    ]
)

# Step 2: Ingest videos from your storage
client.ingest(
    collection_id=collection.id,
    source="s3://my-video-library/",
    process_async=True
)

# Step 3: Search with natural language
results = client.search(
    collection_id=collection.id,
    query="person giving a presentation about AI",
    top_k=10
)

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# Create collection with face identity extraction
collection = client.collections.create(
    name="face_identity_search",
    feature_extractors=["face-identity"]
)

# Ingest images/videos - faces are automatically detected and embedded
client.ingest(
    collection_id=collection.id,
    source="s3://my-photos/",
    process_async=True
)

# Search by uploading a photo of a face
results = client.search(
    collection_id=collection.id,
    query_image_url="https://example.com/query-face.jpg",
    top_k=10,
    filters={"type": "face"}
)

from mixpeek import Mixpeek
import time

client = Mixpeek(token="your_token")

# Sync processing: Wait for result
def sync_example(video_url, collection_id):
    """
    Use for small files or development
    """
    print("Starting sync processing...")
    start = time.time()

    try:
        result = client.index(
            collection_id=collection_id,
            video_url=video_url,
            process_async=False,  # Blocks until complete
            timeout=60  # Max wait time in seconds
        )

        print(f"Completed in {time.time() - start}s")
        print(f"Features extracted: {result.features}")
        return result

    except TimeoutError:
        print("Processing took too long, timed out")
        # Fallback: Retry with async
        return async_example(video_url, collection_id)

# Async processing: Non-blocking
def async_example(video_url, collection_id):
    """
    Use for large files or batch processing
    """
    print("Starting async processing...")

    # Submit job (returns immediately)
    job = client.index(
        collection_id=collection_id,
        video_url=video_url,
        process_async=True,
        webhook_url="https://myapp.com/webhook/complete"  # Get notified
    )

    print(f"Job submitted: {job.id}")

    # Continue with other work...
    # Results will come via webhook

    return job.id

# Polling async job status (if not using webhooks)
def wait_for_async_job(job_id, timeout=600):
    """
    Poll job until complete
    """
    start = time.time()
    poll_interval = 5  # Start with 5s

    while time.time() - start < timeout:
        status = client.jobs.get(job_id)

        if status.status == "completed":
            return status.result

        elif status.status == "failed":
            raise Exception(f"Job failed: {status.error}")

        # Exponential backoff
        print(f"Status: {status.status}, waiting {poll_interval}s...")
        time.sleep(poll_interval)
        poll_interval = min(poll_interval * 1.5, 30)  # Max 30s

    raise TimeoutError(f"Job {job_id} didn't complete in {timeout}s")

# Decision tree
def smart_index(file_url, collection_id, file_size_mb=None):
    """
    Automatically choose async vs sync based on file size
    """
    # Small file or unknown size: try sync
    if file_size_mb is None or file_size_mb < 10:
        try:
            return sync_example(file_url, collection_id)
        except TimeoutError:
            print("Sync timed out, falling back to async")
            return async_example(file_url, collection_id)

    # Large file: always async
    else:
        return async_example(file_url, collection_id)

# Batch processing: Always async
def batch_index(file_urls, collection_id):
    """
    For multiple files, always use async
    """
    job_ids = []

    for url in file_urls:
        job = client.index(
            collection_id=collection_id,
            video_url=url,
            process_async=True
        )
        job_ids.append(job.id)

    print(f"Submitted {len(job_ids)} jobs")
    return job_ids

# Usage
# Small image: sync is fine
sync_example("https://example.com/small-image.jpg", "images")

# Large video: use async
async_example("https://example.com/large-video.mp4", "videos")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# GOOD: Use metadata filters + hybrid search
results = client.search(
    collection_id="products",
    query="comfortable running shoes",
    filters={
        "category": "footwear",
        "price": {"$lt": 200},
        "in_stock": True
    },
    hybrid_search=True,  # Combine vector + keyword
    min_score=0.7  # Filter low-confidence results
)

# BAD: No filters, short query
# results = client.search(collection_id="products", query="shoes")

from mixpeek import Mixpeek

client = Mixpeek(token="your_token")

# COST-OPTIMIZED: Scene detection, selective extractors
collection = client.collections.create(
    name="video_library_optimized",
    feature_extractors=[
        "scene-splitting",  # Only process scene changes
        "video-embedding"   # Only essential embeddings
    ],
    config={
        "scene_threshold": 0.3,  # Adjust sensitivity
        "max_scenes_per_video": 50,  # Cap scenes for very long videos
        "resolution": "480p"  # Lower res for initial processing
    }
)

# Process asynchronously in batches
client.ingest(
    collection_id=collection.id,
    source="s3://my-videos/",
    batch_size=100,
    process_async=True  # Don't block, process in background
)

# EXPENSIVE: Per-frame processing, all extractors
# Don't do this for large libraries!

from mixpeek import Mixpeek
import redis
import hashlib
import json

client = Mixpeek(token="your_token")
cache = redis.Redis(host='localhost', port=6379, db=0)

def search_with_cache(collection_id, query, filters=None, top_k=10, cache_ttl=3600):
    """
    Search with Redis caching
    """
    # Create cache key from query parameters
    cache_key = hashlib.md5(
        json.dumps({
            "collection": collection_id,
            "query": query,
            "filters": filters,
            "top_k": top_k
        }, sort_keys=True).encode()
    ).hexdigest()

    # Check cache
    cached_result = cache.get(cache_key)
    if cached_result:
        print("Cache hit!")
        return json.loads(cached_result)

    # Cache miss - query Mixpeek
    print("Cache miss - querying Mixpeek")
    results = client.search(
        collection_id=collection_id,
        query=query,
        filters=filters,
        top_k=top_k
    )

    # Serialize and cache results
    serialized = json.dumps([
        {"id": r.id, "score": r.score, "metadata": r.metadata}
        for r in results
    ])
    cache.setex(cache_key, cache_ttl, serialized)

    return results

# Optimized search with metadata pre-filtering
def fast_search(collection_id, query, category=None, max_results=10):
    """
    Use filters to reduce search space before vector search
    """
    filters = {}

    # Filter by category first (index scan, very fast)
    if category:
        filters["category"] = category

    # Only search recent items (temporal filter)
    filters["created_at"] = {"$gte": "2024-01-01"}

    # Smaller top_k for faster search
    results = client.search(
        collection_id=collection_id,
        query=query,
        filters=filters,
        top_k=max_results,  # Request only what you need
        approximate=True  # Use ANN (default, but explicit)
    )

    return results

# Pre-warm cache for common queries
def warm_cache(collection_id, common_queries):
    """
    Pre-compute and cache results for popular queries
    """
    for query in common_queries:
        print(f"Warming cache: {query}")
        search_with_cache(collection_id, query)

# Usage
common_queries = [
    "wireless headphones",
    "running shoes",
    "laptop backpack"
]
warm_cache("products", common_queries)

from mixpeek import Mixpeek
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

client = Mixpeek(token="your_token")

# SLOW: Sequential sync processing
def slow_indexing(video_urls, collection_id):
    """
    Don't do this for many files!
    """
    start = time.time()
    for url in video_urls:
        result = client.index(
            collection_id=collection_id,
            video_url=url,
            process_async=False  # Blocks until complete
        )
        print(f"Indexed: {result.id}")

    print(f"Total time: {time.time() - start}s")

# FAST: Async parallel processing
def fast_indexing(video_urls, collection_id, max_workers=10):
    """
    Much faster - parallel async ingestion
    """
    start = time.time()

    def index_video(url):
        return client.index(
            collection_id=collection_id,
            video_url=url,
            process_async=True  # Returns immediately
        )

    # Process multiple files concurrently
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(index_video, url): url for url in video_urls}

        for future in as_completed(futures):
            url = futures[future]
            try:
                result = future.result()
                print(f"Started job: {result.job_id} for {url}")
            except Exception as e:
                print(f"Failed to start job for {url}: {e}")

    print(f"All jobs submitted in: {time.time() - start}s")

# FASTEST: Bulk S3 ingestion
def fastest_indexing(s3_bucket, prefix, collection_id):
    """
    Let Mixpeek handle parallelization
    """
    start = time.time()

    job = client.ingest(
        collection_id=collection_id,
        source=f"s3://{s3_bucket}/{prefix}",
        process_async=True,
        batch_size=500,  # Process in large batches
        config={
            # Only essential extractors
            "feature_extractors": [
                "video-embedding",
                "scene-splitting"
            ],
            # Optimize settings
            "scene_threshold": 0.4,  # Fewer scenes = faster
            "resolution": "720p"  # Lower res if acceptable
        }
    )

    print(f"Bulk job started: {job.id} in {time.time() - start}s")
    return job.id

# Monitor async jobs
def wait_for_completion(job_id, timeout=3600):
    """
    Wait for bulk job to complete
    """
    start = time.time()
    while time.time() - start < timeout:
        status = client.jobs.get(job_id)

        if status.status == "completed":
            print(f"Completed: {status.processed} files in {time.time() - start}s")
            return status

        print(f"Progress: {status.processed}/{status.total}")
        time.sleep(10)

    raise TimeoutError()

# Usage
urls = [f"https://example.com/video{i}.mp4" for i in range(100)]
fast_indexing(urls, "my_collection", max_workers=20)