1  """
  2  Exemples d'appels à l'API d'analyse vidéo
  3  ------------------------------------------
  4  
  5  Ce script démontre comment appeler les nouveaux endpoints d'analyse vidéo
  6  ajoutés à l'API d'inférence multi-session.
  7  """
  8  
  9  import requests
 10  import json
 11  import time
 12  import os
 13  from pprint import pprint
 14  
 15  # Configuration
 16  API_URL = "http://localhost:8000"  # URL de l'API
 17  API_KEY = "votre_clé_api"  # Remplacer par une clé API valide
 18  
 19  # Chemins des fichiers vidéo à analyser
 20  VIDEO_FILE_MANIPULATION = "path/to/your/video_manipulation_sample.mp4"
 21  VIDEO_FILE_NONVERBAL = "path/to/your/nonverbal_sample.mp4"
 22  
 23  def check_api_health():
 24      """Vérifie si l'API est en fonctionnement"""
 25      response = requests.get(f"{API_URL}/api/health")
 26      if response.status_code == 200:
 27          print("✅ L'API est opérationnelle")
 28          return True
 29      else:
 30          print("❌ L'API n'est pas accessible")
 31          return False
 32  
 33  def analyze_video_manipulation(video_path):
 34      """Soumet une vidéo pour analyse des stratégies de manipulation"""
 35      headers = {
 36          "X-API-Key": API_KEY
 37      }
 38      
 39      with open(video_path, "rb") as video_file:
 40          files = {
 41              "file": (os.path.basename(video_path), video_file, "video/mp4")
 42          }
 43          data = {
 44              "keep_video": "false"
 45          }
 46          
 47          print("📤 Envoi de la vidéo pour analyse des stratégies de manipulation...")
 48          response = requests.post(
 49              f"{API_URL}/api/video/manipulation-analysis",
 50              headers=headers,
 51              files=files,
 52              data=data
 53          )
 54      
 55      if response.status_code != 200:
 56          print(f"❌ Erreur: {response.status_code} - {response.text}")
 57          return None
 58      
 59      result = response.json()
 60      print(f"✅ Tâche créée avec ID: {result['task_id']}")
 61      
 62      return result["task_id"]
 63  
 64  def analyze_nonverbal_behavior(video_path):
 65      """Soumet une vidéo pour analyse des comportements non-verbaux"""
 66      headers = {
 67          "X-API-Key": API_KEY
 68      }
 69      
 70      with open(video_path, "rb") as video_file:
 71          files = {
 72              "file": (os.path.basename(video_path), video_file, "video/mp4")
 73          }
 74          data = {
 75              "keep_video": "false"
 76          }
 77          
 78          print("📤 Envoi de la vidéo pour analyse des comportements non-verbaux...")
 79          response = requests.post(
 80              f"{API_URL}/api/video/nonverbal-analysis",
 81              headers=headers,
 82              files=files,
 83              data=data
 84          )
 85      
 86      if response.status_code != 200:
 87          print(f"❌ Erreur: {response.status_code} - {response.text}")
 88          return None
 89      
 90      result = response.json()
 91      print(f"✅ Tâche créée avec ID: {result['task_id']}")
 92      
 93      return result["task_id"]
 94  
 95  def check_task_status(task_id):
 96      """Vérifie l'état d'une tâche d'analyse vidéo"""
 97      headers = {
 98          "X-API-Key": API_KEY
 99      }
100      
101      response = requests.get(
102          f"{API_URL}/api/video/tasks/{task_id}",
103          headers=headers
104      )
105      
106      if response.status_code != 200:
107          print(f"❌ Erreur: {response.status_code} - {response.text}")
108          return None
109      
110      return response.json()
111  
112  def wait_for_completion(task_id, max_wait_time=600):
113      """Attend la fin d'une tâche d'analyse vidéo"""
114      print(f"⏳ Attente de la fin de la tâche {task_id}...")
115      
116      start_time = time.time()
117      
118      while time.time() - start_time < max_wait_time:
119          task_info = check_task_status(task_id)
120          
121          if not task_info:
122              print("❌ Impossible de récupérer l'état de la tâche")
123              return None
124          
125          status = task_info["status"]
126          progress = task_info.get("progress", 0)
127          message = task_info.get("message", "")
128          
129          print(f"📊 Statut: {status} - Progression: {progress:.1f}% - Message: {message}")
130          
131          if status == "completed":
132              print("✅ Traitement terminé!")
133              return task_info
134          
135          if status == "failed":
136              print(f"❌ Échec du traitement: {task_info.get('error', 'Erreur inconnue')}")
137              return task_info
138          
139          # Attendre avant de vérifier à nouveau
140          time.sleep(10)
141      
142      print(f"⚠️ Délai d'attente dépassé ({max_wait_time} secondes)")
143      return None
144  
145  def list_video_tasks():
146      """Liste les tâches d'analyse vidéo"""
147      headers = {
148          "X-API-Key": API_KEY
149      }
150      
151      response = requests.get(
152          f"{API_URL}/api/video/tasks",
153          headers=headers
154      )
155      
156      if response.status_code != 200:
157          print(f"❌ Erreur: {response.status_code} - {response.text}")
158          return None
159      
160      tasks = response.json()
161      print(f"📋 Nombre total de tâches: {tasks['total']}")
162      
163      for task_id, task_info in tasks.get("tasks", {}).items():
164          print(f"- {task_id}: {task_info['status']} ({task_info.get('type', 'unknown')})")
165      
166      return tasks
167  
168  def delete_task(task_id):
169      """Supprime une tâche d'analyse vidéo"""
170      headers = {
171          "X-API-Key": API_KEY
172      }
173      
174      response = requests.delete(
175          f"{API_URL}/api/video/tasks/{task_id}",
176          headers=headers
177      )
178      
179      if response.status_code != 200:
180          print(f"❌ Erreur: {response.status_code} - {response.text}")
181          return False
182      
183      print(f"🗑️ Tâche {task_id} supprimée avec succès")
184      return True
185  
186  def save_results(task_info, output_file):
187      """Sauvegarde les résultats d'une tâche d'analyse vidéo"""
188      if "results" not in task_info:
189          print("❌ Aucun résultat disponible")
190          return False
191      
192      with open(output_file, "w", encoding="utf-8") as f:
193          json.dump(task_info["results"], f, ensure_ascii=False, indent=2)
194      
195      print(f"💾 Résultats sauvegardés dans {output_file}")
196      return True
197  
198  def main():
199      """Fonction principale pour démontrer l'utilisation de l'API d'analyse vidéo"""
200      # Vérifier l'état de l'API
201      if not check_api_health():
202          return
203      
204      # Exemple 1: Analyse des stratégies de manipulation
205      if os.path.exists(VIDEO_FILE_MANIPULATION):
206          print("\n📹 Exemple 1: Analyse des stratégies de manipulation vidéo")
207          task_id = analyze_video_manipulation(VIDEO_FILE_MANIPULATION)
208          
209          if task_id:
210              task_info = wait_for_completion(task_id)
211              
212              if task_info and task_info["status"] == "completed":
213                  save_results(task_info, "manipulation_analysis_results.json")
214      else:
215          print(f"⚠️ Fichier vidéo non trouvé: {VIDEO_FILE_MANIPULATION}")
216      
217      # Exemple 2: Analyse des comportements non-verbaux
218      if os.path.exists(VIDEO_FILE_NONVERBAL):
219          print("\n📹 Exemple 2: Analyse des comportements non-verbaux")
220          task_id = analyze_nonverbal_behavior(VIDEO_FILE_NONVERBAL)
221          
222          if task_id:
223              task_info = wait_for_completion(task_id)
224              
225              if task_info and task_info["status"] == "completed":
226                  save_results(task_info, "nonverbal_analysis_results.json")
227      else:
228          print(f"⚠️ Fichier vidéo non trouvé: {VIDEO_FILE_NONVERBAL}")
229      
230      # Liste des tâches
231      print("\n📋 Liste des tâches d'analyse vidéo:")
232      list_video_tasks()
233  
234  if __name__ == "__main__":
235      main()