Twelve Labs와 함께 비디오 하이라이트 생성기 만들기

Twelvelabs_API=your_api_key_here
API_URL=your_api_url_here

from twelvelabs import TwelveLabs
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("API_KEY")
client = TwelveLabs(api_key=API_KEY)

index = client.index.create(
    name="<YOUR_INDEX_NAME>",
    engines=engines
)
print(f"A new index has been created: Index id={index.id} name={index.name} engines={index.engines}")

pip install -r requirements.txt

import os
import requests
from moviepy.editor import VideoFileClip
from twelvelabs import TwelveLabs
from dotenv import load_dotenv
import io
import m3u8
from urllib.parse import urljoin
import yt_dlp

# Load environment variables
load_dotenv()

API_KEY = os.getenv("API_KEY")
INDEX_ID = os.getenv("INDEX_ID")

def seconds_to_mmss(seconds):
    minutes, seconds = divmod(int(seconds), 60)
    return f"{minutes:02d}:{seconds:02d}"

def mmss_to_seconds(mmss):
    minutes, seconds = map(int, mmss.split(':'))
    return minutes * 60 + seconds

def generate_timestamps(client, video_id, start_time=0):
    try:
        gist = client.generate.summarize(video_id=video_id, type="chapter")
        chapter_text = "\n".join([f"{seconds_to_mmss(chapter.start + start_time)}-{chapter.chapter_title}" for chapter in gist.chapters])
        return chapter_text, gist.chapters[-1].start + start_time
    except Exception as e:
        raise Exception(f"An error occurred while generating timestamps: {str(e)}")

# Utitily function to trim the video based on the time stamps
def trim_video(input_path, output_path, start_time, end_time):
    with VideoFileClip(input_path) as video:
        new_video = video.subclip(start_time, end_time)
        new_video.write_videofile(output_path, codec="libx264", audio_codec="aac")

# Based on the speicific Index_ID, fetching all the video_id
def fetch_existing_videos():
    url = f"https://api.twelvelabs.io/v1.2/indexes/{INDEX_ID}/videos?page=1&page_limit=10&sort_by=created_at&sort_option=desc"
    headers = {"accept": "application/json", "x-api-key": API_KEY, "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json()['data']
    else:
        raise Exception(f"Failed to fetch videos: {response.text}")

# Utility function to retrieve the URL of the video with video_id
def get_video_url(video_id):
    url = f"https://api.twelvelabs.io/v1.2/indexes/{INDEX_ID}/videos/{video_id}"
    headers = {"accept": "application/json", "x-api-key": API_KEY}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        data = response.json()
        return data['hls']['video_url'] if 'hls' in data and 'video_url' in data['hls'] else None
    else:
        raise Exception(f"Failed to get video URL: {response.text}")


# Utility function to handle and process the video clips larger than 30 mins
def process_video(client, video_path, video_type):
    with VideoFileClip(video_path) as clip:
        duration = clip.duration

    if duration > 3600:
        raise Exception("Video duration exceeds 1 hour. Please upload a shorter video.")

    if video_type == "Basic Video (less than 30 mins)":
        task = client.task.create(index_id=INDEX_ID, file=video_path)
        task.wait_for_done(sleep_interval=5)
        if task.status == "ready":
            timestamps, _ = generate_timestamps(client, task.video_id)
            return timestamps, task.video_id
        else:
            raise Exception(f"Indexing failed with status {task.status}")
    
    elif video_type == "Podcast (30 mins to 1 hour)":
        trimmed_path = os.path.join(os.path.dirname(video_path), "trimmed_1.mp4")
        trim_video(video_path, trimmed_path, 0, 1800)
        task1 = client.task.create(index_id=INDEX_ID, file=trimmed_path)
        task1.wait_for_done(sleep_interval=5)
        os.remove(trimmed_path)
        if task1.status != "ready":
            raise Exception(f"Indexing failed with status {task1.status}")
        
        timestamps, end_time = generate_timestamps(client, task1.video_id)
        
        if duration > 1800:
            trimmed_path = os.path.join(os.path.dirname(video_path), "trimmed_2.mp4")
            trim_video(video_path, trimmed_path, 1800, int(duration))
            task2 = client.task.create(index_id=INDEX_ID, file=trimmed_path)
            task2.wait_for_done(sleep_interval=5)
            os.remove(trimmed_path)
            if task2.status != "ready":
                raise Exception(f"Indexing failed with status {task2.status}")
            
            timestamps_2, _ = generate_timestamps(client, task2.video_id, start_time=end_time)
            timestamps += "\n" + timestamps_2
        
        return timestamps, task1.video_id


# Utility function to render the video on the UI
def get_hls_player_html(video_url):
    return f"""
    <script src="https://cdn.jsdelivr.net/npm/hls.js@latest"><&sol;script>
    <style>
        #video-container {{
            position: relative;
            width: 100%;
            padding-bottom: 56.25%;
            overflow: hidden;
            border-radius: 10px;
            box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
        }}
        #video {{
            position: absolute;
            top: 0;
            left: 0;
            width: 100%;
            height: 100%;
            object-fit: contain;
        }}
    <&sol;style>
    <div id="video-container">
        <video id="video" controls><&sol;video>
    <&sol;div>
    <script>
        var video = document.getElementById('video');
        var videoSrc = "{video_url}";
        if (Hls.isSupported()) {{
            var hls = new Hls();
            hls.loadSource(videoSrc);
            hls.attachMedia(video);
            hls.on(Hls.Events.MANIFEST_PARSED, function() {{
                video.pause();
            }});
        }}
        else if (video.canPlayType('application/vnd.apple.mpegurl')) {{
            video.src = videoSrc;
            video.addEventListener('loadedmetadata', function() {{
                video.pause();
            }});
        }}
    <&sol;script>
    """

# 비디오 ID로부터 URL을 조회해 인덱싱된 원본 비디오를 실제로 내려받는 유틸리티 함수
def download_video(url, output_filename):
    ydl_opts = {
        'format': 'best',
        'outtmpl': output_filename,
    }
    
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        ydl.download([url])

# 개별 세그먼트 정보를 프로그램이 구별 가능하도록 분석하는 유틸리티 함수
def parse_segments(segment_text):
    lines = segment_text.strip().split('\n')
    segments = []
    for i, line in enumerate(lines):
        start, description = line.split('-', 1)
        start_time = mmss_to_seconds(start.strip())
        if i < len(lines) - 1:
            end = lines[i+1].split('-')[0]
            end_time = mmss_to_seconds(end.strip())
        else:
            end_time = None
        segments.append((start_time, end_time, description.strip()))
    return segments


# 분석된 정보에 따라 개별 동영상 세그먼트를 슬라이싱하는 핵심 함수
def create_video_segments(video_url, segment_info):
    full_video = "full_video.mp4"
    segments = parse_segments(segment_info)

    try:
        # 원본 동영상 클립 우선 다운로드
        download_video(video_url, full_video)
        
        for i, (start_time, end_time, description) in enumerate(segments):
            output_file = f"{i+1:02d}_{description.replace(' ', '_').lower()}.mp4"
            trim_video(full_video, output_file, start_time, end_time)
            yield output_file, description
        
        os.remove(full_video)
    
    except yt_dlp.utils.DownloadError as e:
        raise Exception(f"다운로드 완료 중 오류 발생: {str(e)}")
    except Exception as e:
        raise Exception(f"예기치 못한 장애 발생: {str(e)}")

# 잘라내기(Trimming) 처리를 마친 각 클립 세그먼트를 내려받는 함수
def download_video_segment(video_id, start_time, end_time=None):
    video_url = get_video_url(video_id)
    if not video_url:
        raise Exception("비디오 URL 조회 실패")

    playlist = m3u8.load(video_url)
    
    start_seconds = mmss_to_seconds(start_time)
    end_seconds = mmss_to_seconds(end_time) if end_time else None

    total_duration = 0
    segments_to_download = []
    for segment in playlist.segments:
        if total_duration >= start_seconds and (end_seconds is None or total_duration < end_seconds):
            segments_to_download.append(segment)
        total_duration += segment.duration
        if end_seconds is not None and total_duration >= end_seconds:
            break

    buffer = io.BytesIO()
    for segment in segments_to_download:
        segment_url = urljoin(video_url, segment.uri)
        response = requests.get(segment_url)
        if response.status_code == 200:
            buffer.write(response.content)
        else:
            raise Exception(f"세그먼트 다운로드 실패: {segment_url}")

    buffer.seek(0)
    return buffer.getvalue()

# 파일 업로드 컨트롤러 및 프론트 정비 함수
def upload_and_process_video():
    video_type = st.selectbox("비디오 유형 선택:", ["Basic Video (less than 30 mins)", "Podcast (30 mins to 1 hour)"])
    uploaded_file = st.file_uploader("적용할 비디오 파일 지정", type=["mp4", "mov", "avi"])

    if uploaded_file and st.button("파일 처리 분석 시행", key="process_video_button"):
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp_file:
            tmp_file.write(uploaded_file.read())
            video_path = tmp_file.name
        try:
            with st.spinner("영상을 활발히 분석 중입니다..."):
                client = TwelveLabs(api_key=API_KEY)
                timestamps, video_id = process_video(client, video_path, video_type)
            st.success("비디오 분석 작업이 모두 성공적으로 끝났습니다!")
            st.session_state.timestamps = timestamps
            st.session_state.video_id = video_id
            st.session_state.video_url = get_video_url(video_id)
            if st.session_state.video_url:
                st.video(st.session_state.video_url)
            else:
                st.error("비디오를 재생하기 위한 유효 URL을 얻지 못했습니다.")
        except Exception as e:
            st.error(str(e))
        finally:
            os.unlink(video_path)

# 기존에 잘 저장되어 있는 인덱스 비디오를 골라 타임스탬프 생성을 호출하는 함수
def select_existing_video():
    try:
        existing_videos = fetch_existing_videos()
        video_options = {f"{video['metadata']['filename']} ({video['_id']})": video['_id'] for video in existing_videos}
        
        if video_options:
            selected_video = st.selectbox("작업할 비디오 선택:", list(video_options.keys()))
            video_id = video_options[selected_video]
            
            st.session_state.video_id = video_id
            st.session_state.video_url = get_video_url(video_id)
            
            if st.session_state.video_url:
                st.markdown(f"### 지정된 비디오: {selected_video}")
                st.video(st.session_state.video_url)
            else:
                st.error("비디오 주소를 정상적으로 받지 못했습니다.")
            
            if st.button("타임스탬프 자동 생성", key="generate_timestamps_button"):
                with st.spinner("타임스탬프를 계산 중입니다..."):
                    client = TwelveLabs(api_key=API_KEY)
                    timestamps, _ = generate_timestamps(client, video_id)
                st.session_state.timestamps = timestamps
        else:
            st.warning("보유하신 인덱스 내에 등록된 영상 자산이 하나도 존재하지 않습니다.")
    except Exception as e:
        st.error(str(e))

# 생성 완료된 자사 세그먼트 클립 목록을 이쁘게 표시하고 보존 다운로드할 수 있는 매개 함수
def display_segment(file_name, description, segment_index):
    if os.path.exists(file_name):
        st.write(f"### {description}")
        st.video(file_name)
        with open(file_name, "rb") as file:
            file_contents = file.read()
        unique_key = f"download_{segment_index}_{uuid.uuid4()}"
        st.download_button(
            label=f"클립 다운로드: {description}",
            data=file_contents,
            file_name=file_name,
            mime="video/mp4",
            key=unique_key
        )
        st.markdown("---")
    else:
        st.warning(f"파일 자산 {file_name}을 로드하지 못했거나 다른 하드 드라이브로 복사 이동되었습니다.")


# 실제 세그먼트를 추출하고 표시하는 핵심 함수
def process_and_display_segments():
    if not st.session_state.video_url:
        st.error("연결된 본 소스의 미디어 URL을 찾을 수 없습니다. 다시 전송 분석을 진행해 주세요.")
        return

    segment_generator = create_video_segments(st.session_state.video_url, st.session_state.timestamps)

    progress_bar = st.progress(0)
    status_text = st.empty()

    st.session_state.video_segments = []  # 비디오 세그먼트 목록 초기화
    total_segments = len(st.session_state.timestamps.split('\n'))

    for i, (file_name, description) in enumerate(segment_generator, 1):
        st.session_state.video_segments.append((file_name, description))
        display_segment(file_name, description, i-1)  # 인덱스 순번을 동봉 전달
        progress = i / total_segments
        progress_bar.progress(progress)
        status_text.text(f"클립 파트 처리 세분화 중 {i}/{total_segments}...")

    progress_bar.progress(1.0)
    status_text.text("모든 클립의 정밀 세분화가 성공적으로 끝났습니다!")


# 생성된 타임라인 인덱스와 분기 세그먼트 카드들을 전반적으로 레이아웃하는 렌더 관리 함수
def display_timestamps_and_segments():
    if st.session_state.timestamps:
        st.subheader("YouTube 설명 등록용 추천 챕터 메타 데이터")
        st.write("아래 텍스트 영역을 통째로 복사하셔서 비디오 업로드 상세 묘사란에 그대로 기재하세요.")
        st.code(st.session_state.timestamps, language="")

        if st.button("개별 비디오 챕터 클립 실물 패키징 생성", key="create_segments_button"):
            try:
                process_and_display_segments()
            except Exception as e:
                st.error(f"개별 미디어 클립 패키징 도중 에러가 발발했습니다: {str(e)}")
                st.exception(e)  # 역추적을 위한 스택 트레이스 표시

        if st.session_state.video_segments:
            st.subheader("추출 성공한 개별 클립 패키지")
            for index, (file_name, description) in enumerate(st.session_state.video_segments):
                display_segment(file_name, description, index)

            if st.button("임시 저장 공간 비우기", key="clear_segments_button"):
                for file_name, _ in st.session_state.video_segments:
                    if os.path.exists(file_name):
                        os.remove(file_name)
                st.session_state.video_segments = []
                st.success("생성되었던 모든 임시 세그먼트 파일들을 스토리지에서 깨끗하게 소거했습니다.")
                st.experimental_rerun()