Twelve Labs와 함께하는 AI 기반 후원 ROI 분석 플랫폼 구축

from twelvelabs import TwelveLabs
from twelvelabs.indexes import IndexesCreateRequestModelsItem

# Create video index with multimodal understanding
index = client.indexes.create(
    index_name="sponsorship-roi-analysis",
    models=[
        IndexesCreateRequestModelsItem(
                model_name="pegasus1.2",
                model_options=["visual", "audio"],
        )
    ]
)

# Upload video for analysis
task = client.tasks.create(
    index_id="<YOUR_INDEX_ID>",
    video_file=video_file,
    language="en"
)

brand_analysis_prompt = """
Analyze this video for comprehensive brand sponsorship measurement.

Focus on these brands: Nike, Adidas, Gatorade

IMPORTANT: Categorize each appearance into:
1. AD PLACEMENTS ("ad_placement"):
   - CTV commercials, digital overlays, squeeze ads

2. IN-GAME PLACEMENTS ("in_game_placement"):
   - Jersey sponsors, stadium signage, product placements

For EACH brand appearance, provide:
- timeline: [start_time, end_time] in seconds
- brand: exact brand name
- type: "logo", "jersey_sponsor", "stadium_signage", "ctv_ad", etc.
- sponsorship_category: "ad_placement" or "in_game_placement"
- prominence: "primary", "secondary", "background"
- context: "game_action", "celebration", "replay", etc.
- sentiment_context: "positive", "neutral", "negative"
- viewer_attention: "high", "medium", "low"

Return ONLY a JSON array.
"""

# Generate structured analysis
result = client.analyze(
    video_id="<YOUR_VIDEO_ID>",
    prompt=brand_analysis_prompt,
    temperature=0.1  # Low temperature for factual extraction
)

# Parse structured JSON response
brand_appearances = json.loads(result.data)

def categorize_sponsorship_placement(placement_type, context):
    """Categorize sponsorship into ad_placement or in_game_placement"""

    # Define categorization rules
    ad_placement_types = {
        'digital_overlay', 'ctv_ad', 'overlay_ad',
        'squeeze_ad', 'commercial'
    }

    in_game_placement_types = {
        'logo', 'jersey_sponsor', 'stadium_signage',
        'product_placement', 'audio_mention'
    }

    # Context-dependent decisions
    if context == 'commercial':
        return "ad_placement"

    if placement_type in ad_placement_types:
        return "ad_placement"

    # Default to organic in-game placement
    return "in_game_placement"

def calculate_placement_effectiveness(brand_data, video_duration):
    """Calculate 0-100 placement effectiveness score"""

    metrics = {
        'optimal_placements': 0,
        'suboptimal_placements': 0,
        'placement_score': 0.0
    }

    # Analyze each placement
    for appearance in brand_data:
        start_time = appearance['timeline'][0]
        end_time = appearance['timeline'][1]
        duration = end_time - start_time

        # High-engagement moment detection
        is_optimal = any(keyword in appearance['description'].lower()
                        for keyword in ['goal', 'celebration', 'replay',
                                       'highlight', 'scoring', 'win'])

        # Primary visibility check
        is_prominent = appearance['prominence'] == 'primary'

        # Positive sentiment check
        is_positive = appearance['sentiment_context'] == 'positive'

        # High attention check
        is_high_attention = appearance['viewer_attention'] == 'high'

        # Calculate quality score (0-100)
        quality_factors = [is_optimal, is_prominent, is_positive, is_high_attention]
        quality_score = (sum(quality_factors) / len(quality_factors)) * 100

        if quality_score >= 75:
            metrics['optimal_placements'] += 1
        else:
            metrics['suboptimal_placements'] += 1

    # Overall placement effectiveness (0-100)
    total_placements = metrics['optimal_placements'] + metrics['suboptimal_placements']
    if total_placements > 0:
        metrics['placement_score'] = (metrics['optimal_placements'] / total_placements) * 100

    return metrics

# Detect high-engagement moments in video
high_engagement_moments = client.analyze(
    video_id=video_id,
    prompt="""
    List ALL high-engagement moments with timestamps:
    - Goals/scores
    - Celebrations
    - Replays of key plays
    - Championship moments
    - Emotional peaks

    Return: [{"timestamp": 45.2, "description": "Game-winning goal"}]
    """
)

# Compare against brand appearances
missed_opportunities = []
for moment in high_engagement_moments:
    # Check if brand was visible during this moment
    brand_visible = any(
        appearance['timeline'][0] <= moment['timestamp'] <= appearance['timeline'][1]
        for appearance in brand_appearances
    )

    if not brand_visible:
        missed_opportunities.append({
            'timestamp': moment['timestamp'],
            'description': moment['description'],
            'potential_value': 'HIGH'
        })

def generate_ai_roi_insights(brand_name, appearances, placement_metrics, brand_intelligence):
    """Generate comprehensive ROI insights using GPT-4"""

    prompt = f"""
    You are a sponsorship ROI analyst. Analyze this brand's performance:

    BRAND: {brand_name}
    APPEARANCES: {len(appearances)}
    PLACEMENT EFFECTIVENESS: {placement_metrics['placement_score']}/100
    OPTIMAL PLACEMENTS: {placement_metrics['optimal_placements']}
    SUBOPTIMAL PLACEMENTS: {placement_metrics['suboptimal_placements']}

    MARKET CONTEXT: {brand_intelligence.get('industry', 'sports')}

    Provide analysis in this JSON structure:
    {{
        "placement_effectiveness_score": 0-100,
        "roi_assessment": {{
            "value_rating": "excellent|good|fair|poor",
            "cost_efficiency": 0-10,
            "exposure_quality": 0-10,
            "audience_reach": 0-10
        }},
        "recommendations": {{
            "immediate_actions": ["specific action 1", "action 2"],
            "future_strategy": ["strategic recommendation"],
            "optimal_moments": ["when to appear"],
            "avoid_these": ["what to avoid"]
        }},
        "executive_summary": "2-3 sentence strategic overview"
    }}
    """

    response = openai_client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a data-driven sponsorship analyst."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,
        max_tokens=1200
    )

    return json.loads(response.choices[0].message.content)

def generate_competitive_analysis(detected_brands, video_context="sports event"):
    """Generate competitive landscape with market share"""

    # Even if only one brand detected, AI generates full competitive context
    competitive_prompt = f"""
    Analyze the competitive landscape for brands detected in a {video_context}.

    DETECTED BRANDS: {', '.join(detected_brands)}

    REQUIREMENTS:
    1. Include the detected brand(s) with "detected_in_video": true
    2. Add 4-7 major market competitors with "detected_in_video": false
    3. Market shares should reflect actual market position

    EXAMPLE: If Nike detected → include Adidas, Under Armour, Puma, New Balance

    Return:
    {{
        "market_category": "Athletic Footwear & Apparel",
        "total_market_size": "$180B globally",
        "competitors": [
            {{
                "brand": "Nike",
                "market_share": 38.5,
                "prominence": "High",
                "positioning": "Innovation and athlete partnerships",
                "detected_in_video": true
            }},
            {{
                "brand": "Adidas",
                "market_share": 22.3,
                "prominence": "High",
                "positioning": "European heritage, football dominance",
                "detected_in_video": false
            }}
        ]
    }}
    """

    response = openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": competitive_prompt}],
        temperature=0.3
    )

    return json.loads(response.choices[0].message.content)

def analyze_multiple_videos_with_progress(video_ids, job_id, selected_brands):
    """Analyze multiple videos in parallel with progress tracking"""

    max_workers = min(4, len(video_ids))  # Limit concurrent API calls

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all video analysis tasks
        future_to_video = {
            executor.submit(analyze_single_video_parallel, video_id, selected_brands): video_id
            for video_id in video_ids
        }

        individual_analyses = []
        completed = 0

        # Process as they complete (not in order)
        for future in as_completed(future_to_video):
            result = future.result()
            individual_analyses.append(result)
            completed += 1

            # Update progress: 80% for analysis, 20% for combining
            progress = int((completed / len(video_ids)) * 80)
            update_progress(job_id, progress, f'Completed {completed}/{len(video_ids)} videos')

    # Combine results with temporal continuity
    return combine_video_analyses(individual_analyses)

def combine_video_analyses(individual_analyses):
    """Combine analyses with temporal offset for unified timeline"""

    cumulative_duration = 0
    all_detections = []

    for analysis in individual_analyses:
        video_duration = analysis['summary']['video_duration_minutes'] * 60

        # Offset all timestamps by cumulative duration
        for detection in analysis['raw_detections']:
            original_timeline = detection['timeline']
            detection['timeline'] = [
                original_timeline[0] + cumulative_duration,
                original_timeline[1] + cumulative_duration
            ]
            all_detections.append(detection)

        # Update cumulative duration for next video
        cumulative_duration += video_duration

    return {
        'combined_duration': cumulative_duration,
        'raw_detections': all_detections  # Unified timeline
    }