デベロッパーハブ

from twelvelabs import TwelveLabs

from twelvelabs.indexes import IndexesCreateRequestModelsItem

from twelvelabs.tasks import TasksRetrieveResponse

client = TwelveLabs(api_key="<YOUR_API_KEY>")

# インデックスの作成

index = client.indexes.create(

index_name="<YOUR_INDEX_NAME>",

models=[

IndexesCreateRequestModelsItem(

model_name="pegasus1.2", model_options=["visual", "audio"]

)

]

)

print(f"Created index: id={index.id}")

# タスクの作成

task = client.tasks.create(

index_id=index.id, video_url="<YOUR_VIDEO_URL>")

print(f"Created task: id={task.id}")

# タスク更新時に呼び出されるコールバック関数

def on_task_update(task: TasksRetrieveResponse):

print(f" Status={task.status}")

# タスクの完了を待機

task = client.tasks.wait_for_done(task_id=task.id, callback=on_task_update)

if task.status != "ready":

raise RuntimeError(f"Indexing failed with status {task.status}")

print(

f"Upload complete. The unique identifier of your video is {task.video_id}.")

# ビデオの解析ストリームを開始

text_stream = client.analyze_stream(video_id=task.video_id, prompt="<YOUR_PROMPT>")

for text in text_stream:

if text.event_type == "text_generation":

print(text.text)