使用Amazon Nova模型实现自动化视频高光剪辑

# Nova模型分析视频
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1', 
                       config=Config(read_timeout=1800))

prompt = """You are an expert in football video analysis.
Identify ONLY goal moments from this match video.

**Output Format (JSON only):**
[
    {
        "start_time": "MM:SS",
        "end_time": "MM:SS",
        "description": "Goal description",
        "scene_type": "Goal"
    }
]

**Critical Constraints:**
- All timestamps MUST be within video duration
- Output ONLY valid JSON array
- NO overlapping timestamps"""

messages = [{
    "role": "user",
    "content": [
        {"video": {"format": "mp4", "source": {"s3Location": {"uri": s3_uri, "bucketOwner": account_id}}}},
        {"text": prompt}
    ]
}]

response = bedrock.converse(
    modelId= <nova-lite-model-id>, #具体名称可以参考：https://aws.amazon.com/nova
    messages=messages,
    inferenceConfig={"maxTokens": 65535, "temperature": 0.0, "topP": 1.0},
    additionalModelRequestFields={
        "reasoningConfig": {"type": "enabled", "maxReasoningEffort": "low"}
    }
)

output = response['output']['message']['content'][0]['text']

### Task:
You are an expert in video content analysis and temporal localization.
 
### Analysis Process (Follow these steps):
Step 1: Watch the entire video and identify all highlight moments
Step 2: For each moment, determine precise start and end timestamps
Step 3: Verify all timestamps are within the video duration
Step 4: Output structured JSON format only

**Analyze from these perspectives:** 
- Visual dynamics: motion intensity, camera movement, visual effects
- Emotional impact: excitement level, dramatic tension
- Technical complexity: skill difficulty, coordination required
- Narrative significance: story turning points, key moments
- Audience appeal: shareability, memorable elements

**Output Format:**
Generate detailed, time-stamped descriptions of events.
Each event follows the format: "#START - END seconds# description"

**Example:**
#0.8 - 11.3 seconds# Athlete performs a high jump over obstacle
#32.5 - 50.0 seconds# Crowd celebrates as player scores

**Output Format (JSON):**
[
    {
        "start_time": "MM:SS",
        "end_time": "MM:SS",
        "description": "Detailed event description",
        "scene_type": "Action|Transition|Climax"
    }
]

**CRITICAL CONSTRAINTS:**
- Video duration: exactly 3 minutes 26 seconds (00:00 to 03:26)
- All timestamps MUST be within this range
- No events beyond 03:26
- Use MM:SS format consistently

import boto3
import json
import subprocess
import base64
import glob
from sklearn.metrics.pairwise import cosine_similarity

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

# 1. LLM视频分析 - Nova模型理解视频生成高光要点
def analyze_video(video_path):
    with open(video_path, 'rb') as f:
        video_base64 = base64.b64encode(f.read()).decode('utf-8')
    
    response = bedrock.invoke_model(
        modelId="us.amazon.nova-lite-v1:0",
        body=json.dumps({
            "messages": [{"role": "user", "content": [
                {"video": {"format": "mp4", "source": {"bytes": video_base64}}},
                {"text": """请分析这个视频并提炼高光要点。                       
                            ## 高光片段判定标准：
                            - 动作精彩或技巧性强的时刻
                            - 情感表达丰富或戏剧性的瞬间  
                            - 关键转折点或重要事件
                            - 视觉效果突出或构图优美的片段
                            - 具有故事性或叙事价值的时刻
                            
                            ## 输出要求：
                            请按以下格式输出高光要点，每个要点包含优先级（1=最重要，2=重要，3=一般）：
                            
                            **视频总结：**
                            [简要描述视频的整体内容和主题]
                            
                            **高光要点列表：**
                            A. [优先级1] - [具体的高光内容描述]
                            B. [优先级2]  - [具体的高光内容描述]  
                            C. [优先级1]  - [具体的高光内容描述]
                            ...
                            
                            请确保：
                            1. 按视频时间顺序排列要点
                            2. 每个要点都有明确的优先级标记
                            3. 描述具体且便于后续匹配
                            4. 重点关注真正精彩的时刻，而非简单概括"""}
            ]}]
        })
    )
    return json.loads(response["body"].read())["output"]["message"]["content"][0]["text"]

# 2. 视频切片 - FFmpeg按固定间隔切分视频
def extract_clips(video_path, clip_duration=3):
    # 获取视频时长
    duration = float(subprocess.check_output(['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration', '-of', 'csv=p=0', video_path]))
    
    clips = []
    for i in range(0, int(duration), clip_duration):
        clip_path = f"clips/clip_{i:04d}_{i}s.mp4"
        subprocess.run(['ffmpeg', '-i', video_path, '-ss', str(i), '-t', str(clip_duration), 
                       '-c:v', 'libx264', clip_path, '-y', '-loglevel', 'quiet'])
        clips.append({'timestamp': i, 'path': clip_path})
    return clips

# 3. 语义匹配 - Nova MME生成嵌入后计算要点与视频片段相似度
def get_embedding(content, content_type="text"):
    if content_type == "text":
        request = {
            "taskType": "SINGLE_EMBEDDING",
            "singleEmbeddingParams": {
                "embeddingPurpose": "GENERIC_INDEX",
                "embeddingDimension": 1024,
                "text": {"truncationMode": "END", "value": content}
            }
        }
    else:  # video
        with open(content, 'rb') as f:
            video_base64 = base64.b64encode(f.read()).decode('utf-8')
        request = {
            "taskType": "SINGLE_EMBEDDING",
            "singleEmbeddingParams": {
                "embeddingPurpose": "GENERIC_INDEX",
                "embeddingDimension": 1024,
                "video": {
                    "format": "mp4",
                    "embeddingMode": "AUDIO_VIDEO_COMBINED",
                    "source": {"bytes": video_base64}
                }
            }
        }
    
    response = bedrock.invoke_model(modelId="amazon.nova-2-multimodal-embeddings-v1:0", body=json.dumps(request))
    return json.loads(response["body"].read())["embeddings"][0]["embedding"]

def semantic_match(analysis, clips):
    # 提取要点
    points = [line.strip() for line in analysis.split('\n') if line.strip().startswith(('A.', 'B.', 'C.'))]
    
    selected_clips = []
    # 为每个要点找最佳匹配片段
    for point in points:
        text_emb = get_embedding(point)
        best_clip, best_similarity = None, -1
        
        for clip in clips:
            video_emb = get_embedding(clip['path'], "video")
            similarity = cosine_similarity([text_emb], [video_emb])[0][0]
            
            if similarity > best_similarity:
                best_similarity = similarity
                best_clip = {'path': clip['path'], 'timestamp': clip['timestamp'], 'similarity': similarity}
        
        if best_clip and best_similarity > 0.15:
            selected_clips.append(best_clip)
    
    # 按时间顺序排序
    selected_clips.sort(key=lambda x: x['timestamp'])
    return selected_clips

# 4. 拼接高光视频 - FFmpeg合并选中片段
def create_video(selected_clips, output_path):
    # 生成拼接列表
    with open('concat_list.txt', 'w') as f:
        for clip in selected_clips:
            f.write(f"file '{clip['path']}'\n")
    
    # FFmpeg拼接
    subprocess.run(['ffmpeg', '-f', 'concat', '-safe', '0', '-i', 'concat_list.txt', 
                   '-c', 'copy', output_path, '-y', '-loglevel', 'quiet'])

# demo：完整流程
analysis = analyze_video("sample/action.mp4")
clips = extract_clips("sample/action.mp4", 3)
selected_clips = semantic_match(analysis, clips)
create_video(selected_clips, "highlight_video.mp4")
# 流程: 视频→Nova理解→视频切片→MME匹配→拼接
# 返回: highlight_video.mp4

**视频总结：**
这个视频展示了一只橙色小狗在房子门口的阶梯上睡觉。视频中的场景是一个带有花园和邮箱的房子，背景中还有一辆自行车停在路边。

**高光要点列表：**
A. [优先级1] - 0:00 - 视频开始时，展示了房子和橙色小狗在阶梯上睡觉的场景。
B. [优先级2] - 0:19 - 橙色小狗开始翻身并醒来。
C. [优先级1] - 0:21 - 黄色小狗被一个蓝色的球吸引，并开始追逐。
D. [优先级2] - 0:25 - 黄色小狗追逐球的动作，展示了它的活泼和好奇心。
E. [优先级3] - 0:30 - 黄色小狗追逐球的过程中，展示了房子的细节和背景。
F. [优先级2] - 0:35 - 黄色小狗最终放弃追逐。
G. [优先级3] - 0:40 - 视频结尾，展示了房子和花园的全景。

import boto3
import json
import subprocess
import base64
from sklearn.metrics.pairwise import cosine_similarity

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

# 1. VLM视频分析
def analyze_video(video_path):
    # 同 “2.1 基本方案”部分案例代码
    pass
    
# 2. FFmpeg抽帧：将视频抽帧为图片用于后续语义匹配
def extract_frames(video_path, interval=1):
    duration = float(subprocess.check_output(['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration', '-of', 'csv=p=0', video_path]))
    
    frames = []
    for i, t in enumerate(range(0, int(duration), interval)):
        frame_path = f"frame_{i:04d}.jpg"
        subprocess.run(['ffmpeg', '-i', video_path, '-ss', str(t), '-vframes', '1', frame_path, '-y', '-loglevel', 'quiet'])
        frames.append({'timestamp': t, 'path': frame_path})
    return frames

# 3. 语义匹配：Nova MME生成嵌入后计算要点与图片相似度
def get_embedding(content, content_type="text"):
    # 生成嵌入
    if content_type == "text":
        data = {"text": {"value": content}}
    else:
        with open(content, 'rb') as f:
            data = {"image": {"format": "jpeg", "source": {"bytes": base64.b64encode(f.read()).decode()}}}
    
    response = bedrock.invoke_model(
        modelId="amazon.nova-2-multimodal-embeddings-v1:0",
        body=json.dumps({"taskType": "SINGLE_EMBEDDING", "singleEmbeddingParams": data})
    )
    return json.loads(response["body"].read())["embeddings"][0]["embedding"]

def semantic_match(analysis, frames):
    # 提取要点
    points = [line.strip() for line in analysis.split('\n') if line.strip().startswith(('A.', 'B.', 'C.'))]
    
    selected_clips = []
    # 针对每一个要点，匹配处Top10最相关的帧集合
    for point in points:
        text_emb = get_embedding(point)
        best_frames = []
        
        for frame in frames:
            img_emb = get_embedding(frame['path'], "image")
            similarity = cosine_similarity([text_emb], [img_emb])[0][0]
            if similarity > 0.1:
                best_frames.append({'timestamp': frame['timestamp'], 'similarity': similarity})
        
        best_frames.sort(key=lambda x: x['similarity'], reverse=True)
        selected_clips.extend(best_frames[:10])  # Top10
    
    return selected_clips

# 4. 生成高光视频
def create_video(video_path, clips, output_path):
    clips.sort(key=lambda x: x['timestamp'])
    
    # 按照帧集合提取片段
    temp_clips = []
    for i, clip in enumerate(clips):
        clip_path = f"clip_{i}.mp4"
        subprocess.run(['ffmpeg', '-i', video_path, '-ss', str(clip['timestamp']), '-t', '2', clip_path, '-y', '-loglevel', 'quiet'])
        temp_clips.append(clip_path)
    
    # 拼接
    with open('list.txt', 'w') as f:
        for clip in temp_clips:
            f.write(f"file '{clip}'\n")
    
    subprocess.run(['ffmpeg', '-f', 'concat', '-safe', '0', '-i', 'list.txt', '-c', 'copy', output_path, '-y', '-loglevel', 'quiet'])

# demo：完整流程
analysis = analyze_video("input.mp4")
frames = extract_frames("input.mp4")
clips = semantic_match(analysis, frames)
create_video("input.mp4", clips, "highlight.mp4")
# 流程: 视频→Nova理解→视频抽帧为图片→MME匹配→拼接
# 返回: highlight.mp4

**视频总结：**
这段视频展示了一只橙色和白色相间的小猫在草地上的活动。从开始到结束，小猫一直在草地上探索、玩耍，并最终发现并尝试吃一个蛋壳。

原始高光要点列表：
A. [优先级1] [0:00-0:05] - 小猫在草地上坐下，观察周围环境，展示了它的好奇心和警觉性
B. [优先级2] [0:23-0:25] - 小猫抬头看向远处，表现出对环境的探索和兴趣
C. [优先级1] [0:27-0:30] - 小猫发现地上的蛋壳，并开始尝试吃掉，展示了它的食欲和探索行为
D. [优先级2] [0:33-0:35] - 小猫尝试吃蛋壳时，表现出一些困惑和不确定的表情，增加了视频的趣味性
E. [优先级1] [0:37-0:40] - 小猫最终成功吃掉了蛋壳，并继续在草地上活动，展示了它的满足感