1
from twelvelabs import TwelveLabs
2
from twelvelabs.indexes import IndexesCreateRequestModelsItem
3
from twelvelabs.tasks import TasksRetrieveResponse
4
5
# 1. Initialize the client
6
client = TwelveLabs(api_key="<YOUR_API_KEY>")
7
8
# 2. Create an index
9
index = client.indexes.create(
10
    index_name="<YOUR_INDEX_NAME>",
11
    models=[
12
        IndexesCreateRequestModelsItem(
13
            model_name="marengo2.7",
14
            model_options=["visual", "audio"]
15
        )
16
    ]
17
)
18
if index.id is None:
19
    raise RuntimeError("Failed to create an index.")
20
print(f"Created index: id={index.id}")
21
22
# 3. Upload a video
23
task = client.tasks.create(
24
    index_id=index.id, video_url="<YOUR_VIDEO_URL>")
25
print(f"Created task: id={task.id}")
26
27
# 4. Monitor the indexing process
28
def on_task_update(task: TasksRetrieveResponse):
29
    print(f"  Status={task.status}")
30
task = client.tasks.wait_for_done(task_id=task.id, callback=on_task_update)
31
if task.status != "ready":
32
    raise RuntimeError(f"Indexing failed with status {task.status}")
33
print(
34
    f"Upload complete. The unique identifier of your video is {task.video_id}.")
35
36
# 5. Perform a search request
37
search_pager = client.search.query(
38
    index_id=index.id, query_text="<YOUR_QUERY>", search_options=["visual", "audio"],)
39
40
# 6. Process the search results
41
print("Search results:")
42
for clip in search_pager:
43
    print(
44
        f" video_id {clip.video_id} score={clip.score} start={clip.start} end={clip.end} confidence={clip.confidence}"
45
    )