# --- START OF FILE app.py --- import os import logging from flask import Flask, make_response, render_template, request, redirect, url_for, session, jsonify, flash, Response, stream_with_context from datetime import datetime, timedelta import psycopg2 from psycopg2.extras import RealDictCursor from google import genai from google.genai import types from utils import load_prompt # Assurez-vous que utils.py et la fonction load_prompt existent # --- Configuration --- logging.basicConfig(level=logging.INFO) # Initialisation de Flask app = Flask(__name__) app.secret_key = os.environ.get("FLASK_SECRET_KEY", "uyyhhy77uu") app.permanent_session_lifetime = timedelta(days=200) # URLs et Clés API DATABASE_URL = os.environ.get("DATABASE") GOOGLE_API_KEY = os.environ.get("TOKEN") # Configuration du client Gemini try: client = genai.Client(api_key=GOOGLE_API_KEY) except Exception as e: logging.error(f"Erreur lors de l'initialisation du client GenAI: {e}") client = None 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"}, ] # --- Helpers --- def create_connection(): return psycopg2.connect(DATABASE_URL) # --- Routes Principales --- @app.route('/') def philosophie(): return render_template("philosophie.html") # --- Routes API (Non-Streaming) --- @app.route('/api/philosophy/courses', methods=['GET']) def get_philosophy_courses(): try: with create_connection() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT id, title, author, updated_at FROM cours_philosophie ORDER BY title") courses = cur.fetchall() return jsonify(courses) except Exception as e: logging.error(f"Erreur lors de la récupération des cours : {e}") return jsonify({"error": "Erreur interne du serveur."}), 500 @app.route('/api/philosophy/courses/', methods=['GET']) def get_philosophy_course(course_id): try: with create_connection() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT id, title, content, author, created_at, updated_at FROM cours_philosophie WHERE id = %s", (course_id,)) course = cur.fetchone() if course: return jsonify(course) return jsonify({"error": "Cours non trouvé"}), 404 except Exception as e: logging.error(f"Erreur lors de la récupération du cours : {e}") return jsonify({"error": "Erreur interne du serveur."}), 500 # --- Routes de Génération (Streaming) --- def process_and_stream(model_id, prompt_content): """Génère et stream le contenu depuis le modèle Gemini.""" if not client: yield "Erreur: Le client API n'est pas configuré." return try: stream = client.models.generate_content_stream( model=model_id, contents=prompt_content, config=types.GenerateContentConfig(safety_settings=SAFETY_SETTINGS) ) for chunk in stream: if chunk.text: yield chunk.text except Exception as e: logging.error(f"Erreur de streaming Gemini ({model_id}): {e}") yield f"\n\n**Erreur de génération :** {str(e)}" @app.route('/stream_philo', methods=['POST']) @app.route('/stream_philo_deepthink', methods=['POST']) def stream_philo_text(): data = request.json phi_prompt = data.get('question', '').strip() phi_type = data.get('type', '1') course_id = data.get('courseId') if not phi_prompt: return Response("Erreur: Veuillez saisir un sujet.", status=400, mimetype='text/plain') # Déterminer le modèle et le fichier de prompt is_deepthink = 'deepthink' in request.path model_id = "gemini-2.5-pro" if is_deepthink else "gemini-2.5-flash" prompt_files = {'1': 'philo_type1.txt', '2': 'philo_type2.txt'} prompt_file = prompt_files.get(phi_type) if not prompt_file: return Response(f"Erreur: Type de sujet '{phi_type}' non valide.", status=400, mimetype='text/plain') # Charger et formater le prompt prompt_template = load_prompt(prompt_file) final_prompt = prompt_template.format(phi_prompt=phi_prompt) # Ajouter le contenu du cours si disponible if course_id: try: with create_connection() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT content FROM cours_philosophie WHERE id = %s", (course_id,)) result = cur.fetchone() if result and result['content']: final_prompt += f"\n\n--- EXTRAIT DE COURS POUR CONTEXTE ---\n{result['content']}" except Exception as e: logging.error(f"Erreur DB pour le cours {course_id}: {e}") # Continuer sans le contenu du cours en cas d'erreur def generate(): yield from process_and_stream(model_id, final_prompt) return Response(stream_with_context(generate()), mimetype='text/plain') @app.route('/stream_philo_image', methods=['POST']) def stream_philo_image(): if 'image' not in request.files: return Response("Erreur: Fichier image manquant.", status=400, mimetype='text/plain') image_file = request.files['image'] if not image_file.filename: return Response("Erreur: Aucun fichier sélectionné.", status=400, mimetype='text/plain') try: img_bytes = image_file.read() image_part = types.Part.from_bytes(data=img_bytes, mime_type=image_file.mimetype) prompt_text = load_prompt('philo_image_analysis.txt') contents = [prompt_text, image_part] # Le modèle vision pro est souvent le meilleur pour les images model_id = "gemini-2.5-pro" def generate(): yield from process_and_stream(model_id, contents) return Response(stream_with_context(generate()), mimetype='text/plain') except Exception as e: logging.error(f"Erreur lors du traitement de l'image : {e}") return Response("Erreur interne lors du traitement de l'image.", status=500, mimetype='text/plain') # --- Routes d'Administration (inchangées) --- @app.route('/admin/philosophy/courses', methods=['GET', 'POST', 'DELETE']) def manage_philosophy_courses(): # ... (Le code de cette route reste le même qu'avant) if request.method == 'GET': try: conn = psycopg2.connect(DATABASE_URL) cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute("SELECT * FROM cours_philosophie") courses = cur.fetchall() cur.close() conn.close() return render_template('philosophy_courses.html', courses=courses) except Exception as e: flash(f'Erreur lors de la récupération des cours : {e}', 'danger') return redirect(url_for('manage_philosophy_courses')) elif request.method == 'POST': if 'title' in request.form: try: title = request.form.get('title') content = request.form.get('content') author = request.form.get('author') conn = psycopg2.connect(DATABASE_URL) cur = conn.cursor() cur.execute("INSERT INTO cours_philosophie (title, content, author) VALUES (%s, %s, %s)", (title, content, author)) conn.commit() cur.close() conn.close() flash('Cours ajouté avec succès !', 'success') return redirect(url_for('manage_philosophy_courses')) except Exception as e: flash(f'Erreur lors de l\'ajout du cours : {e}', 'danger') return redirect(url_for('manage_philosophy_courses')) else: try: course_id = request.form.get('id') conn = psycopg2.connect(DATABASE_URL) cur = conn.cursor() cur.execute("DELETE FROM cours_philosophie WHERE id = %s", (course_id,)) conn.commit() cur.close() conn.close() flash('Cours supprimé avec succès !', 'success') return redirect(url_for('manage_philosophy_courses')) except Exception as e: flash(f'Erreur lors de la suppression du cours : {e}', 'danger') return redirect(url_for('manage_philosophy_courses')) # --- END OF FILE app.py ---