Twelve Labsを使用した、AI駆動のスポンサーシップROI分析プラットフォームの構築

from twelvelabs import TwelveLabs
from twelvelabs.indexes import IndexesCreateRequestModelsItem

# マルチモーダル理解を備えたビデオインデックスを作成
index = client.indexes.create(
    index_name="sponsorship-roi-analysis",
    models=[
        IndexesCreateRequestModelsItem(
                model_name="pegasus1.2",
                model_options=["visual", "audio"],
        )
    ]
)

# 分析用にビデオをアップロード
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):
    """Sponsorshipをad_placementまたはin_game_placementに分類する"""

    # 分類ルールの定義
    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'
    }

    # コンテキストに基づく判定
    if context == 'commercial':
        return "ad_placement"

    if placement_type in ad_placement_types:
        return "ad_placement"

    # デフォルトはオーガニック（インゲームプレースメント）とする
    return "in_game_placement"

def calculate_placement_effectiveness(brand_data, video_duration):
    """0-100のプレースメント効果スコアを算出"""

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

    # 各プレースメントを分析
    for appearance in brand_data:
        start_time = appearance['timeline'][0]
        end_time = appearance['timeline'][1]
        duration = end_time - start_time

        # 高エンゲージメント瞬間（モーメント）の検出
        is_optimal = any(keyword in appearance['description'].lower()
                        for keyword in ['goal', 'celebration', 'replay',
                                       'highlight', 'scoring', 'win'])

        # 主要な視認性のチェック
        is_prominent = appearance['prominence'] == 'primary'

        # ポジティブセンチメントのチェック
        is_positive = appearance['sentiment_context'] == 'positive'

        # 高注視率のチェック
        is_high_attention = appearance['viewer_attention'] == 'high'

        # 品質スコア（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

    # 総合プレースメント効果 (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

# ビデオ内の高エンゲージメントモーメントを検出します
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"}]
    """
)

# ブランド出現データと比較します
missed_opportunities = []
for moment in high_engagement_moments:
    # このモーメントでブランドが表示されていたかをチェック
    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):
    """GPT-4を使用して包括的なROIインサイトを生成します"""

    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"):
    """市場シェアに基づく競合環境分析を生成します"""

    # 検出されたブランドが1つのみであっても、AIが完全な競合背景を生成します
    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):
    """進捗値を追跡しながら、複数の動画を並列処理で分析"""

    max_workers = min(4, len(video_ids))  # APIの同時リクエスト件数を制限

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # すべてのビデオ分析タスクを並列キューへ投入します
        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

        # 順不同で処理が完了したものから順次処理していきます
        for future in as_completed(future_to_video):
            result = future.result()
            individual_analyses.append(result)
            completed += 1

            # 進捗ゲージを更新：分析ワーク（80%分）、マージ（結合）ワーク（20%分）
            progress = int((completed / len(video_ids)) * 80)
            update_progress(job_id, progress, f'Completed {completed}/{len(video_ids)} videos')

    # 時間の連続性を維持しながら集計結果を統合します
    return combine_video_analyses(individual_analyses)

def combine_video_analyses(individual_analyses):
    """時間的なオフセットを織り込み、複数の映像ソースを一元的な時間軸データへ集約する"""

    cumulative_duration = 0
    all_detections = []

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

        # すべてのタイムスタンプを累積時間でオフセット調整します
        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)

        # 次のビデオに向け、現在までに処理されたビデオの全体時間を加算・保持します
        cumulative_duration += video_duration

    return {
        'combined_duration': cumulative_duration,
        'raw_detections': all_detections  # 統合されたタイムラインデータ
    }