Explorons le code AiApi
Fonctionnalités
1. Envoi de fichiers (texte, vidéo, audio) pour transcription et post-traitement
Fonctionnement
- Upload de fichier : L’utilisateur peut envoyer un fichier via un formulaire HTML. Le fichier est ensuite traité pour déterminer son type (texte, vidéo, audio) et est ajouté à IPFS (InterPlanetary File System).
- Transcription : Si le fichier est une vidéo ou un fichier audio, il est transcrit en texte à l’aide du modèle Whisper.
- Post-traitement : Le texte obtenu peut être résumé ou traduit en utilisant le modèle Ollama Phi3.
Code impliqué
-
Formulaire HTML pour l’upload :
<form id="upload-form" enctype="multipart/form-data" method="post"> <input type="file" id="file" accept="video/*,audio/*,text/*,.txt" required> <input type="button" value="Upload" onclick="uploadFile()"> <div id="loading-indicator" class="spinner"></div> </form> -
Endpoint pour l’upload :
@app.post("/upload") async def create_upload_file(file: UploadFile = File(...)): # Validation de la taille du fichier max_file_size = 100 * 1024 * 1024 # 100MB if file.file.__sizeof__() > max_file_size: raise HTTPException(status_code=400, detail="File size exceeds the limit of 100MB") # Détection du type de fichier mime_type = get_mime_type(file) file_type = ( "text" if mime_type.startswith("text/") else "video" if mime_type.startswith("video/") or "MP4" in mime_type else "audio" if mime_type.startswith("Audio") or "MP3" in mime_type else "unknown" ) # Sauvegarde du fichier temporairement temp_file_path = f"/tmp/{file.filename}" with open(temp_file_path, "wb") as f: f.write(file.file.read()) # Calcul du hash du fichier original_file_hash = calculate_file_hash(temp_file_path) # Ajout du fichier à IPFS result = subprocess.run(["ipfs", "add", "-wq", temp_file_path], capture_output=True, text=True) if result.returncode != 0: raise HTTPException(status_code=500, detail="Error adding file to IPFS") cid = result.stdout.strip().split('\n')[-1] # Vérification de l'intégrité du fichier retrieved_file_path = f"/tmp/retrieved_{file.filename}" get_result = subprocess.run(["ipfs", "get", "-o", retrieved_file_path, cid], capture_output=True, text=True) if get_result.returncode != 0: raise HTTPException(status_code=500, detail="Error retrieving file from IPFS") retrieved_file_hash = calculate_file_hash(retrieved_file_path) if original_file_hash != retrieved_file_hash: raise HTTPException(status_code=500, detail="File integrity check failed after IPFS upload") # Suppression des fichiers temporaires os.remove(temp_file_path) os.remove(retrieved_file_path) # Réponse avec les informations du fichier output = {"cid": cid, "file": file.filename, "file_type": file_type, "mime_type": mime_type} return output -
Transcription et post-traitement :
@app.get("/g1vlog") async def ai_stt(cid: str, file: str): getlog = subprocess.run(["ipfs", "get", "-o", "vlog.mp4", f"{cid}/{file}"], capture_output=True, text=True) speech = model.transcribe("vlog.mp4", language="fr")['text'] subprocess.run(["rm", "-Rf", "vlog.mp4"]) output = {"speech": speech} return output @app.get("/tellme") async def ai_tellme(cid: str): result = subprocess.run(["ipfs", "cat", cid], capture_output=True, text=True) ipfsget = result.stdout.strip() file_size_kb = len(ipfsget.encode('utf-8')) / 1024 if file_size_kb > MAX_FILE_SIZE_KB: raise HTTPException(status_code=400, detail="File size exceeds the maximum allowed size.") curl_data['prompt'] = curl_data['prompt'].format(ipfsget) r = requests.post("http://localhost:11434/api/generate", json=curl_data) tellme = r.json()['response'] output = {"system" : curl_data['system'], "prompt" : curl_data['prompt'], "tellme" : tellme} return output
2. Fournir une URL YouTube pour téléchargement, extraction de texte, résumé et traduction
Fonctionnement
- Téléchargement de la vidéo : L’utilisateur fournit une URL YouTube. La vidéo est téléchargée à l’aide de
yt-dlp. - Transcription : La vidéo téléchargée est transcrite en texte à l’aide du modèle Whisper.
- Résumé et traduction : Le texte transcrit est résumé et traduit en utilisant le modèle Ollama Phi3.
Code impliqué
-
Formulaire HTML pour l’URL YouTube :
<form id="youtube-form" method="get"> <input type="text" id="youtube-url" placeholder="Enter YouTube URL" required> <input type="text" id="public-key" placeholder="Enter Public Key" required> <input type="button" id="youtube-button" value="Process" onclick="processYouTube()"> <div id="youtube-loading-indicator" class="spinner"></div> </form> -
Endpoint pour le traitement de l’URL YouTube :
@app.get("/youtube") async def ai_tube(url: str, pubkey: str): logs = [] logs.append(f"Received URL: {url}") logs.append(f"Received Public Key: {pubkey}") # Utilisation de yt-dlp pour obtenir la durée de la vidéo try: video_info = subprocess.check_output(["yt-dlp", "--get-duration", url], text=True) logs.append(f"Video duration info: {video_info}") except subprocess.CalledProcessError as e: logs.append(f"Error getting video duration: {e}") return {"error": "Failed to get video duration", "logs": logs} # Conversion de la durée en secondes try: duration_seconds = sum(int(x) * 60 ** i for i, x in enumerate(reversed(video_info.strip().split(":")))) logs.append(f"Video duration in seconds: {duration_seconds}") except ValueError as e: logs.append(f"Error converting video duration: {e}") return {"error": "Failed to convert video duration", "logs": logs} # Vérification de la durée de la vidéo if duration_seconds > 900: logs.append("Video is too long") return {"error": "Video is too long. Please provide a video shorter than 15 minutes.", "logs": logs} # Téléchargement de la vidéo try: logs.append("Downloading video...") subprocess.run(["yt-dlp", "-f", "233", "-o", "videodl.mp4", url], check=True) logs.append("Video downloaded successfully") except subprocess.CalledProcessError as e: logs.append(f"Error downloading video: {e}") return {"error": "Failed to download video", "logs": logs} # Transcription de la vidéo try: logs.append("Transcribing video...") speech = model.transcribe("videodl.mp4", language="fr")['text'] logs.append(f"Transcription result: {speech}") except Exception as e: logs.append(f"Error transcribing video: {e}") return {"error": "Failed to transcribe video", "logs": logs} # Suppression du fichier vidéo téléchargé try: logs.append("Removing downloaded video file...") subprocess.run(["rm", "videodl.mp4"], check=True) logs.append("Downloaded video file removed") except subprocess.CalledProcessError as e: logs.append(f"Error removing video file: {e}") return {"error": "Failed to remove video file", "logs": logs} # Résumé et traduction du texte curl_data['prompt'] = curl_data['prompt'].format(speech) logs.append(f"Prompt for summary: {curl_data['prompt']}") try: r = requests.post("http://localhost:11434/api/generate", json=curl_data) summary = r.json()['response'] logs.append(f"Summary result: {summary}") except requests.RequestException as e: logs.append(f"Error generating summary: {e}") return {"error": "Failed to generate summary", "logs": logs} curl_data['prompt'] = "Translate this text in French (or english if is is already in french): " + summary logs.append(f"Prompt for translation: {curl_data['prompt']}") try: r = requests.post("http://localhost:11434/api/generate", json=curl_data) translation = r.json()['response'] logs.append(f"Translation result: {translation}") except requests.RequestException as e: logs.append(f"Error generating translation: {e}") return {"error": "Failed to generate translation", "logs": logs} output = {"speech": speech, "summary": summary, "translation": translation, "logs": logs} logs.append(f"Output: {output}") return output
Ce microservice permet donc de gérer des fichiers multimédias et des URL YouTube pour en extraire des transcriptions, des résumés et des traductions, en utilisant des outils comme Whisper, yt-dlp, et le modèle Ollama Phi3.
Citations:
[1] https://github.com/papiche/AiApi
Une "IA maison" pour transcrire, traduire et résumer...
IA Libre