ステップ4：CloudWatchによる配信監視・管理システムの構築

import json
import boto3
import time
from decimal import Decimal
from datetime import datetime, timezone
 
dynamodb = boto3.resource('dynamodb')
sessions_table = dynamodb.Table('StreamingSessions')
metrics_table = dynamodb.Table('StreamingMetrics')
cloudwatch = boto3.client('cloudwatch')
 
def lambda_handler(event, context):
    """
    配信イベントの処理とDynamoDBへの記録
    """
 
    try:
        event_type = event.get('eventType')
        session_id = event.get('sessionId')
        channel_id = event.get('channelId')
        timestamp = int(time.time())
 
        if event_type == 'stream_start':
            handle_stream_start(session_id, channel_id, timestamp, event)
        elif event_type == 'stream_end':
            handle_stream_end(session_id, channel_id, timestamp, event)
        elif event_type == 'metrics_update':
            handle_metrics_update(session_id, channel_id, timestamp, event)
 
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Successfully processed {event_type}',
                'sessionId': session_id
            })
        }
 
    except Exception as e:
        print(f"Error processing event: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': str(e)
            })
        }
 
def handle_stream_start(session_id, channel_id, timestamp, event):
    """配信開始イベントの処理"""
 
    # 配信セッション情報をDynamoDBに記録
    sessions_table.put_item(
        Item={
            'sessionId': session_id,
            'timestamp': timestamp,
            'channelId': channel_id,
            'eventType': 'stream_start',
            'streamerUserId': event.get('streamerUserId'),
            'streamTitle': event.get('streamTitle', ''),
            'streamQuality': event.get('streamQuality', '720p'),
            'status': 'live'
        }
    )
 
    # CloudWatchカスタムメトリクスを送信
    cloudwatch.put_metric_data(
        Namespace='LiveStreaming/Sessions',
        MetricData=[
            {
                'MetricName': 'StreamsStarted',
                'Value': 1,
                'Unit': 'Count',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            }
        ]
    )
 
    print(f"Stream started: {session_id}")
 
def handle_stream_end(session_id, channel_id, timestamp, event):
    """配信終了イベントの処理"""
 
    duration = event.get('duration', 0)
    peak_viewers = event.get('peakViewers', 0)
    total_chat_messages = event.get('totalChatMessages', 0)
 
    # 配信終了情報をDynamoDBに記録
    sessions_table.put_item(
        Item={
            'sessionId': session_id,
            'timestamp': timestamp,
            'channelId': channel_id,
            'eventType': 'stream_end',
            'duration': duration,
            'peakViewers': peak_viewers,
            'totalChatMessages': total_chat_messages,
            'avgBitrate': event.get('avgBitrate', 0),
            'status': 'ended'
        }
    )
 
    # CloudWatchカスタムメトリクスを送信
    cloudwatch.put_metric_data(
        Namespace='LiveStreaming/Sessions',
        MetricData=[
            {
                'MetricName': 'StreamsEnded',
                'Value': 1,
                'Unit': 'Count',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            },
            {
                'MetricName': 'StreamDuration',
                'Value': duration,
                'Unit': 'Seconds',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            },
            {
                'MetricName': 'PeakViewers',
                'Value': peak_viewers,
                'Unit': 'Count',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            }
        ]
    )
 
    print(f"Stream ended: {session_id}, Duration: {duration}s, Peak viewers: {peak_viewers}")
 
def handle_metrics_update(session_id, channel_id, timestamp, event):
    """配信メトリクス更新の処理"""
 
    current_viewers = event.get('currentViewers', 0)
    bitrate = event.get('bitrate', 0)
    frame_rate = event.get('frameRate', 0)
    chat_messages_per_minute = event.get('chatMessagesPerMinute', 0)
 
    # メトリクスデータをDynamoDBに記録
    metrics_table.put_item(
        Item={
            'sessionId': session_id,
            'metricTimestamp': timestamp,
            'channelId': channel_id,
            'currentViewers': current_viewers,
            'bitrate': Decimal(str(bitrate)),
            'frameRate': Decimal(str(frame_rate)),
            'chatMessagesPerMinute': chat_messages_per_minute,
            'cpuUsage': event.get('cpuUsage', 0),
            'memoryUsage': event.get('memoryUsage', 0)
        }
    )
 
    # CloudWatchカスタムメトリクスを送信
    cloudwatch.put_metric_data(
        Namespace='LiveStreaming/Metrics',
        MetricData=[
            {
                'MetricName': 'CurrentViewers',
                'Value': current_viewers,
                'Unit': 'Count',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            },
            {
                'MetricName': 'StreamBitrate',
                'Value': bitrate,
                'Unit': 'Bits/Second',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            },
            {
                'MetricName': 'ChatActivity',
                'Value': chat_messages_per_minute,
                'Unit': 'Count/Minute',
                'Timestamp': datetime.fromtimestamp(timestamp, timezone.utc),
                'Dimensions': [
                    {
                        'Name': 'ChannelId',
                        'Value': channel_id
                    }
                ]
            }
        ]
    )
 
> **注意**: CloudWatchカスタムメトリクスには明示的なタイムスタンプを含めることで、データの正確性が向上します。タイムスタンプはUTC時間で指定する必要があります。
 
    print(f"Metrics updated for session: {session_id}")

import json
import boto3
from datetime import datetime, timedelta
from decimal import Decimal
import statistics
 
dynamodb = boto3.resource('dynamodb')
sessions_table = dynamodb.Table('StreamingSessions')
metrics_table = dynamodb.Table('StreamingMetrics')
s3 = boto3.client('s3')
 
def lambda_handler(event, context):
    """
    配信分析レポートの生成
    """
 
    try:
        report_type = event.get('reportType', 'daily')
        channel_id = event.get('channelId')
 
        if report_type == 'daily':
            report_data = generate_daily_report(channel_id)
        elif report_type == 'weekly':
            report_data = generate_weekly_report(channel_id)
        elif report_type == 'monthly':
            report_data = generate_monthly_report(channel_id)
        else:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'Invalid report type'})
            }
 
        # レポートをS3に保存
        report_key = f"reports/{channel_id}/{report_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        s3.put_object(
            Bucket='live-streaming-metadata-reports',
            Key=report_key,
            Body=json.dumps(report_data, default=decimal_default),
            ContentType='application/json'
        )
 
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'{report_type} report generated successfully',
                'reportUrl': f's3://live-streaming-metadata-reports/{report_key}',
                'reportData': report_data
            }, default=decimal_default)
        }
 
    except Exception as e:
        print(f"Error generating report: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
 
def generate_daily_report(channel_id):
    """日次レポートの生成"""
 
    # 過去24時間のデータを取得
    end_time = datetime.now()
    start_time = end_time - timedelta(days=1)
 
    # 配信セッション情報を取得
    sessions = get_sessions_in_period(channel_id, start_time, end_time)
 
    # メトリクスデータを取得
    metrics = get_metrics_in_period(channel_id, start_time, end_time)
 
    # レポートデータを生成
    report = {
        'reportType': 'daily',
        'channelId': channel_id,
        'period': {
            'start': start_time.isoformat(),
            'end': end_time.isoformat()
        },
        'summary': {
            'totalSessions': len(sessions),
            'totalStreamTime': sum([s.get('duration', 0) for s in sessions]),
            'avgSessionDuration': calculate_avg_duration(sessions),
            'maxConcurrentViewers': max([m.get('currentViewers', 0) for m in metrics]) if metrics else 0,
            'totalChatMessages': sum([s.get('totalChatMessages', 0) for s in sessions])
        },
        'sessions': sessions[:10],  # 最新10セッション
        'hourlyBreakdown': generate_hourly_breakdown(metrics),
        'qualityMetrics': calculate_quality_metrics(metrics),
        'chatAnalytics': analyze_chat_activity(sessions, metrics)
    }
 
    return report
 
def get_sessions_in_period(channel_id, start_time, end_time):
    """期間内の配信セッションを取得"""
 
    start_timestamp = int(start_time.timestamp())
    end_timestamp = int(end_time.timestamp())
 
    response = sessions_table.query(
        IndexName='ChannelIdIndex',
        KeyConditionExpression='channelId = :channel_id AND #ts BETWEEN :start_ts AND :end_ts',
        ExpressionAttributeNames={
            '#ts': 'timestamp'
        },
        ExpressionAttributeValues={
            ':channel_id': channel_id,
            ':start_ts': start_timestamp,
            ':end_ts': end_timestamp
        }
    )
 
    return response['Items']
 
def get_metrics_in_period(channel_id, start_time, end_time):
    """期間内のメトリクスデータを取得"""
 
    start_timestamp = int(start_time.timestamp())
    end_timestamp = int(end_time.timestamp())
 
    response = metrics_table.query(
        IndexName='ChannelIdIndex',
        KeyConditionExpression='channelId = :channel_id AND metricTimestamp BETWEEN :start_ts AND :end_ts',
        ExpressionAttributeValues={
            ':channel_id': channel_id,
            ':start_ts': start_timestamp,
            ':end_ts': end_timestamp
        }
    )
 
    return response['Items']
 
def calculate_avg_duration(sessions):
    """平均配信時間の計算"""
    durations = [s.get('duration', 0) for s in sessions if s.get('duration', 0) > 0]
    return statistics.mean(durations) if durations else 0
 
def generate_hourly_breakdown(metrics):
    """時間別の視聴者数推移を生成"""
    hourly_data = {}
 
    for metric in metrics:
        timestamp = metric['metricTimestamp']
        hour = datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:00')
 
        if hour not in hourly_data:
            hourly_data[hour] = {
                'viewers': [],
                'bitrate': [],
                'chatActivity': []
            }
 
        hourly_data[hour]['viewers'].append(metric.get('currentViewers', 0))
        hourly_data[hour]['bitrate'].append(float(metric.get('bitrate', 0)))
        hourly_data[hour]['chatActivity'].append(metric.get('chatMessagesPerMinute', 0))
 
    # 各時間の平均値を計算
    breakdown = []
    for hour, data in sorted(hourly_data.items()):
        breakdown.append({
            'hour': hour,
            'avgViewers': statistics.mean(data['viewers']) if data['viewers'] else 0,
            'avgBitrate': statistics.mean(data['bitrate']) if data['bitrate'] else 0,
            'avgChatActivity': statistics.mean(data['chatActivity']) if data['chatActivity'] else 0
        })
 
    return breakdown
 
def calculate_quality_metrics(metrics):
    """配信品質メトリクスの計算"""
    if not metrics:
        return {}
 
    bitrates = [float(m.get('bitrate', 0)) for m in metrics]
    frame_rates = [float(m.get('frameRate', 0)) for m in metrics]
 
    return {
        'avgBitrate': statistics.mean(bitrates) if bitrates else 0,
        'minBitrate': min(bitrates) if bitrates else 0,
        'maxBitrate': max(bitrates) if bitrates else 0,
        'avgFrameRate': statistics.mean(frame_rates) if frame_rates else 0,
        'bitrateStability': calculate_stability(bitrates),
        'qualityScore': calculate_quality_score(bitrates, frame_rates)
    }
 
def analyze_chat_activity(sessions, metrics):
    """チャット活動の分析"""
    total_messages = sum([s.get('totalChatMessages', 0) for s in sessions])
    total_duration = sum([s.get('duration', 0) for s in sessions])
 
    chat_rates = [m.get('chatMessagesPerMinute', 0) for m in metrics]
 
    return {
        'totalMessages': total_messages,
        'messagesPerMinute': total_messages / (total_duration / 60) if total_duration > 0 else 0,
        'peakChatRate': max(chat_rates) if chat_rates else 0,
        'avgChatRate': statistics.mean(chat_rates) if chat_rates else 0,
        'chatEngagementScore': calculate_chat_engagement(total_messages, total_duration)
    }
 
def calculate_stability(values):
    """数値の安定性を計算（変動係数）"""
    if not values or len(values) < 2:
        return 1.0
 
    mean_val = statistics.mean(values)
    if mean_val == 0:
        return 0.0
 
    std_dev = statistics.stdev(values)
    return max(0, 1 - (std_dev / mean_val))  # 変動が少ないほど1に近い
 
def calculate_quality_score(bitrates, frame_rates):
    """配信品質スコアの計算（0-100）"""
    if not bitrates or not frame_rates:
        return 0
 
    avg_bitrate = statistics.mean(bitrates)
    avg_framerate = statistics.mean(frame_rates)
    bitrate_stability = calculate_stability(bitrates)
    framerate_stability = calculate_stability(frame_rates)
 
    # 品質スコア計算（独自のアルゴリズム）
    bitrate_score = min(100, (avg_bitrate / 5000) * 40)  # 5Mbpsで40点
    framerate_score = min(100, (avg_framerate / 30) * 30)  # 30fpsで30点
    stability_score = (bitrate_stability + framerate_stability) * 15  # 安定性で30点
 
    return min(100, bitrate_score + framerate_score + stability_score)
 
def calculate_chat_engagement(total_messages, total_duration):
    """チャットエンゲージメントスコアの計算"""
    if total_duration == 0:
        return 0
 
    messages_per_minute = total_messages / (total_duration / 60)
 
    # エンゲージメントスコア（0-10）
    if messages_per_minute >= 50:
        return 10  # 非常に高い
    elif messages_per_minute >= 20:
        return 8   # 高い
    elif messages_per_minute >= 10:
        return 6   # 中程度
    elif messages_per_minute >= 5:
        return 4   # やや低い
    elif messages_per_minute >= 1:
        return 2   # 低い
    else:
        return 1   # 非常に低い
 
def decimal_default(obj):
    """DecimalをJSONシリアライズ可能な形式に変換"""
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError