6장: 비디오 이해와 분석

import cv2
from pathlib import Path
 
def extract_frames_uniform(
    video_path: str,
    num_frames: int = 10,
    output_dir: str = "/tmp/frames"
) -> list[str]:
    """비디오에서 균일 간격으로 프레임 추출"""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
 
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    duration = total_frames / fps
 
    # 균일 간격 계산
    interval = total_frames // num_frames
    frame_paths = []
 
    for i in range(num_frames):
        frame_idx = i * interval
        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
        ret, frame = cap.read()
        if ret:
            timestamp = frame_idx / fps
            path = f"{output_dir}/frame_{i:03d}_{timestamp:.1f}s.jpg"
            cv2.imwrite(path, frame)
            frame_paths.append(path)
 
    cap.release()
    return frame_paths

import numpy as np
 
def extract_scene_changes(
    video_path: str,
    threshold: float = 30.0,
    max_frames: int = 20
) -> list[tuple[int, float]]:
    """장면 전환 지점에서 프레임 추출"""
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    scene_frames = []
    prev_frame = None
 
    frame_idx = 0
    while True:
        ret, frame = cap.read()
        if not ret:
            break
 
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
 
        if prev_frame is not None:
            # 프레임 간 차이 계산
            diff = cv2.absdiff(prev_frame, gray)
            mean_diff = np.mean(diff)
 
            if mean_diff > threshold:
                timestamp = frame_idx / fps
                scene_frames.append((frame_idx, timestamp))
 
        prev_frame = gray
        frame_idx += 1
 
    cap.release()
 
    # 최대 프레임 수 제한
    if len(scene_frames) > max_frames:
        indices = np.linspace(0, len(scene_frames) - 1, max_frames, dtype=int)
        scene_frames = [scene_frames[i] for i in indices]
 
    return scene_frames

import anthropic
import base64
 
client = anthropic.Anthropic()
 
def analyze_video_frames(frame_paths: list[str], question: str) -> str:
    """추출된 프레임으로 비디오 분석"""
    content = []
 
    for i, path in enumerate(frame_paths):
        # 타임스탬프 추출 (파일명에서)
        timestamp = path.split("_")[-1].replace("s.jpg", "")
 
        with open(path, "rb") as f:
            img_data = base64.standard_b64encode(f.read()).decode("utf-8")
 
        content.append({
            "type": "text",
            "text": f"[프레임 {i+1} — {timestamp}초]",
        })
        content.append({
            "type": "image",
            "source": {"type": "base64", "media_type": "image/jpeg", "data": img_data},
        })
 
    content.append({
        "type": "text",
        "text": f"""위 프레임들은 하나의 비디오에서 시간순으로 추출된 것입니다.
 
질문: {question}
 
비디오의 시간적 흐름을 고려하여 답변해주세요.""",
    })
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": content}],
    )
    return response.content[0].text

def analyze_video_with_transcript(
    frame_paths: list[str],
    transcript: str,
    task: str
) -> str:
    """프레임과 자막을 결합한 분석"""
    content = []
 
    # 프레임 추가
    for path in frame_paths:
        with open(path, "rb") as f:
            img_data = base64.standard_b64encode(f.read()).decode("utf-8")
        content.append({
            "type": "image",
            "source": {"type": "base64", "media_type": "image/jpeg", "data": img_data},
        })
 
    # 자막과 분석 지시 추가
    content.append({
        "type": "text",
        "text": f"""위 이미지들은 비디오의 핵심 프레임입니다.
 
아래는 비디오의 음성 자막입니다:
---
{transcript}
---
 
작업: {task}
 
시각적 정보와 음성 내용을 모두 활용하여 분석해주세요.""",
    })
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": content}],
    )
    return response.content[0].text

async def summarize_lecture(video_path: str) -> dict:
    """강의 영상 자동 요약"""
    # 1. 음성 추출 및 STT
    transcript = await transcribe_audio(video_path)
 
    # 2. 슬라이드 전환 지점에서 프레임 추출
    frames = extract_scene_changes(video_path, threshold=25.0, max_frames=15)
    frame_paths = save_frames(video_path, frames)
 
    # 3. 프레임 + 자막으로 종합 분석
    summary = analyze_video_with_transcript(
        frame_paths,
        transcript,
        task="""이 강의를 분석하여 다음을 작성해주세요:
 
1. 강의 주제와 목표
2. 핵심 개념별 요약 (슬라이드 기반)
3. 주요 코드 예제나 수식 (있다면)
4. 핵심 요약 (500자 이내)
5. Q&A나 토론 내용 (있다면)"""
    )
 
    return {
        "transcript": transcript,
        "frame_count": len(frame_paths),
        "summary": summary,
    }

demo_analysis_prompt = """이 제품 데모 비디오를 분석해주세요.
 
1. **제품 개요**: 어떤 제품/서비스인지
2. **핵심 기능**: 시연된 주요 기능 (시간순)
3. **UI/UX 인상**: 인터페이스 디자인 평가
4. **기술 스택 추정**: 화면에서 파악 가능한 기술
5. **경쟁 우위**: 차별화 포인트
6. **개선 제안**: 발견된 이슈나 개선점"""

def analyze_security_footage(
    frame_paths: list[str],
    context: str = "사무실 입구"
) -> dict:
    """보안 영상에서 이상 상황 감지"""
    response = analyze_video_frames(
        frame_paths,
        question=f"""이 {context} 보안 카메라 영상을 분석하여:
 
1. 감지된 인원 수와 대략적인 시간대
2. 비정상적인 행동이나 이벤트
3. 출입 패턴 요약
4. 보안 관점에서의 주의사항
 
개인 식별 정보는 포함하지 마세요."""
    )
    return {"analysis": response, "frame_count": len(frame_paths)}

def optimize_frame_selection(
    video_path: str,
    budget_frames: int = 8,
) -> list[str]:
    """비용 예산에 맞는 프레임 선택"""
    # 1. 많은 프레임을 저해상도로 추출
    all_frames = extract_frames_uniform(video_path, num_frames=50)
 
    # 2. 프레임 간 유사도 계산하여 중복 제거
    unique_frames = remove_similar_frames(all_frames, similarity_threshold=0.9)
 
    # 3. 예산에 맞게 균일 샘플링
    if len(unique_frames) > budget_frames:
        indices = np.linspace(0, len(unique_frames) - 1, budget_frames, dtype=int)
        unique_frames = [unique_frames[i] for i in indices]
 
    # 4. 선택된 프레임만 고해상도로 추출
    return extract_high_res_frames(video_path, unique_frames)