Create app.py

app.py

@@ -0,0 +1,204 @@
+from flask import Flask, render_template, request, jsonify
+import google.generativeai as genai
+import os
+from PIL import Image
+import tempfile
+
+app = Flask(__name__)
+
+# Configuration de l'API Gemini
+token = os.environ.get("TOKEN")
+genai.configure(api_key=token)
+
+generation_config = {
+    "temperature": 1,
+    "max_output_tokens": 8192,
+}
+
+safety_settings = [
+    {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
+]
+
+
+@app.route('/')
+def generale():
+    return render_template("generale.html")
+
+
+
+
+def upload_and_process_file(file_path):
+    """Upload et traite un fichier avec l'API Gemini avec gestion des erreurs améliorée"""
+    max_retries = 3
+    retry_delay = 2  # secondes
+    
+    for attempt in range(max_retries):
+        try:
+            print(f"Tentative d'upload {attempt + 1}/{max_retries} pour {file_path}")
+            
+            # Vérification du fichier
+            if not os.path.exists(file_path):
+                raise FileNotFoundError(f"Le fichier {file_path} n'existe pas")
+                
+            file_size = os.path.getsize(file_path)
+            if file_size == 0:
+                raise ValueError(f"Le fichier {file_path} est vide")
+                
+            # Upload du fichier
+            uploaded_file = genai.upload_file(path=file_path)
+            print(f"Upload réussi: {uploaded_file.uri}")
+
+            # Attente du traitement
+            timeout = 300  # 5 minutes
+            start_time = time.time()
+            
+            while uploaded_file.state.name == "PROCESSING":
+                if time.time() - start_time > timeout:
+                    raise TimeoutError("Timeout pendant le traitement du fichier")
+                
+                print(f"En attente du traitement... Temps écoulé: {int(time.time() - start_time)}s")
+                time.sleep(10)
+                uploaded_file = genai.get_file(uploaded_file.name)
+
+            if uploaded_file.state.name == "FAILED":
+                raise ValueError(f"Échec du traitement: {uploaded_file.state.name}")
+                
+            print(f"Traitement terminé avec succès: {uploaded_file.uri}")
+            return uploaded_file
+
+        except ssl.SSLError as e:
+            print(f"Erreur SSL lors de l'upload (tentative {attempt + 1}): {e}")
+            if attempt < max_retries - 1:
+                time.sleep(retry_delay * (attempt + 1))
+            else:
+                raise
+                
+        except Exception as e:
+            print(f"Erreur lors de l'upload (tentative {attempt + 1}): {e}")
+            if attempt < max_retries - 1:
+                time.sleep(retry_delay * (attempt + 1))
+            else:
+                raise
+                
+                
+
+def is_youtube_url(url):
+    """Vérifie si l'URL est une URL YouTube"""
+    parsed = urlparse(url)
+    return any(domain in parsed.netloc for domain in ['youtube.com', 'youtu.be'])
+
+def download_youtube_video(url):
+    """Télécharge une vidéo YouTube avec gestion des erreurs améliorée"""
+    try:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            ydl_opts = {
+                'format': 'best[filesize<50M]',  # Limite la taille du fichier
+                'outtmpl': os.path.join(temp_dir, '%(title)s.%(ext)s'),
+                'quiet': True,
+                'no_warnings': True,
+                'extract_flat': False
+            }
+            
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                print(f"Téléchargement de la vidéo: {url}")
+                info = ydl.extract_info(url, download=True)
+                video_path = os.path.join(temp_dir, ydl.prepare_filename(info))
+                
+                if not os.path.exists(video_path):
+                    raise FileNotFoundError(f"La vidéo n'a pas été téléchargée correctement: {video_path}")
+                
+                # Copie vers un fichier temporaire permanent
+                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(video_path)[1])
+                with open(video_path, 'rb') as f:
+                    temp_file.write(f.read())
+                
+                print(f"Vidéo téléchargée avec succès: {temp_file.name}")
+                return temp_file.name
+                
+    except Exception as e:
+        print(f"Erreur lors du téléchargement de la vidéo: {e}")
+        return None
+
+def telecharger_pdf(url):
+    """Télécharge un PDF et retourne le chemin du fichier"""
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_file:
+            temp_file.write(response.content)
+            return temp_file.name
+    except Exception as e:
+        print(f"Erreur lors du téléchargement du PDF : {e}")
+        return None
+
+def allowed_file(filename):
+    """Vérifie si l'extension du fichier est autorisée"""
+    ALLOWED_EXTENSIONS = {'pdf', 'png', 'jpg', 'jpeg', 'gif', 'mp4', 'mov', 'avi'}
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+
+@app.route('/submit', methods=['POST'])
+def submit_question():
+    question = request.form.get('question')
+    urls = request.form.getlist('urls')
+    files = request.files.getlist('files')
+    
+    print("URLs reçues:", urls)
+    content = [question]
+    temp_files = []
+    
+    try:
+        # Traitement des fichiers uploadés
+        for file in files:
+            if file and allowed_file(file.filename):
+                with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.filename)[1]) as temp_file:
+                    file.save(temp_file.name)
+                    temp_files.append(temp_file.name)
+                    if file.filename.lower().endswith(('.png', '.jpg', '.jpeg', '.gif')):
+                        content.append(PIL.Image.open(temp_file.name))
+                    else:
+                        uploaded_file = upload_and_process_file(temp_file.name)
+                        content.append(uploaded_file)
+
+        # Traitement des URLs
+        for url in urls:
+            print(f"Traitement de l'URL : {url}")
+            if is_youtube_url(url):
+                video_path = download_youtube_video(url)
+                if video_path:
+                    temp_files.append(video_path)
+                    uploaded_file = upload_and_process_file(video_path)
+                    content.append(uploaded_file)
+            elif url.lower().endswith('.pdf'):
+                pdf_path = telecharger_pdf(url)
+                if pdf_path:
+                    temp_files.append(pdf_path)
+                    uploaded_file = upload_and_process_file(pdf_path)
+                    content.append(uploaded_file)
+
+       
+       
+
+        # Génération de contenu avec Gemini
+        model = genai.GenerativeModel(
+            model_name="models/gemini-1.5-flash-002",
+            safety_settings=safety_settings,
+            system_instruction="Tu es un assistant intelligent. ton but est d'assister au mieux que tu peux. tu as été créé par Aenir et tu t'appelles Mariam."
+        )
+        response = model.generate_content(content, request_options={"timeout": 600})
+        return jsonify({"response": response.text})
+
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+    finally:
+        # Nettoyage des fichiers temporaires
+        for temp_file in temp_files:
+            try:
+                os.unlink(temp_file)
+            except Exception as e:
+                print(f"Erreur lors de la suppression du fichier temporaire {temp_file}: {e}") 
+
+