Twelve Labs APIを使用して動画内からブランドロゴを見つける方法

<pre><code class="python">%env API_URL = https://api.twelvelabs.io/v1.1
%env API_KEY= <あなたのAPIキー>

!pip install requests

import os
import requests
import glob
from pprint import pprint

# APIのURLと自分のAPIキーを取得
API_URL = os.getenv("API_URL")
assert API_URL

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

<pre><code class="python"># `/indexes` エンドポイントのURLを構築
INDEXES_URL = f"{API_URL}/indexes"

# リクエストのヘッダーを設定
default_header = {
    "x-api-key": API_KEY
}

# 指定された名前でインデックスを作成する関数を定義
def create_index(index_name, index_options, engine):
    # data という名前の辞書を宣言
    data = {
        "engine_id": engine,
        "index_options": index_options,
        "index_name": index_name,
    }

    # インデックスを作成
    response = requests.post(INDEXES_URL, headers=default_header, json=data)

    # インデックスの一意の識別子を保存
    INDEX_ID = response.json().get('_id')

    # ステータスコードが201であるか確認し、成功を出力
    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


# インデックスを作成
index_id = create_index(index_name = "extract_text", index_options=["logo"], engine = "marengo2.5")

# 作成されたインデックスIDを出力
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'  # 動画ファイルを含むフォルダー

def upload_video(file_name):
    # インデックスに動画が既に存在するか検証
    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

    # 動画がまだインデックスに存在しない場合、現在の動画のインデックス作成用の新しいタスク作成コードに進む
    print("Entering task creation code for the file: ", file_name)
    
    if file_name.endswith('.mp4'):  # ファイルがMP4動画であることを確認
        file_path = os.path.join(video_folder, file_name)  # 動画ファイルのフルパスを取得
        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")),] #動画は動画ファイル自体と同じ名前でプラットフォーム上にインデックス登録されます。
            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)
            # ステータスコードが201であるか確認し、成功を出力
            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")

# 動画ファイルのリストを取得
video_files = [f for f in os.listdir(video_folder) if f.endswith('.mp4')]

# ThreadPoolExecutorを作成
with ThreadPoolExecutor() as executor:
    # 全動画ファイルに対してupload_videoを並行処理するようexecutorを使用
    executor.map(upload_video, video_files)
      </code></pre>

<pre><code class="python"># インデックス内のすべての動画を一覧表示する
default_header = {
    "x-api-key": API_KEY
}
INDEX_ID='##4a73aa8b1dd6cde172a9##'
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': '##d978c86daab572f3481##',
           '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">VIDEO_ID = '###a849b86daab572f349242'
LOGO_URL = f"{API_URL}/indexes/{INDEX_ID}/videos/{VIDEO_ID}/logo"

# GETリクエストを実行
response = requests.get(LOGO_URL, headers=default_header)
print (f"Status code: {response.status_code}")
logo_data = response.json()
pprint (logo_data)
  </code></pre>

<pre><code class="python">Status code: 200
{'data': [{'end': 16, 'start': 15, 'value': 'Ducati Corse'},
          {'end': 23, 'start': 22, 'value': 'Bank of Jordan'},
          {'end': 23, 'start': 22, 'value': 'Han Chiang High School'},
          {'end': 24, 'start': 23, 'value': 'Peugeot'},
          {'end': 25, 'start': 24, 'value': 'Dr Lal Path Labs'},
          {'end': 26, 'start': 25, 'value': 'Z8Games'},
          {'end': 29, 'start': 27, 'value': 'Sky Sports'},
          {'end': 29, 'start': 28, 'value': 'Tout'},
          {'end': 31, 'start': 30, 'value': 'Sky UK'},
          {'end': 31, 'start': 30, 'value': 'New Balance'},
          {'end': 31, 'start': 30, 'value': 'Industria'},
          {'end': 33, 'start': 32, 'value': 'Esport3'},
          {'end': 35, 'start': 32, 'value': 'Nissan'},
          {'end': 33, 'start': 32, 'value': 'GoCar'},
          {'end': 34, 'start': 33, 'value': 'Land Bank of the Philippines'},
          {'end': 34, 'start': 33, 'value': 'Z8Games'},
          {'end': 37, 'start': 36, 'value': 'Tout'},
          {'end': 37, 'start': 36, 'value': 'Zazzle'},
          {'end': 39, 'start': 38, 'value': 'Z8Games'},
          {'end': 39, 'start': 38, 'value': 'Giochi Preziosi'},
          {'end': 41, 'start': 40, 'value': 'Mini'}],
 'id': '###a849b86daab572f349242',
 'index_id': '###a73aa8b1dd6cde172a933'}
 </code></pre>

<pre><code class="python"># `/search` エンドポイントのURLを構築
SEARCH_URL = f"{API_URL}/search/"

# `data` という名前の辞書を宣言
data = {
    "index_id": INDEX_ID,
    "query": "honda",
    "search_options": [
        "logo"
    ]
}

# 後でflaskアプリケーションに渡すためにクエリを抽出
input_query = data["query"]

# 検索リクエストを実行
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="python">Status code: 200 - Success
{'data': [{'confidence': 'high',
           'end': 19,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 18,
           'video_id': '###d96af86daab572f348###'},
          {'confidence': 'high',
           'end': 104,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 103,
           'video_id': '###d978c86daab572f348###'},
          {'confidence': 'high',
           'end': 137,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 136,
           'video_id': '###d978c86daab572f348###'},
          {'confidence': 'high',
           'end': 269,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 268,
           'video_id': '###d96af86daab572f348###'},
          {'confidence': 'high',
           'end': 392,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 391,
           'video_id': '###d96af86daab572f348###'},
          {'confidence': 'high',
           'end': 478,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 477,
           'video_id': '###d975786daab572f348###'},
          {'confidence': 'high',
           'end': 483,
           'metadata': [{'text': 'Honda', 'type': 'logo'}],
           'score': 92.28,
           'start': 482,
           'video_id': '###d978c86daab572f348###'}],
 'page_info': {'limit_per_page': 10,
               'next_page_token': 'fc3c420a-5971-428a-8796-9e5c077754c0-1',
               'page_expired_at': '2023-05-26T18:40:26Z',
               'total_results': 47},
 'search_pool': {'index_id': '###d9556f607a5a7bd9ea###',
                 'total_count': 5,
                 'total_duration': 9003}}
                 </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">
{'###d96af86daab572f348###': [{'confidence': 'high',
                               'end': 19,
                               'start': 18,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 104,
                               'start': 103,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 137,
                               'start': 136,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 269,
                               'start': 268,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 392,
                               'start': 391,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 478,
                               'start': 477,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 483,
                               'start': 482,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 491,
                               'start': 487,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 561,
                               'start': 560,
                               'text': 'Honda'},
                              {'confidence': 'high',
                               'end': 586,
                               'start': 585,
                               'text': 'Honda'}]}
</code></pre>

<pre><code class="python">video_id = ocr_data.get('id')
data_list = logo_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

# データをpickleファイルに保存
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__)

# pickleファイルからデータをロード
with open('data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)

# データへのアクセス
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>ロゴ検出</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;
        }

        /* 検索スタイル */
        .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>動画全体のロゴ検出</h1>
    <h3>動画ファイル: <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">
        お使いのブラウザはビデオタグをサポートしていません。
    </video>
    <br /> <br /> <br />
    <table>
        <tr>
            <th>開始</th>
            <th>終了</th>
            <th>値</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>ロゴ検索結果</h1>  
        <h2>動画ファイル: <i>{{ video_id_name_dict[video_id] }}</i></h2>
        <h2>入力されたクエリ: <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">
            お使いのブラウザはビデオタグをサポートしていません。
        </video>
        <table>
            <tr>
                <th>開始</th>
                <th>終了</th>
                <th>信頼度</th>
                <th>テキスト</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>