Twelve Labs APIを使用してビデオOCRを実行する方法

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

!pip install requests

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


# Create the indexes
INDEX_ID = create_index(index_name = "extract_text", index_options=["text_in_video"], engine = "marengo2.5")

# Print the created index IDs
print(f"Created index IDs: {INDEX_ID}")
  </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 = 'static'  # folder containing the video files

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"># List all the videos in an index
default_header = {
    "x-api-key": API_KEY
}
INDEX_ID='644a73aa8b1dd6cde172a933'
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']]
pprint(video_id_name_list)
  </code></pre>

<pre><code class="python">{'data': [{'_id': '###a917186daab572f349243',
           'created_at': '2023-04-27T14:18:48Z',
           'metadata': {'duration': 1300.173875,
                        'engine_id': 'marengo2.5',
                        'filename': 'A Brief History of Film.mp4',
                        'fps': 23.976023976023978,
                        'height': 720,
                        'size': 188214297,
                        'width': 1280},
           'updated_at': '2023-04-27T14:20:11Z'},
          {'_id': '###3da86daab572f349241',
           'created_at': '2023-04-27T13:08:19Z',
           'metadata': {'duration': 550.7,
                        'engine_id': 'marengo2.5',
                        'filename': 'GPT - Explained!.mp4',
                        'fps': 30,
                        'height': 720,
                        'size': 22838593,
                        'width': 1152},
           'updated_at': '2023-04-27T13:08:42Z'}],
 'page_info': {'limit_per_page': 10,
               'page': 1,
               'total_duration': 5402.873875,
               'total_page': 1,
               'total_results': 3}}

[{'video_id': '###a849b86daab572f349242',
  'video_name': 'A Brief History of Film.mp4'},
 {'video_id': '###a73da86daab572f349241', 'video_name': 'GPT - Explained!.mp4'}]
   </code></pre>

<pre><code class="python">VIDEO_ID = '###a849b86daab572f349242'
TEXT_IN_VIDEO_URL = f"{API_URL}/indexes/{INDEX_ID}/videos/{VIDEO_ID}/text-in-video"

response = requests.get(TEXT_IN_VIDEO_URL, headers=default_header)
print (f"Status code: {response.status_code}")
ocr_data = response.json()
pprint (ocr_data)
  </code></pre>

<pre><code class="python">Status code: 200
{'data': [{'end': 3, 'start': 1, 'value': 'Film Thought Project'},
          {'end': 6, 'start': 5, 'value': 'Film'},
          {'end': 22,
           'start': 18,
           'value': "'L'arrivée d'un train en gare de La Ciotat"},
          {'end': 28, 'start': 18, 'value': 'Year:'},
          {'end': 28, 'start': 23, 'value': '2015'},
          {'end': 28, 'start': 23, 'value': 'Production Co.'},
          {'end': 28, 'start': 23, 'value': 'Alejandro G. Iñárritu'},
          {'end': 28, 'start': 23, 'value': 'Regency Enterprises'},
          {'end': 28, 'start': 23, 'value': "'The Revenant'"},
          {'end': 30, 'start': 29, 'value': "Let's"},
          {'end': 40, 'start': 32, 'value': 'Film:'},
          {'end': 34, 'start': 33, 'value': 'Film Thought Project'},
          {'end': 40, 'start': 35, 'value': 'Director:'},
          {'end': 40, 'start': 35, 'value': 'Production Co.'},
          {'end': 40, 'start': 36, 'value': 'Alfred Hitchcock'},
          {'end': 40, 'start': 36, 'value': '1958'},
          {'end': 40, 'start': 36, 'value': 'Alfred J. Hitchcock Productions'},
          {'end': 40, 'start': 37, 'value': 'Year:'},
          {'end': 40, 'start': 38, 'value': "'Vertigo'"},
          {'end': 45, 'start': 44, 'value': 'PRESS START'},
          {'end': 46, 'start': 45, 'value': '2020'},
          {'end': 47, 'start': 46, 'value': '2018'},
          {'end': 48, 'start': 47, 'value': '1975'},
          {'end': 53, 'start': 49, 'value': '1870s'},
          {'end': 61, 'start': 67, 'value': 'Eadweard Muybridge'},
          {'end': 69, 'start': 75, 'value': 'See you soon'}],
 'id': '###a849b86daab572f349242',
 'index_id': '###a73aa8b1dd6cde172a933'}
 </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,
    "query": "horse",
    "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())
search_data = response.json()
</code></pre>

<pre><code class="python">Status code: 200 - Success
{'data': [{'confidence': 'high',
           'end': 64,
           'metadata': [{'text': 'THE HORSE IN MOTION.',
                         'type': 'text_in_video'}],
           'score': 92.28,
           'start': 63,
           'video_id': '###a849b86daab572f349242'},
          {'confidence': 'high',
           'end': 91,
           'metadata': [{'text': 'THE HORSE IN MOTION.',
                         'type': 'text_in_video'}],
           'score': 92.28,
           'start': 88,
           'video_id': '###a849b86daab572f349242'}],
 'page_info': {'limit_per_page': 10,
               'page_expired_at': '2023-05-12T00:03:43Z',
               'total_results': 2},
 'search_pool': {'index_id': '###a73aa8b1dd6cde172a933',
                 'total_count': 3,
                 'total_duration': 5403}}
                 </code></pre>

<pre><code class="python">video_data = [{'start': d['start'], 'end': d['end'], 'confidence': d['confidence'], 'text': d['metadata'][0]['text']} for d in search_data['data']]
video_search_dict = {}

for vd in video_data:
    if search_data['data'][0]['video_id'] in video_search_dict:
        video_search_dict[search_data['data'][0]['video_id']].append(vd)
    else:
        video_search_dict[search_data['data'][0]['video_id']] = [vd]

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

<pre><code class="python">
{'###a849b86daab572f349242': [{'confidence': 'high',
                               'end': 64,
                               'start': 63,
                               'text': 'THE HORSE IN MOTION.'},
                              {'confidence': 'high',
                               'end': 91,
                               'start': 88,
                               'text': 'THE HORSE IN MOTION.'}]}
</code></pre>

<pre><code class="python">video_id = ocr_data.get('id')
data_list = ocr_data.get('data')

data_to_save = {
    'video_id': video_id,
    'data_list': data_list,
    'video_id_name_list': video_id_name_list,
    'video_search_dict': video_search_dict
}

import pickle

# Save data to a pickle file
with open('data.pkl', 'wb') as f:
    pickle.dump(data_to_save, f)
    </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__)

# Load data from a pickle file
with open('data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)

# Access the data
video_id = loaded_data['video_id']
data_list = loaded_data['data_list']
video_id_name_list = loaded_data['video_id_name_list']
video_search_dict = loaded_data['video_search_dict']

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

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

@app.route('/')
def home():
    for item in data_list:
        if ":" not in str(item['start']):
            item['start'] = int(item['start'])
            item['start'] = f"{item['start'] // 60}:{item['start'] % 60:02}"
        if ":" not in str(item['end']):
            item['end'] = int(item['end'])
            item['end'] = f"{item['end'] // 60}:{item['end'] % 60:02}"


    video_id_name_dict = {video['video_id']: video['video_name'] for video in video_id_name_list}
    # video_name = video_id_name_dict.get(video_id)
    return render_template('index.html', data=data_list[:10], video_id_name_dict=video_id_name_dict, video_id=video_id, video_search_dict = video_search_dict)

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

<pre><code class="language-html"><!DOCTYPE html>
<html>
<head>
    <link rel="shortcut icon" href="#" />
    <title>Video OCR</title>
    <style>
        body {
            text-align: center;
            font-family: Arial, sans-serif;
            color: #333;
            background-color: #f5f5f5;
        }
        h1, h2 {
            color: #444;
        }
        table {
            margin: 0 auto;
            border-collapse: collapse;
            width: 80%;
            margin-top: 20px;
        }
        th, td {
            border: 1px solid #ddd;
            padding: 8px;
            text-align: center;
        }
        th {
            padding-top: 12px;
            padding-bottom: 12px;
            text-decoration: underline;
            color: black;
        }
        video {
            width: 40%;
            height: auto;
            margin-top: 20px;
        }

        /* search style */
        .video-container {
            text-align: center;
            margin-bottom: 2em;
            padding: 1em;
            background-color: #fff;
            border: 1px solid #ddd;
            border-radius: 4px;
            box-shadow: 0 2px 4px rgba(0,0,0,0.1);
        }
        table {
            margin: 0 auto;
            margin-bottom: 1em;
        }
        th, td {
            padding: 0.5em;
            border: 1px solid #ddd;
        }
    </style>
    <script>
        function playVideo(timeString) {
            var timeParts = timeString.split(":");
            var time = parseInt(timeParts[0]) * 60 + parseInt(timeParts[1]);
            var video = document.querySelector('#mainVideo');
            video.currentTime = time;
            video.play();
        }
    </script>    
</head>

<body>
    <h1>Video OCR</h1>
    <h3>Video file: <i>{{ video_id_name_dict[video_id]}}</i></h3>
    <video id="mainVideo" controls>
        <source src="{{ url_for('static', filename=video_id_name_dict[video_id]|string) }}" type="video/mp4">
        Your browser does not support the video tag.
    </video>
    <br /> <br /> <br />
    <table>
        <tr>
            <th>Start</th>
            <th>End</th>
            <th>Value</th>
        </tr>
        {% for item in data %}
        <tr>
            <td><a href="javascript:void(0)" onclick="playVideo('{{ item['start'] }}')">{{ item['start'] }}</a></td>
            <td>{{ item['end'] }}</td>
            <td>{{ item['value'] }}</td>
        </tr>
        {% endfor %}
    </table>
    <br /> <br />
      
    {% for video_id, results in video_search_dict.items() %}
    <div class="video-container">
        <h1>Text-in-video Search Results</h1>  
        <h2>Video file: <i>{{ video_id_name_dict[video_id] }}</i></h2>
        <h2>Entered query: <i>{{input_query}}</i></h2>
        {% for result in results %}
        <video controls preload="metadata" style="width: 40%;">
            <source src="{{ url_for('static', filename=video_id_name_dict[video_id]) }}#t={{ result['start'] }},{{ result['end'] }}" type="video/mp4">
            Your browser does not support the video tag.
        </video>
        <table>
            <tr>
                <th>Start</th>
                <th>End</th>
                <th>Confidence</th>
                <th>Text</th>
            </tr>
            <tr>
                <td>{{ result['start'] }}</td>
                <td>{{ result['end'] }}</td>
                <td>{{ result['confidence'] }}</td>
                <td>{{ result['text'] }}</td>
            </tr>
        </table>
        {% endfor %}
    </div>
    {% endfor %}
</body>
</html>
</code></pre>

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