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,
25
    video_url="<YOUR_VIDEO_URL>" # Or use video_file to upload a file from the local file system
26
    )
27
print(f"Created task: id={task.id}")
28
29
# 4. Monitor the indexing process
30
def on_task_update(task: TasksRetrieveResponse):
31
    print(f"  Status={task.status}")
32
task = client.tasks.wait_for_done(task_id=task.id, callback=on_task_update)
33
if task.status != "ready":
34
    raise RuntimeError(f"Indexing failed with status {task.status}")
35
print(
36
    f"Upload complete. The unique identifier of your video is {task.video_id}.")
37
38
# 5. Perform a search request
39
search_pager = client.search.query(
40
    index_id=index.id, query_text="<YOUR_QUERY>", search_options=["visual", "audio"],)
41
42
# 6. Process the search results
43
print("Search results:")
44
for clip in search_pager:
45
    print(
46
        f" video_id {clip.video_id} score={clip.score} start={clip.start} end={clip.end} confidence={clip.confidence}"
47
    )