1
def query(self, query_text_list, file_name=None):
2
   all_results = []  # List of lists of dictionaries
3
4
5
   for query_text in query_text_list:
6
       search_results = self.client.search.query(
7
           index_id=os.getenv("INDEX_ID"),
8
           query_text=query_text,
9
           # query={"text": query_text},
10
          
11
           # query = {
12
           #     "$not": {
13
           #         "origin": {
14
           #             "text": query_text
15
           #         },
16
           #         "sub": {
17
           #             "$or": [
18
           #                 {
19
           #                     "text": "News Anchor"
20
           #                 },
21
           #                 {
22
           #                     "text": "Podcast"
23
           #                 }
24
           #             ]
25
           #         }
26
           #     }
27
           # },
28
29
30
           options=["visual","audio"]
31
       )
32
      
33
       query_result = []  # List of dictionaries for this query
34
       while True:
35
          
36
           try:
37
               # Retrieve each page of search results
38
               page_data = next(search_results)
39
               for clip in page_data:  # Limiting to 3 results per page
40
                   query_result.append({
41
                       "video_id": clip.video_id,
42
                       "score": clip.score,
43
                       "start": clip.start,
44
                       "end": clip.end,
45
                       "confidence": clip.confidence,
46
                       # "metadata": clip.metadata,
47
                   })
48
           except StopIteration:
49
               break
50
       all_results.append(query_result)  # Append this query's results as a list
51
52
53
   if file_name:
54
       self.save_json(all_results, file_name)
55
56
57
   return all_results  # List of lists of dictionaries

1
def get_video_info(video_id):
2
   api_key= os.getenv("TWELVE_LABS_KEY")
3
4
5
   index_id=os.getenv("INDEX_ID")
6
   url = f"https://api.twelvelabs.io/v1.3/indexes/{index_id}/videos/{video_id}"
7
   headers = {
8
       "x-api-key": api_key,
9
       "Content-Type": "application/json",
10
   }
11
  
12
   response = requests.get(url, headers=headers)
13
14
15
   if response.status_code == 200:
16
       video_info = response.json()  # Parse response as JSON
17
       return video_info
18
   elif response.status_code == 400:
19
       print("The request has failed. Check your parameters.")
20
       return None
21
   else:
22
       print(f"Error: Status code {response.status_code}")
23
       return None

1
    def __init__(self):
2
        self.api_url = os.getenv("TWELVE_LABS_BASE_URL")
3
        self.headers = {"x-api-key": os.getenv("TWELVE_LABS_API_KEY")}
4
5
    def index_youtube_video(self, index_id: str, youtube_url: str):
6
        task_url = f"{self.api_url}/tasks/external-provider"
7
        data = {
8
            "index_id": index_id,
9
            "url": youtube_url,
10
        }
11
        response = requests.post(task_url, headers=self.headers, json=data)
12
        # you can get the video ID from video_id
13
        return response.json()
14
15
    def highlight_video(self, body: HighlightVideoBody) -> HighlightVideoResponse:
16
        url = f"{self.api_url}/summarize"
17
        data = body.dict()
18
        response = requests.post(url, headers=self.headers, json=data)
19
        return HighlightVideoResponse(**response.json())
20
21
    def summarize(self, body: HighlightVideoBody) -> HighlightVideoResponse:
22
        url = f"{self.api_url}/summarize"
23
        data = body.dict()
24
        response = requests.post(url, headers=self.headers, json=data)
25
        return response.json()