Twelve Labs API로 손쉬운 비디오 분류하기: ML 학습이 필요 없습니다!

<pre><code class="json">"classes": [
      {
            "name": "Doge_Bruno",
            "prompts": [
                "playing with my dog",
                "my dog doing funny things",
                "dog playing with water"
            ]
        },
        {
            "name": "Kitty_Karla",
            "prompts": [
                "cat jumping",
                "cat playing with toys"
            ]
        }
  ]
  </code></pre>

<pre><code class="python">%env API_KEY=<your_API_key>
%env API_URL=https://api.twelvelabs.io/v1.1

!pip install requests
!pip install flask

import os
import requests
import glob
from pprint import pprint


#Retrieve the URL of the API and the API key
API_URL = os.getenv("API_URL")
assert API_URL

API_KEY = os.getenv("API_KEY")
assert API_KEY
</code></pre>

<pre><code class="python"># Construct the URL of the `/indexes` endpoint
INDEXES_URL = f"{API_URL}/indexes"

# Set the header of the request
default_header = {
    "x-api-key": API_KEY
}

# Define a function to create an index with a given name
def create_index(index_name, index_options, engine):
    # Declare a dictionary named data
    data = {
        "engine_id": engine,
        "index_options": index_options,
        "index_name": index_name,
    }

    # Create an index
    response = requests.post(INDEXES_URL, headers=default_header, json=data)

    # Store the unique identifier of your index
    INDEX_ID = response.json().get('_id')

    # Check if the status code is 201 and print success
    if response.status_code == 201:
        print(f"Status code: {response.status_code} - The request was successful and a new index was created.")
    else:
        print(f"Status code: {response.status_code}")
    pprint(response.json())
    return INDEX_ID


# Create the indexes
index_id_content_classification = create_index(index_name = "insta+tiktok", index_options=["visual", "conversation", "text_in_video", "logo"], engine = "marengo2.5")

# Print the created index IDs
print(f"Created index IDs: {index_id_content_classification}")
</code></pre>

<pre><code class="bash">Status code: 201 - The request was successful and a new index was created.
{'_id': '64544b858b1dd6cde172af77'}
Created index IDs: 64544b858b1dd6cde172af77
</code></pre>

<pre><code class="python">import os
import requests
from concurrent.futures import ThreadPoolExecutor

TASKS_URL = f"{API_URL}/tasks"
TASK_ID_LIST = []
video_folder = 'classify'  # folder containing the video files
INDEX_ID = '64544b858b1dd6cde172af77'

def upload_video(file_name):
    # Validate if a video already exists in the index
    task_list_response = requests.get(
        TASKS_URL,
        headers=default_header,
        params={"index_id": INDEX_ID, "filename": file_name},
    )
    if "data" in task_list_response.json():
        task_list = task_list_response.json()["data"]
        if len(task_list) > 0:
            if task_list[0]['status'] == 'ready': 
                print(f"Video '{file_name}' already exists in index {INDEX_ID}")
            else:
                print("task pending or validating")
            return

    # Proceed further to create a new task to index the current video if the video didn't exist in the index already
    print("Entering task creation code for the file: ", file_name)
    
    if file_name.endswith('.mp4'):  # Make sure the file is an MP4 video
        file_path = os.path.join(video_folder, file_name)  # Get the full path of the video file
        with open(file_path, "rb") as file_stream:
            data = {
                "index_id": INDEX_ID,
                "language": "en"
            }
            file_param = [
                ("video_file", (file_name, file_stream, "application/octet-stream")),] #The video will be indexed on the platform using the same name as the video file itself.
            response = requests.post(TASKS_URL, headers=default_header, data=data, files=file_param)
            TASK_ID = response.json().get("_id")
            TASK_ID_LIST.append(TASK_ID)
            # Check if the status code is 201 and print success
            if response.status_code == 201:
                print(f"Status code: {response.status_code} - The request was successful and a new resource was created.")
            else:
                print(f"Status code: {response.status_code}")
            print(f"File name: {file_name}")
            pprint(response.json())
            print("\n")

# Get list of video files
video_files = [f for f in os.listdir(video_folder) if f.endswith('.mp4')]

# Create a ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
    # Use executor to run upload_video in parallel for all video files
    executor.map(upload_video, video_files)
</code></pre>

<pre><code class="python">import time
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed

def monitor_upload_status(task_id):
    TASK_STATUS_URL = f"{API_URL}/tasks/{task_id}"
    while True:
        response = requests.get(TASK_STATUS_URL, headers=default_header)
        STATUS = response.json().get("status")
        if STATUS == "ready":
            return task_id, STATUS
        remain_seconds = response.json().get('process', {}).get('remain_seconds', None)
        upload_percentage = response.json().get('process', {}).get('upload_percentage', None)
        if remain_seconds is not None:
            print(f"Task ID: {task_id}, Remaining seconds: {remain_seconds}, Upload Percentage: {upload_percentage}")
        else:
             print(f"Task ID: {task_id}, Status: {STATUS}")
        time.sleep(10)

# Define starting time
start = time.time()
print("Starting to monitor...")

# Monitor the indexing process for all tasks
with ThreadPoolExecutor() as executor:
    futures = {executor.submit(monitor_upload_status, task_id) for task_id in TASK_ID_LIST}
    with open('upload_status.csv', 'w', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(["Task ID", "Status"])
        for future in as_completed(futures):
            task_id, status = future.result()
            writer.writerow([task_id, status])

# Define ending time
end = time.time()
print("Monitoring finished")
print("Time elapsed (in seconds): ", end - start)
</code></pre>

<pre><code class="language-plaintext">Starting to monitor...
Monitoring finished
Time elapsed (in seconds):  253.00311
</code></pre>

<pre><code class="python"># List all the videos in an index
INDEX_ID='64544b858b1dd6cde172af77'
default_header = {
    "x-api-key": API_KEY
}
# INDEX_ID='64502d238b1dd6cde172a9c5' #movies
# INDEX_ID= '64399bc25b65d57eaecafb35' #lex
INDEXES_VIDEOS_URL = f"{API_URL}/indexes/{INDEX_ID}/videos?page_limit=20"
response = requests.get(INDEXES_VIDEOS_URL, headers=default_header)

response_json = response.json()
pprint(response_json)

video_id_name_list = [{'video_id': video['_id'], 'video_name': video['metadata']['filename']} for video in response_json['data']]

pprint(video_id_name_list)
</code></pre>

<pre><code class="python">{'data': [{'_id': '64544bb486daab572f3494a0',
           'created_at': '2023-05-05T00:19:33Z',
           'metadata': {'duration': 507.5,
                        'engine_id': 'marengo2.5',
                        'filename': 'JetTila.mp4',
                        'fps': 30,
                        'height': 720,
                        'size': 44891944,
                        'width': 1280},
           'updated_at': '2023-05-05T00:20:04Z'},
          {'_id': '64544bad86daab572f34949f',
           'created_at': '2023-05-05T00:19:32Z',
           'metadata': {'duration': 516.682833,
                        'engine_id': 'marengo2.5',
                        'filename': 'Kylie.mp4',
                        'fps': 23.976023976023978,
                        'height': 720,
                        'size': 37594080,
                        'width': 1280},
           'updated_at': '2023-05-05T00:19:57Z'},
          {'_id': '64544b9286daab572f34949e',
           'created_at': '2023-05-05T00:19:27Z',
           'metadata': {'duration': 34.333333,
                        'engine_id': 'marengo2.5',
                        'filename': 'Oh_my.mp4',
                        'fps': 30,
                        'height': 1280,
                        'size': 10480126,
                        'width': 720},
           'updated_at': '2023-05-05T00:19:30Z'},
          .
					.
					.
					{'_id': '64544b8786daab572f349496',
           'created_at': '2023-05-05T00:19:18Z',
           'metadata': {'duration': 14.333333,
                        'engine_id': 'marengo2.5',
                        'filename': 'cats.mp4',
                        'fps': 30,
                        'height': 1280,
                        'size': 1304438,
                        'width': 720},
           'updated_at': '2023-05-05T00:19:19Z'}],
 'page_info': {'limit_per_page': 20,
               'page': 1,
               'total_duration': 1363.925599,
               'total_page': 1,
               'total_results': 11}}
[{'video_id': '64544bb486daab572f3494a0', 'video_name': 'JetTila.mp4'},
 {'video_id': '64544bad86daab572f34949f', 'video_name': 'Kylie.mp4'},
 {'video_id': '64544b9286daab572f34949e', 'video_name': 'Oh_my.mp4'},
 {'video_id': '64544b8e86daab572f34949c', 'video_name': 'Pitbull.mp4'},
 {'video_id': '64544b9286daab572f34949d', 'video_name': 'She.mp4'},
 {'video_id': '64544b8d86daab572f34949b', 'video_name': 'fun.mp4'},
 {'video_id': '64544b8986daab572f349497', 'video_name': 'Dance.mp4'},
 {'video_id': '64544b8986daab572f349498', 'video_name': 'Jennie.mp4'},
 {'video_id': '64544b8a86daab572f349499', 'video_name': 'McDonald.mp4'},
 {'video_id': '64544b8c86daab572f34949a', 'video_name': 'Orangutan.mp4'},
 {'video_id': '64544b8786daab572f349496', 'video_name': 'cats.mp4'}]
 </code></pre>

<pre><code class="python"># Construct the URL of the `/classify/bulk` endpoint
CLASSIFY_BULK_URL = f"{API_URL}/classify/bulk"

# Set the header of the request
headers = {
    "x-api-key": API_KEY
}

# Declare a dictionary named `data`
data =  {
  "conversation_option": "semantic",
  "options": ["visual", "conversation"],
  "index_id" : INDEX_ID,
  "include_clips": False,
  "classes": [
      {
            "name": "BeautyTok",
            "prompts": [
                "Makeup",
                "Skincare",
                "cosmetic products",
                "doing nails",
                "doing hair",
                "DanceTok",
                "dance tutorial",
                "dance competition",
                "dance challenge",
                "dance trend",
                "dancing with friends"
            ]
        },
        {
            "name": "CookTok",
            "prompts": [
                "cooking tutorial",
                "cooking utensils",
                "baking tutorials",
                "recipes",
                "restaurants",
                "food",
                "pasta"
            ]
        },
        {
            "name": "AnimalTok",
            "prompts": [
                "dog",
                "cat",
                "birds",
                "fish",
                "playing with pets",
                "pets doing funny things"
            ]
        },
        {
            "name": "ArtTok",
            "prompts": [
                "handicraft",
                "drawing",
                "graffiti",
                "sketching",
                "digital art",
                "coloring",
                "sketchbook",
                "artwork",
                "artists"
            ]
        }
  ]
}

# Make a classification request
response = requests.post(CLASSIFY_BULK_URL, headers=headers, json=data)
print (f'Status code: {response.status_code}')
pprint(response.json())
duration_data = response.json()
</code></pre>

<pre><code class="python">Status code: 200
{'data': [{'classes': [{'duration_ratio': 1,
                        'name': 'AnimalTok',
                        'score': 95.35111111111111}],
           'video_id': '64544b8786daab572f349496'},
          {'classes': [{'duration_ratio': 1,
                        'name': 'AnimalTok',
                        'score': 95.14666666666668}],
           'video_id': '64544b8e86daab572f34949c'},
          .
			.
			.
          {'classes': [{'duration_ratio': 0.8175611166393244,
                        'name': 'ArtTok',
                        'score': 89.45777777777778}],
           'video_id': '64544b9286daab572f34949d'}],
 'page_info': {'limit_per_page': 10,
               'next_page_token': '',
               'page_expired_at': '2023-05-05T17:37:30Z',
               'prev_page_token': '',
               'total_results': 10}}
 </code></pre>

<pre><code class="python">
# Construct the URL of the `/classify/bulk` endpoint
CLASSIFY_BULK_URL = f"{API_URL}/classify/bulk"

# Set the header of the request
headers = {
    "x-api-key": API_KEY
}

# Declare a dictionary named `data`
data =  {
  "conversation_option": "semantic",
  "options": ["visual", "conversation"],
  "index_id" : INDEX_ID,
  "include_clips": True,
  "classes": [
      {
            "name": "BeautyTok",
            "prompts": [
                "Makeup",
                "Skincare",
                "cosmetic products",
                "doing nails",
                "doing hair",
                "DanceTok",
                "dance tutorial",
                "dance competition",
                "dance challenge",
                "dance trend",
                "dancing with friends"
            ]
        },
        {
            "name": "CookTok",
            "prompts": [
                "cooking tutorial",
                "cooking utensils",
                "baking tutorials",
                "recipes",
                "restaurants",
                "food",
                "pasta"
            ]
        },
        {
            "name": "AnimalTok",
            "prompts": [
                "dog",
                "cat",
                "birds",
                "fish",
                "playing with pets",
                "pets doing funny things"
            ]
        },
        {
            "name": "ArtTok",
            "prompts": [
                "handicraft",
                "drawing",
                "graffiti",
                "sketching",
                "digital art",
                "coloring",
                "sketchbook",
                "artwork",
                "artists"
            ]
        }
  ]
}

# Make a classification request
response = requests.post(CLASSIFY_BULK_URL, headers=headers, json=data)
print (f'Status code: {response.status_code}')
pprint(response.json())
duration_data = response.json()
</code></pre>

<pre><code class="python">{'data': [{'classes': [{'clips': [{'end': 14,
                                   'option': '',
                                   'prompt': 'cat',
                                   'score': 84.77,
                                   'start': 0},
                                  {'end': 14,
                                   'option': '',
                                   'prompt': 'pets doing funny things',
                                   'score': 83.56,
                                   'start': 8.34375},
                                  {'end': 14,
                                   'option': '',
                                   'prompt': 'playing with pets',
                                   'score': 68.11,
                                   'start': 8.34375},
                                  {'end': 8.34375,
                                   'option': '',
                                   'prompt': 'pets doing funny things',
                                   'score': 58.26,
                                   'start': 0}],
                        'duration_ratio': 1,
                        'name': 'AnimalTok',
                        'score': 95.35111111111111}],
			.
			.
			.	
           'video_id': '64544b8786daab572f349496'},
          {'classes': [{'clips': [{'end': 49.189,
                                   'option': '',
                                   'prompt': 'artists',
                                   'score': 78.14,
                                   'start': 0.55},
                                  {'end': 23.659,
                                   'option': '',
                                   'prompt': 'sketching',
                                   'score': 58.85,
                                   'start': 0.55}],
                        'duration_ratio': 0.8175611166393244,
                        'name': 'ArtTok',
                        'score': 89.45777777777778}],
			.
			.
			.
			'video_id': '64544b9286daab572f34949d'}],
 'page_info': {'limit_per_page': 10,
               'next_page_token': '',
               'page_expired_at': '2023-05-05T17:37:30Z',
               'prev_page_token': '',
               'total_results': 10}}
               </code></pre>

<pre><code class="python">import pickle

with open('video_id_name_list.pickle', 'wb') as handle:
    pickle.dump(video_id_name_list, handle, protocol=pickle.HIGHEST_PROTOCOL)

with open('duration_data.pickle', 'wb') as handle:
    pickle.dump(duration_data, handle, protocol=pickle.HIGHEST_PROTOCOL)

with open('clips_data.pickle', 'wb') as handle:
    pickle.dump(clips_data, handle, protocol=pickle.HIGHEST_PROTOCOL)
    </code></pre>

<pre><code class="markdown">my_flask_app/
│   app.py
│   sample_notebook.ipynb
└───templates/
│	│   index.html
└───classify/
	│   <your_video_1.mp4>
	│   <your_video_2.mp4>
	│   <your_video_3.mp4>
			.
			.
			.
</code></pre>

<pre><code class="python">from flask import Flask, render_template, send_from_directory
import pickle
import os
from collections import defaultdict

app = Flask(__name__)

# Replace the following dictionaries with your data
with open('video_id_name_list.pickle', 'rb') as handle:
    video_id_name_list = pickle.load(handle)

with open('duration_data.pickle', 'rb') as handle:
    duration_data = pickle.load(handle)

with open('clips_data.pickle', 'rb') as handle:
    clips_data = pickle.load(handle)

VIDEO_DIRECTORY = os.path.join(os.path.dirname(os.path.realpath(__file__)), "classify")

@app.route('/<path:filename>')
def serve_video(filename):
    print(VIDEO_DIRECTORY, filename)
    return send_from_directory(directory=VIDEO_DIRECTORY, path=filename)

def get_top_clips(clips_data, num_clips=3):
    top_clips = defaultdict(list)
    for video in clips_data['data']:
        video_id = video['video_id']
        unique_prompts = set()
        for clip_class in video['classes']:
            for clip in clip_class['clips']:
                if clip['prompt'] not in unique_prompts and len(unique_prompts) < num_clips:
                    top_clips[video_id].append(clip)
                    unique_prompts.add(clip['prompt'])
    return top_clips

@app.route('/')
def home():
    video_id_name_dict = {video['video_id']: video['video_name'] for video in video_id_name_list}
    top_clips = get_top_clips(clips_data)
    return render_template('index.html', video_mapping=video_id_name_dict, duration_data=duration_data, top_clips=top_clips)


if __name__ == '__main__':
    app.run(debug=True)
</code></pre>

<pre><code class="language-html"><!doctype html>
<html lang="en">
<head>
    <link rel="shortcut icon" href="#" />
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Classification Output</title>
    <style>
        body {
            background-color: #f0e7cc;
            font-family: Arial, sans-serif;
        }
        .container {
            width: 80%;
            margin: 0 auto;
            padding-bottom: 30px;
        }
        video {
            display: block;
            margin: 0 auto;
        }
        .progress-bar {
            width: 20%;
            height: 20px;
            background-color: #f0f0f0;
            margin-bottom: 10px;
            display: inline-block;
            vertical-align: middle;
            margin-left: 10px;
        }
        .progress {
            height: 100%;
            background-color: #4caf50;
        }
        .score {
            display: inline-block;
            vertical-align: middle;
            font-weight: bold;
        }
        .video-category {
            padding-bottom: 30px;
            margin-bottom: 20px;
        }
        button {
            background-color: #008CBA;
            border: none;
            color: white;
            padding: 8px 16px;
            text-align: center;
            text-decoration: none;
            display: inline-block;
            font-size: 14px;
            margin: 4px 2px;
            cursor: pointer;
            border-radius: 4px;
            box-shadow: 0 1px 2px rgba(0,0,0,0.2);
            transition: background-color 0.2s ease;
        }
        button:hover {
            background-color: #006494;
        }
        h1 {
            color: #070707;
            text-align: center;
            background-color: #cb6a16; /* Change the background color to your desired color */
            padding: 10px;
            margin-bottom: 20px;  
        }
        .clips-container {
            text-align: center;
        }

    </style>
    <script>
        function playClip(videoPlayer, start, end) {
            videoPlayer.currentTime = start;
            videoPlayer.play();
            videoPlayer.ontimeupdate = function () {
                if (videoPlayer.currentTime >= end) {
                    videoPlayer.pause();
                }
            };
        }
    </script>
</head>
<body>
    <div class="container">
        <h1>Video Classification</h1>
        {% for video in duration_data.data %}
            {% set video_id = video.video_id %}
            {% set video_path = video_mapping[video_id] %}
            <div class="video-category">
                <h2>{{ video.classes[0].name }}</h2>
                <p>Video file: {{video_path}}</p>
                <div>
                    <span class="score">Score: {{ '%.2f'|format(video.classes[0].score) }}</span>
                    <div class="progress-bar">
                        <div class="progress" style="width: {{ video.classes[0].score }}%;"></div>
                    </div>
                </div>
                <div>
                    <span class="score">Duration: {{ '%.2f%%'|format(video.classes[0].duration_ratio * 100) }}</span>
                    <div class="progress-bar">
                        <div class="progress" style="width: {{ video.classes[0].duration_ratio * 100 }}%;"></div>
                    </div>
                </div>
                <video id="video-{{ video_id }}" width="480" height="320" controls>
                    <source src="{{ url_for('serve_video', filename=video_path) }}" type="video/mp4">
                    Your browser does not support the video tag.
                </video>
                <div class="clips-container">
                    <h3>Sample Clips</h3>
                    {% for clip in top_clips[video_id] %}
                    <button onclick="playClip(document.getElementById('video-{{ video_id }}'), {{ clip.start }}, {{ clip.end }})">Clip: {{ clip.prompt }} | Start {{ '%.2f'|format(clip.start) }}s - End {{ '%.2f'|format(clip.end) }}s - Score {{ '%.2f'|format(clip.score) }}</button>
                    {% endfor %}
            </div>
            <br /><br />
        {% endfor %}
    </div>
</body>
</html>
</code></pre>

<pre><code class="python">%run app.py
</code></pre>