인공지능 비디오 검색의 시작: Twelve Labs API로 검색 쿼리 정밀 결합하기

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

<pre><code class="python">!pip install requests
!pip install flask

import os
import requests
import glob
from pprint import pprint

# Retrieve the URL of the API and my 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

# Specify the names of the indexes
index_names = ["formula_one", "tokyo_drift"]

# Create the indexes
index_id_formula_one = create_index(index_name = "formula_one", index_options=["visual", "conversation", "text_in_video", "logo"], engine = "marengo2.5")
index_id_tokyo_drift = create_index(index_name = "tokyo_drift", index_options=["visual", "conversation", "logo"], engine = "marengo2.5")

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

<pre><code class="language-plaintext">Status code: 201 - The request was successful and a new resource was created.
{'_id': '##38fb695b65d57eaecaf8##'}
Status code: 201 - The request was successful and a new resource was created.
{'_id': '##38fb695b65d57eaecaf8##'}
Created index IDs: ##38fb695b65d57eaecaf8##, ##38fb695b65d57eaecaf8##
</code></pre>

<pre><code class="python">TASKS_URL = f"{API_URL}/tasks"
TASK_ID_LIST = []
video_folder = 'static'  # folder containing the video files
INDEX_ID = index_id_tom  # change this to the other index id while creating the index for lex fridman podcast videos
# INDEX_ID = '##38d9c4e4225d1c0eb1e8##'

# Iterate through all the video files in the folder
for file_name in os.listdir(video_folder):
    # 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")
            continue

    #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")
</code></pre>

<pre><code class="language-plaintext">Entering task creation code for the file:  20211113T190000Z.mp4
Status code: 201 - The request was successful and a new resource was created.
File name: 20211113T190000Z.mp4
{'_id': '6438fb6e5b65d57eaecaf8bc'}


Entering task creation code for the file:  20211113T193300Z.mp4
Status code: 201 - The request was successful and a new resource was created.
File name: 20211113T193300Z.mp4
{'_id': '##38fb755b65d57eaecaf8##'}
</code></pre>

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

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":
            print(f"Task ID: {task_id}, Status code: {STATUS}")
            break   
        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"Remaining seconds: {remain_seconds}, Upload Percentage: {upload_percentage}")
        else:
             print('.', end='')
        time.sleep(10)
        
# Define starting time
start = time.time()
print("Starting to monitor...")

# Monitor the indexing process for all tasks
for task_id in TASK_ID_LIST:
    print("Current Task being monitored: ", task_id)
    monitor_upload_status(task_id)

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

<pre><code class="language-plaintext">Starting to monitor...
Current Task being monitored:  ##38fb6e5b65d57eaecaf8##
........Remaining seconds: 264.48919677734375, Upload Percentage: 0
Remaining seconds: 258.351806640625, Upload Percentage: 2
Remaining seconds: 253.1555633544922, Upload Percentage: 4
Remaining seconds: 247.93516540527344, Upload Percentage: 6
Remaining seconds: 242.26431274414062, Upload Percentage: 8
Remaining seconds: 237.22894287109375, Upload Percentage: 10
Remaining seconds: 231.01914978027344, Upload Percentage: 12
Remaining seconds: 224.7932891845703, Upload Percentage: 15
Remaining seconds: 218.599609375, Upload Percentage: 17
</code></pre>

<pre><code class="python"># List all the videos in an index
INDEX_ID='##38d9c4e4225d1c0eb1e8##'
INDEXES_VIDEOS_URL = f"{API_URL}/indexes/{INDEX_ID}/videos"
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']]

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

<pre><code class="bash">{'data': [{'_id': '##3d978c86daab572f3481##',
           'created_at': '2023-04-17T18:56:51Z',
           'metadata': {'duration': 1800.56,
                        'engine_id': 'marengo2.5',
                        'filename': '20211113T190000Z.mp4',
                        'fps': 25,
                        'height': 396,
                        'size': 415876158,
                        'width': 704},
           'updated_at': '2023-04-17T19:01:32Z'},
          {'_id': '##3d975786daab572f3481##',
           'created_at': '2023-04-17T18:56:44Z',
           'metadata': {'duration': 1800.56,
                        'engine_id': 'marengo2.5',
                        'filename': '20211114T170000Z.mp4',
                        'fps': 25,
                        'height': 396,
                        'size': 387273943,
                        'width': 704},
           'updated_at': '2023-04-17T19:00:39Z'},
          {'_id': '##3d972e86daab572f3481##',
           'created_at': '2023-04-17T18:56:38Z',
           'metadata': {'duration': 1800.56,
                        'engine_id': 'marengo2.5',
                        'filename': '20211113T193000Z.mp4',
                        'fps': 25,
                        'height': 396,
                        'size': 386209689,
                        'width': 704},
           'updated_at': '2023-04-17T18:59:58Z'},
          {'_id': '##3d96d386daab572f3481##',
           'created_at': '2023-04-17T18:56:28Z',
           'metadata': {'duration': 1800.52,
                        'engine_id': 'marengo2.5',
                        'filename': '20211121T133000Z.mp4',
                        'fps': 25,
                        'height': 396,
                        'size': 348611416,
                        'width': 704},
           'updated_at': '2023-04-17T18:58:27Z'},
          {'_id': '##3d96af86daab572f3481##',
           'created_at': '2023-04-17T18:56:08Z',
           'metadata': {'duration': 1800.56,
                        'engine_id': 'marengo2.5',
                        'filename': '20211113T200000Z.mp4',
                        'fps': 25,
                        'height': 396,
                        'size': 327766175,
                        'width': 704},
           'updated_at': '2023-04-17T18:57:51Z'}],
 'page_info': {'limit_per_page': 10,
               'page': 1,
               'total_duration': 9002.76,
               'total_page': 1,
               'total_results': 5}}

[{'video_id': '##3d978c86daab572f3481##', 'video_name': '20211113T190000Z.mp4'},
 {'video_id': '##3d975786daab572f3481##', 'video_name': '20211114T170000Z.mp4'},
 {'video_id': '##3d972e86daab572f3481##', 'video_name': '20211113T193000Z.mp4'},
 {'video_id': '##3d96d386daab572f3481##', 'video_name': '20211121T133000Z.mp4'},
 {'video_id': '##3d96af86daab572f3481##', 'video_name': '20211113T200000Z.mp4'}]
 </code></pre>

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

# Declare a dictionary named `data`
data = {
    "index_id": INDEX_ID,
    "search_options": ["visual"],
    "query": {
                "$and": [
                    {
                        "text": "winning trophy",
                        "search_options": ["visual"]
                    },
                    {
                        "text": "crypto.com",
                        "search_options": ["text_in_video"]
                    }
                ]
            }
        }
# Make a search request
response = requests.post(SEARCH_URL, headers=default_header, json=data)
if response.status_code == 200:
    print(f"Status code: {response.status_code} - Success")
else:
    print(f"Status code: {response.status_code}")
pprint(response.json())
</code></pre>

<pre><code class="bash">Status code: 200 - Success
{'data': [{'confidence': 'high',
           'end': 20,
           'score': 92.28,
           'start': 18,
           'video_id': '##3d96af86daab572f3481##'},
          {'confidence': 'high',
           'end': 43,
           'score': 92.28,
           'start': 42,
           'video_id': '##3d978c86daab572f3481##'},
          {'confidence': 'high',
           'end': 71,
           'score': 92.28,
           'start': 61,
           'video_id': '##3d978c86daab572f3481##'},
          {'confidence': 'high',
           'end': 62,
           'score': 92.28,
           'start': 61,
           'video_id': '##3d96af86daab572f3481##'},
          {'confidence': 'high',
           'end': 67,
           'score': 92.28,
           'start': 65,           
           'video_id': '##3d96af86daab572f3481##'}],
 'page_info': {'limit_per_page': 10,
               'next_page_token': '##69daa3-827f-4165-982d-ec0d34f97c7c-1',
               'page_expired_at': '2023-04-17T23:56:00Z',
               'total_results': 110},
 'search_pool': {'index_id': '##3d9556f607a5a7bd9ea5##',
                 'total_count': 5,
                 'total_duration': 9003}}
</code></pre>

<div style="position: relative; padding-bottom: 47.8125%; height: 0;"><iframe src="https://www.loom.com/embed/1a08de21a4d14f85ab3ee125660438da" frameborder="0" webkitallowfullscreen mozallowfullscreen allowfullscreen style="position: absolute; top: 0; left: 0; width: 100%; height: 100%;"></iframe></div>

<div style="position: relative; padding-bottom: 47.8125%; height: 0;"><iframe src="https://www.loom.com/embed/e2cb3c58f49c4b00b0c6c5b2f745acfc" frameborder="0" webkitallowfullscreen mozallowfullscreen allowfullscreen style="position: absolute; top: 0; left: 0; width: 100%; height: 100%;"></iframe></div>

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

# Declare a dictionary named `data`
data = {
    "index_id": INDEX_ID,
    "search_options": ["visual"],
    "query": {
                "$and": [
                    {
                        "text": "drift",
                        "search_options": ["visual"]
                    },
                    {
                        "text": "mitsubishi",
                        "search_options": ["logo"]
                    }
                ]
            }
        }
# Make a search request
response = requests.post(SEARCH_URL, headers=default_header, json=data)
if response.status_code == 200:
    print(f"Status code: {response.status_code} - Success")
else:
    print(f"Status code: {response.status_code}")
pprint(response.json())
</code></pre>

<pre><code class="bash">Status code: 200 - Success
{'data': [{'confidence': 'high',
           'end': 3710,
           'score': 92.28,
           'start': 3705,
           'video_id': '##3e3ace86daab572f3481##'}],
 'page_info': {'limit_per_page': 10,
               'page_expired_at': '2023-04-18T09:09:59Z',
               'total_results': 1},
 'search_pool': {'index_id': '##3e3647f607a5a7bd9ea5##',
                 'total_count': 1,
                 'total_duration': 6246}}
</code></pre>

<div style="position: relative; padding-bottom: 47.8125%; height: 0;"><iframe src="https://www.loom.com/embed/a7fc79ff424f4d50b7a42dc2bd134473" frameborder="0" webkitallowfullscreen mozallowfullscreen allowfullscreen style="position: absolute; top: 0; left: 0; width: 100%; height: 100%;"></iframe></div>

<pre><code class="python">response_data = response.json()  

# Extract unique video IDs
unique_video_ids = list(set([item['video_id'] for item in response_data['data']]))

# Create empty start and end instances lists for each video ID
video_start_end_lists = {video_id: {'starts': [], 'ends': []} for video_id in unique_video_ids}

def find_video_name(video_id, video_id_name_list):
    for video in video_id_name_list:
        if video['video_id'] == video_id:
            return video['video_name']
    return None

# Append start and end instances to their respective lists
for item in response_data['data']:
    video_id = item['video_id']
    video_start_end_lists[video_id]['starts'].append(item['start'])
    video_start_end_lists[video_id]['ends'].append(item['end'])

for video_id, timestamps in video_start_end_lists.items():
    video_name = find_video_name(video_id, video_id_name_list)
    if video_name:
        timestamps['video_name'] = video_name
    else:
        print(f"No video name found for ID '{video_id}'")

# Print the result
pprint(video_start_end_lists)
</code></pre>

<pre><code class="bash">{'##3d96af86daab572f3481##': {'ends': [20, 62, 67, 114],
                              'starts': [18, 61, 65, 111],
                              'video_name': '20211113T200000Z.mp4'},
 '643d972e86daab572f34810d': {'ends': [84],
                              'starts': [68],
                              'video_name': '20211113T193000Z.mp4'},
 '643d975786daab572f34810e': {'ends': [79],
                              'starts': [70],
                              'video_name': '20211114T170000Z.mp4'},
 '643d978c86daab572f34810f': {'ends': [43, 71, 85, 95],
                              'starts': [42, 61, 84, 91],
                              'video_name': '20211113T190000Z.mp4'}}
</code></pre>

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

with open("lists.pkl", "wb") as f:
    pickle.dump(video_start_end_lists, f)
</code></pre>

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

<pre><code class="python">
from flask import Flask, render_template
import pickle

with open("lists.pkl", "rb") as f:
   video_start_end_lists = pickle.load(f)

app = Flask(__name__)

@app.route("/")
def index():
    return render_template("index_local.html", video_start_end_lists=video_start_end_lists)

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

<pre><code class="language-html"><!DOCTYPE html>
    <html>
      <head>
        <link rel="shortcut icon" href="#" />
        <style>
          body {
            background-color: #FFE0B2; /* Light Orange */
            font-family: Arial, sans-serif;
            text-align: center;
            margin: 0;
          }
          h1 {
            font-size: 3em;
            color: #000000; /* Black */
            background-color: #9ACD32; /* Light Green */
            padding: 20px;
            margin: 0;
          }
          h2 {
            font-size: 2em;
            color: #000000; /* Black */
            margin-bottom: 20px;
            text-align: left;
            padding-left: 20px;
          }
          .video-container {
            display: flex;
            flex-wrap: wrap;
            padding: 2px;
            justify-content: space-evenly;
            gap: 2px;
          }
          .video-item {
            display: flex;
            flex-direction: column;
            align-items: center;
            width: 45%;
            height: 450px;
            margin: 20px;
            text-align: center;
            background-color: #FFFFFF;
            box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
            border-radius: 10px;
            padding: 20px;
          }
          .video-item video {
            width: 100%;
            height: 380px;
            margin: 0; /* Remove bottom margin */
            border-radius: 5px;
          }
          .video-item p {
            font-size: 16px;
            margin-top: 10px;
            font-weight: bold;
            color: #212121; /* Dark Grey */
          }
          .video-item span {
            color: #9ACD32; /* Light Green */
          }
        </style>
      </head>
      <body>
        <h1>My Favorite Scenes</h1>
        {% for video_id, segments in video_start_end_lists.items() %}
        <div class="video-section">
          <h2>Scenes from {{segments['video_name']}}</h2>
          <div class="video-container">
            {% for i in range(segments['starts']|length) %}
              <div class="video-item">
                <video id="video_{{ video_id }}_{{ i }}" width="560" height="315" controls>
                  <source src="{{ url_for('static', filename= segments['video_name']) }}" type="video/mp4">
                  Your browser does not support the video tag.
                </video>
                <p>Start: <span>{{ segments['starts'][i] }}</span> | End: <span>{{ segments['ends'][i] }}</span></p>
                <script>
                  document.getElementById("video_{{ video_id }}_{{ i }}").addEventListener("loadedmetadata", function() {
                    this.currentTime = {{ segments['starts'][i] }};
                  });
                  document.getElementById("video_{{ video_id }}_{{ i }}").addEventListener("timeupdate", function() {
                    if (this.currentTime >= {{ segments['ends'][i] }}) {
                      this.pause();
                    }
                  });
                </script>
              </div>
            {% endfor %}
          </div>
        </div>
        {% endfor %}
      </body>
    </html>
</code></pre>