| |
|
|
| import os |
| import json |
| import mimetypes |
| from flask import Flask, request, session, jsonify, redirect, url_for, flash, render_template |
| from dotenv import load_dotenv |
| import google.generativeai as genai |
| import requests |
| from werkzeug.utils import secure_filename |
| import markdown |
| from flask_session import Session |
| import pprint |
|
|
| |
| load_dotenv() |
|
|
| app = Flask(__name__) |
|
|
| |
| |
| |
| app.config['SECRET_KEY'] = os.getenv('FLASK_SECRET_KEY', 'une-super-cle-secrete-a-changer') |
|
|
| |
| UPLOAD_FOLDER = 'temp' |
| ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg'} |
| app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER |
| app.config['MAX_CONTENT_LENGTH'] = 25 * 1024 * 1024 |
|
|
| |
| os.makedirs(UPLOAD_FOLDER, exist_ok=True) |
| print(f"Dossier d'upload configuré : {os.path.abspath(UPLOAD_FOLDER)}") |
|
|
| |
| app.config['SESSION_TYPE'] = 'filesystem' |
| app.config['SESSION_PERMANENT'] = False |
| app.config['SESSION_USE_SIGNER'] = True |
| app.config['SESSION_FILE_DIR'] = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'flask_session') |
|
|
| app.config['SESSION_COOKIE_SAMESITE'] = 'None' |
| |
| app.config['SESSION_COOKIE_SECURE'] = True |
|
|
| |
| os.makedirs(app.config['SESSION_FILE_DIR'], exist_ok=True) |
| print(f"Dossier pour les sessions serveur configuré : {app.config['SESSION_FILE_DIR']}") |
|
|
| |
| server_session = Session(app) |
|
|
| |
| MODEL_FLASH = 'gemini-2.0-flash' |
| MODEL_PRO = 'gemini-2.5-pro-exp-03-25' |
| SYSTEM_INSTRUCTION = "Tu es un assistant intelligent et amical nommé Mariam. Tu assistes les utilisateurs au mieux de tes capacités. Tu as été créé par Aenir." |
| 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"}, |
| ] |
| GEMINI_CONFIGURED = False |
| try: |
| gemini_api_key = os.getenv("GOOGLE_API_KEY") |
| if not gemini_api_key: |
| print("ERREUR: Clé API GOOGLE_API_KEY manquante dans le fichier .env") |
| else: |
| genai.configure(api_key=gemini_api_key) |
| models_list = [m.name for m in genai.list_models()] |
| if f'models/{MODEL_FLASH}' in models_list and f'models/{MODEL_PRO}' in models_list: |
| print(f"Configuration Gemini effectuée. Modèles requis ({MODEL_FLASH}, {MODEL_PRO}) disponibles.") |
| print(f"System instruction: {SYSTEM_INSTRUCTION}") |
| GEMINI_CONFIGURED = True |
| else: |
| print(f"ERREUR: Les modèles requis ({MODEL_FLASH}, {MODEL_PRO}) ne sont pas tous disponibles via l'API.") |
| print(f"Modèles trouvés: {models_list}") |
|
|
| except Exception as e: |
| print(f"ERREUR Critique lors de la configuration initiale de Gemini : {e}") |
| print("L'application fonctionnera sans les fonctionnalités IA.") |
|
|
| |
|
|
| def allowed_file(filename): |
| """Vérifie si l'extension du fichier est autorisée.""" |
| return '.' in filename and \ |
| filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS |
|
|
| def perform_web_search(query): |
| """Effectue une recherche web via l'API Serper.""" |
| serper_api_key = os.getenv("SERPER_API_KEY") |
| if not serper_api_key: |
| print("AVERTISSEMENT: Clé API SERPER_API_KEY manquante. Recherche web désactivée.") |
| return None |
| search_url = "https://google.serper.dev/search" |
| headers = {'X-API-KEY': serper_api_key, 'Content-Type': 'application/json'} |
| payload = json.dumps({"q": query, "gl": "fr", "hl": "fr"}) |
| try: |
| print(f"--- LOG WEBSEARCH: Recherche Serper pour: '{query}'") |
| response = requests.post(search_url, headers=headers, data=payload, timeout=10) |
| response.raise_for_status() |
| data = response.json() |
| print("--- LOG WEBSEARCH: Résultats de recherche Serper obtenus.") |
| return data |
| except requests.exceptions.RequestException as e: |
| print(f"--- LOG WEBSEARCH: Erreur lors de la recherche web : {e}") |
| return None |
| except json.JSONDecodeError as e: |
| print(f"--- LOG WEBSEARCH: Erreur JSON Serper : {e}") |
| return None |
|
|
| def format_search_results(data): |
| """Met en forme les résultats de recherche (format Markdown).""" |
| if not data: return "Aucun résultat de recherche web trouvé pertinent." |
| results = [] |
| if data.get('answerBox'): |
| ab = data['answerBox'] |
| title = ab.get('title', '') |
| snippet = ab.get('snippet') or ab.get('answer', '') |
| if snippet: results.append(f"**Réponse rapide : {title}**\n{snippet}\n") |
| if data.get('knowledgeGraph'): |
| kg = data['knowledgeGraph'] |
| title = kg.get('title', '') |
| type = kg.get('type', '') |
| description = kg.get('description', '') |
| if title and description: results.append(f"**{title} ({type})**\n{description}\n") |
| if kg.get('attributes'): |
| results.append("\n**Attributs :**") |
| for attr, value in kg['attributes'].items(): results.append(f"- {attr}: {value}") |
| results.append("") |
| if data.get('organic'): |
| results.append("**Pages web pertinentes :**") |
| for i, item in enumerate(data['organic'][:3], 1): |
| title = item.get('title', 'Sans titre') |
| link = item.get('link', '#') |
| snippet = item.get('snippet', 'Pas de description.') |
| results.append(f"{i}. **[{title}]({link})**\n {snippet}\n") |
| if data.get('peopleAlsoAsk'): |
| results.append("**Questions liées :**") |
| for i, item in enumerate(data['peopleAlsoAsk'][:2], 1): results.append(f"- {item.get('question', '')}") |
| if not results: return "Aucun résultat structuré trouvé dans la recherche web." |
| print(f"--- LOG WEBSEARCH: Résultats formatés (début): '{(' '.join(results))[:100]}...'") |
| return "\n".join(results) |
|
|
| def prepare_gemini_history(chat_history): |
| """Convertit l'historique stocké en session au format attendu par Gemini API.""" |
| print(f"--- DEBUG [prepare_gemini_history]: Entrée avec {len(chat_history)} messages") |
| gemini_history = [] |
| for i, message in enumerate(list(chat_history)): |
| role = 'user' if message.get('role') == 'user' else 'model' |
| text_part = message.get('raw_text') |
| |
| print(f" [prepare_gemini_history] Message {i} (rôle session: {message.get('role')}, rôle gemini: {role}): raw_text présent? {'Oui' if text_part is not None else 'NON'}, contenu début: '{str(text_part)[:60]}...'") |
|
|
| if text_part: |
| parts = [text_part] |
| gemini_history.append({'role': role, 'parts': parts}) |
| else: |
| |
| print(f" AVERTISSEMENT [prepare_gemini_history]: raw_text vide ou absent pour le message {i}, ignoré pour l'historique Gemini.") |
|
|
| print(f"--- DEBUG [prepare_gemini_history]: Sortie avec {len(gemini_history)} messages formatés pour Gemini") |
| return gemini_history |
|
|
| |
|
|
| @app.route('/') |
| def root(): |
| """Sert la page HTML principale.""" |
| print("--- LOG: Appel route '/' ---") |
| return render_template('index.html') |
|
|
| @app.route('/api/history', methods=['GET']) |
| def get_history(): |
| """Fournit l'historique de chat stocké en session au format JSON.""" |
| print("\n--- DEBUG [/api/history]: Début requête GET ---") |
| if 'chat_history' not in session: |
| session['chat_history'] = [] |
| print(" [/api/history]: Session 'chat_history' initialisée (vide).") |
|
|
| display_history = [] |
| current_history = session.get('chat_history', []) |
| print(f" [/api/history]: Historique récupéré de la session serveur: {len(current_history)} messages.") |
|
|
| |
| |
| |
|
|
| for i, msg in enumerate(current_history): |
| |
| if isinstance(msg, dict) and 'role' in msg and 'text' in msg: |
| display_history.append({ |
| 'role': msg.get('role'), |
| 'text': msg.get('text') |
| }) |
| else: |
| |
| print(f" AVERTISSEMENT [/api/history]: Format invalide dans l'historique session au message {i}: {msg}") |
|
|
| print(f" [/api/history]: Historique préparé pour le frontend: {len(display_history)} messages.") |
| return jsonify({'success': True, 'history': display_history}) |
|
|
| @app.route('/api/chat', methods=['POST']) |
| def chat_api(): |
| """Gère les nouvelles requêtes de chat via AJAX.""" |
| print(f"\n---===================================---") |
| print(f"--- DEBUG [/api/chat]: Nouvelle requête POST ---") |
|
|
| if not GEMINI_CONFIGURED: |
| print("--- ERREUR [/api/chat]: Tentative d'appel sans configuration Gemini valide.") |
| return jsonify({'success': False, 'error': "Le service IA n'est pas configuré correctement."}), 503 |
|
|
| |
| prompt = request.form.get('prompt', '').strip() |
| use_web_search = request.form.get('web_search', 'false').lower() == 'true' |
| file = request.files.get('file') |
| use_advanced = request.form.get('advanced_reasoning', 'false').lower() == 'true' |
|
|
| print(f" [/api/chat]: Prompt reçu: '{prompt[:50]}...'") |
| print(f" [/api/chat]: Recherche Web: {use_web_search}, Raisonnement Avancé: {use_advanced}") |
| print(f" [/api/chat]: Fichier: {file.filename if file else 'Aucun'}") |
|
|
| |
| if not prompt and not file: |
| print("--- ERREUR [/api/chat]: Prompt et fichier vides.") |
| return jsonify({'success': False, 'error': 'Veuillez fournir un message ou un fichier.'}), 400 |
|
|
| |
| if 'chat_history' not in session: |
| session['chat_history'] = [] |
| history_before_user_add = list(session.get('chat_history', [])) |
| print(f"--- DEBUG [/api/chat]: Historique en session AVANT ajout user message: {len(history_before_user_add)} messages") |
| |
| |
| |
| |
|
|
| uploaded_gemini_file = None |
| uploaded_filename = None |
| filepath_to_delete = None |
|
|
| |
| if file and file.filename != '': |
| print(f"--- LOG [/api/chat]: Traitement du fichier '{file.filename}'") |
| if allowed_file(file.filename): |
| try: |
| filename = secure_filename(file.filename) |
| filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) |
| file.save(filepath) |
| filepath_to_delete = filepath |
| uploaded_filename = filename |
| print(f" [/api/chat]: Fichier '{filename}' sauvegardé dans '{filepath}'") |
| mime_type = mimetypes.guess_type(filepath)[0] or 'application/octet-stream' |
| print(f" [/api/chat]: Upload Google AI (Mime: {mime_type})...") |
| uploaded_gemini_file = genai.upload_file(path=filepath, mime_type=mime_type) |
| print(f" [/api/chat]: Fichier Google AI '{uploaded_gemini_file.name}' uploadé.") |
| except Exception as e: |
| print(f"--- ERREUR [/api/chat]: Échec traitement/upload fichier '{filename}': {e}") |
| if filepath_to_delete and os.path.exists(filepath_to_delete): |
| try: os.remove(filepath_to_delete) |
| except OSError as del_e: print(f" Erreur suppression fichier temp après erreur: {del_e}") |
| return jsonify({'success': False, 'error': f"Erreur traitement fichier: {e}"}), 500 |
| else: |
| print(f"--- ERREUR [/api/chat]: Type de fichier non autorisé: {file.filename}") |
| return jsonify({'success': False, 'error': f"Type de fichier non autorisé."}), 400 |
|
|
| |
| raw_user_text = prompt |
| display_user_text = f"[{uploaded_filename}] {prompt}" if uploaded_filename and prompt else (prompt or f"[{uploaded_filename}]") |
| user_history_entry = { |
| 'role': 'user', |
| 'text': display_user_text, |
| 'raw_text': raw_user_text, |
| } |
|
|
| |
| if not isinstance(session.get('chat_history'), list): |
| print("--- ERREUR [/api/chat]: 'chat_history' n'est pas une liste! Réinitialisation.") |
| session['chat_history'] = [] |
| session['chat_history'].append(user_history_entry) |
|
|
| |
| history_after_user_add = list(session.get('chat_history', [])) |
| print(f"--- DEBUG [/api/chat]: Historique en session APRES ajout user message: {len(history_after_user_add)} messages") |
| |
| |
|
|
|
|
| |
| current_gemini_parts = [] |
| |
| if uploaded_gemini_file and not raw_user_text: |
| raw_user_text = f"Décris le contenu de ce fichier : {uploaded_filename}" |
| final_prompt_for_gemini = raw_user_text |
| current_gemini_parts.append(uploaded_gemini_file) |
| current_gemini_parts.append(final_prompt_for_gemini) |
| print(f" [/api/chat]: Fichier seul détecté, prompt généré: '{final_prompt_for_gemini}'") |
| elif uploaded_gemini_file and raw_user_text: |
| final_prompt_for_gemini = raw_user_text |
| current_gemini_parts.append(uploaded_gemini_file) |
| current_gemini_parts.append(final_prompt_for_gemini) |
| else: |
| final_prompt_for_gemini = raw_user_text |
| if final_prompt_for_gemini: |
| current_gemini_parts.append(final_prompt_for_gemini) |
|
|
| |
| if use_web_search and final_prompt_for_gemini: |
| print(f"--- LOG [/api/chat]: Activation recherche web pour: '{final_prompt_for_gemini[:60]}...'") |
| search_data = perform_web_search(final_prompt_for_gemini) |
| if search_data: |
| formatted_results = format_search_results(search_data) |
| enriched_prompt = f"""Voici la question originale de l'utilisateur:\n"{final_prompt_for_gemini}"\n\nInformations pertinentes de recherche web:\n--- DEBUT RESULTATS WEB ---\n{formatted_results}\n--- FIN RESULTATS WEB ---\n\nRéponds à la question originale en te basant sur ces informations ET ta connaissance générale.""" |
| |
| if current_gemini_parts and isinstance(current_gemini_parts[-1], str): |
| current_gemini_parts[-1] = enriched_prompt |
| final_prompt_for_gemini = enriched_prompt |
| print(f" [/api/chat]: Prompt enrichi avec recherche web.") |
| else: |
| print(f" [/api/chat]: Recherche web sans résultat pertinent ou erreur.") |
|
|
| |
| if not current_gemini_parts: |
| print("--- ERREUR [/api/chat]: Aucune donnée (texte ou fichier valide) à envoyer après traitement.") |
| if session.get('chat_history'): session['chat_history'].pop() |
| return jsonify({'success': False, 'error': "Impossible d'envoyer une requête vide."}), 400 |
|
|
| |
| try: |
| |
| |
| history_for_gemini_prep = list(session.get('chat_history', []))[:-1] |
| gemini_history_to_send = prepare_gemini_history(history_for_gemini_prep) |
|
|
| |
| contents_for_gemini = gemini_history_to_send + [{'role': 'user', 'parts': current_gemini_parts}] |
|
|
| |
| selected_model_name = MODEL_PRO if use_advanced else MODEL_FLASH |
| print(f"--- DEBUG [/api/chat]: Préparation de l'envoi à l'API Gemini (Modèle: {selected_model_name}) ---") |
| print(f" Nombre total de tours (historique + actuel): {len(contents_for_gemini)}") |
| print(f" Nombre de messages d'historique formatés envoyés: {len(gemini_history_to_send)}") |
| print(f" Contenu détaillé des 'parts' envoyées:") |
| for i, turn in enumerate(contents_for_gemini): |
| role = turn.get('role') |
| parts_details = [] |
| for part in turn.get('parts', []): |
| if isinstance(part, str): |
| parts_details.append(f"Text({len(part)} chars): '{part[:60].replace(chr(10), ' ')}...'") |
| elif hasattr(part, 'name') and hasattr(part, 'mime_type'): |
| parts_details.append(f"File(name={part.name}, mime={part.mime_type})") |
| else: |
| parts_details.append(f"UnknownPart({type(part)})") |
| print(f" Turn {i} (role: {role}): {', '.join(parts_details)}") |
| print("--------------------------------------------------------------------") |
|
|
| |
| active_model = genai.GenerativeModel( |
| model_name=selected_model_name, |
| safety_settings=SAFETY_SETTINGS, |
| system_instruction=SYSTEM_INSTRUCTION |
| ) |
| print(f"--- LOG [/api/chat]: Envoi de la requête à {selected_model_name}...") |
| response = active_model.generate_content(contents_for_gemini) |
|
|
| |
| response_text_raw = "" |
| response_html = "" |
| try: |
| if response.parts: |
| response_text_raw = response.text |
| print(f"--- LOG [/api/chat]: Réponse reçue de Gemini (brute, début): '{response_text_raw[:100]}...'") |
| else: |
| feedback_info = f"Feedback: {response.prompt_feedback}" if response.prompt_feedback else "Pas de feedback détaillé." |
| print(f"--- AVERTISSEMENT [/api/chat]: Réponse Gemini sans 'parts'. {feedback_info}") |
| if response.prompt_feedback and response.prompt_feedback.block_reason: |
| reason = response.prompt_feedback.block_reason.name |
| response_text_raw = f"Désolé, ma réponse a été bloquée ({reason})." |
| if reason == 'SAFETY' and response.prompt_feedback.safety_ratings: |
| blocked_cats = [r.category.name for r in response.prompt_feedback.safety_ratings if r.probability.name not in ['NEGLIGIBLE', 'LOW']] |
| if blocked_cats: response_text_raw += f" Catégories: {', '.join(blocked_cats)}." |
| else: |
| response_text_raw = "Désolé, je n'ai pas pu générer de réponse." |
| print(f" [/api/chat]: Message d'erreur généré: '{response_text_raw}'") |
|
|
| |
| response_html = markdown.markdown(response_text_raw, extensions=['fenced_code', 'tables', 'nl2br']) |
| if response_html != response_text_raw: |
| print(f" [/api/chat]: Réponse convertie en HTML.") |
|
|
| except ValueError as e: |
| print(f"--- ERREUR [/api/chat]: Impossible d'extraire le texte de la réponse Gemini. Erreur: {e}") |
| print(f" Réponse brute: {response}") |
| response_text_raw = "Désolé, erreur interne lors de la lecture de la réponse." |
| response_html = markdown.markdown(response_text_raw) |
| except Exception as e_resp: |
| print(f"--- ERREUR [/api/chat]: INATTENDUE lors traitement réponse Gemini : {e_resp}") |
| print(f" Réponse brute: {response}") |
| response_text_raw = f"Désolé, erreur inattendue ({type(e_resp).__name__})." |
| response_html = markdown.markdown(response_text_raw) |
|
|
| |
| assistant_history_entry = { |
| 'role': 'assistant', |
| 'text': response_html, |
| 'raw_text': response_text_raw |
| } |
| if not isinstance(session.get('chat_history'), list): |
| print("--- ERREUR [/api/chat]: 'chat_history' n'est pas liste avant ajout assistant! Réinitialisation.") |
| session['chat_history'] = [user_history_entry] |
| session['chat_history'].append(assistant_history_entry) |
|
|
| |
| history_final_turn = list(session.get('chat_history', [])) |
| print(f"--- DEBUG [/api/chat]: Historique en session FINAL après ajout assistant: {len(history_final_turn)} messages") |
| |
| |
|
|
| |
| print(f"--- LOG [/api/chat]: Envoi de la réponse HTML au client.") |
| print(f"---==================================---\n") |
| return jsonify({'success': True, 'message': response_html}) |
|
|
| except Exception as e: |
| print(f"--- ERREUR CRITIQUE [/api/chat]: Échec appel Gemini ou traitement réponse : {e}") |
| |
| current_history = session.get('chat_history') |
| if isinstance(current_history, list) and current_history: |
| try: |
| if current_history[-1].get('role') == 'user': |
| current_history.pop() |
| print(" [/api/chat]: Dernier message user retiré de l'historique suite à l'erreur.") |
| else: |
| print(" [/api/chat]: Dernier message n'était pas 'user', historique non modifié après erreur.") |
| except Exception as pop_e: |
| print(f" Erreur lors tentative retrait message user: {pop_e}") |
| print(f"---==================================---\n") |
| return jsonify({'success': False, 'error': f"Erreur interne: {e}"}), 500 |
|
|
| finally: |
| |
| |
| if uploaded_gemini_file: |
| try: |
| print(f"--- LOG [/api/chat FINALLY]: Tentative suppression fichier Google AI : {uploaded_gemini_file.name}") |
| genai.delete_file(uploaded_gemini_file.name) |
| print(f" [/api/chat FINALLY]: Fichier Google AI '{uploaded_gemini_file.name}' supprimé.") |
| except Exception as e_del_gcp: |
| print(f" AVERTISSEMENT [/api/chat FINALLY]: Échec suppression fichier Google AI '{uploaded_gemini_file.name}': {e_del_gcp}") |
| |
| if filepath_to_delete and os.path.exists(filepath_to_delete): |
| try: |
| os.remove(filepath_to_delete) |
| print(f"--- LOG [/api/chat FINALLY]: Fichier temporaire local '{filepath_to_delete}' supprimé.") |
| except OSError as e_del_local: |
| print(f"--- ERREUR [/api/chat FINALLY]: Échec suppression fichier local '{filepath_to_delete}': {e_del_local}") |
|
|
|
|
| @app.route('/clear', methods=['POST']) |
| def clear_chat(): |
| """Efface l'historique de chat dans la session.""" |
| print("\n--- DEBUG [/clear]: Requête POST reçue ---") |
| session.clear() |
| print(" [/clear]: Session serveur effacée.") |
| is_ajax = 'XMLHttpRequest' == request.headers.get('X-Requested-With') or \ |
| 'application/json' in request.headers.get('Accept', '') |
| if is_ajax: |
| print(" [/clear]: Réponse JSON (AJAX).") |
| return jsonify({'success': True, 'message': 'Historique effacé.'}) |
| else: |
| print(" [/clear]: Réponse Flash + Redirect (non-AJAX).") |
| flash("Conversation effacée.", "info") |
| return redirect(url_for('root')) |
|
|
|
|
| |
| if __name__ == '__main__': |
| print("--- Démarrage du serveur Flask ---") |
| port = int(os.environ.get('PORT', 5001)) |
| |
| |
| app.run(debug=True, host='0.0.0.0', port=port) |
|
|
| |