# app.py import os import logging import json from flask import Flask, jsonify, render_template, request from pydantic import BaseModel, Field from typing import List, Optional import psycopg2 from psycopg2.extras import RealDictCursor from google import genai from google.genai import types from utils import load_prompt from datetime import datetime # --- Configuration de l'application --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') app = Flask(__name__) app.secret_key = os.environ.get("FLASK_SECRET_KEY", "un-secret-par-defaut") # --- Configuration de la base de données et de l'API --- DATABASE_URL = os.environ.get("DATABASE") GOOGLE_API_KEY = os.environ.get("TOKEN") # --- Modèles de Données Pydantic (inchangés) --- class Argument(BaseModel): paragraphe_argumentatif: str = Field(description="Un unique paragraphe formant un argument complet. Il doit commencer par un connecteur logique (ex: 'Premièrement,'), suivi de son développement.") class Partie(BaseModel): chapeau: str = Field(description="La phrase d'introduction de la partie.") arguments: list[Argument] = Field(description="La liste des paragraphes argumentatifs qui suivent le chapeau.") transition: Optional[str] = Field(description="Phrase ou court paragraphe de transition.", default=None) class Dissertation(BaseModel): sujet: str = Field(description="Le sujet exact de la dissertation, tel que posé par l'utilisateur.") prof: str = Field(description="Le nom du professeur, qui est toujours 'Mariam AI'.", default="Mariam AI") introduction: str = Field(description="L'introduction complète de la dissertation.") parties: List[Partie] conclusion: str = Field(description="La conclusion complète de la dissertation.") # --- Configuration Gemini --- try: if not GOOGLE_API_KEY: logging.warning("La variable d'environnement TOKEN (GOOGLE_API_KEY) n'est pas définie.") client = None else: 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 de base de données --- def create_connection(): """Crée et retourne une connexion à la base de données PostgreSQL.""" if not DATABASE_URL: logging.error("La variable d'environnement DATABASE n'est pas configurée.") return None try: return psycopg2.connect(DATABASE_URL) except psycopg2.OperationalError as e: logging.error(f"Impossible de se connecter à la base de données : {e}") return None def init_database(): """Initialise la table de sauvegarde des dissertations si elle n'existe pas.""" conn = create_connection() if not conn: return False try: with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS dissertations ( id SERIAL PRIMARY KEY, user_ip VARCHAR(45), user_agent TEXT, question TEXT NOT NULL, dissertation_type VARCHAR(10) NOT NULL, course_id INTEGER, course_title VARCHAR(255), generated_content JSONB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, pdf_generated BOOLEAN DEFAULT FALSE ) """) conn.commit() return True except Exception as e: logging.error(f"Erreur lors de l'initialisation de la base de données : {e}") return False finally: if conn: conn.close() def save_dissertation(user_ip, user_agent, question, dissertation_type, course_id, course_title, content): """Sauvegarde une dissertation générée en base de données.""" conn = create_connection() if not conn: return None try: with conn.cursor() as cur: cur.execute(""" INSERT INTO dissertations (user_ip, user_agent, question, dissertation_type, course_id, course_title, generated_content) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id """, (user_ip, user_agent, question, dissertation_type, course_id, course_title, json.dumps(content))) conn.commit() return cur.fetchone()[0] except Exception as e: logging.error(f"Erreur lors de la sauvegarde de la dissertation : {e}") return None finally: if conn: conn.close() # --- Routes Principales --- @app.route('/') def philosophie(): return render_template("philosophie.html") @app.route('/gestion') def gestion(): """Page de gestion pour afficher toutes les dissertations.""" return render_template("gestion.html") # --- Routes API --- @app.route('/api/philosophy/courses', methods=['GET']) def get_philosophy_courses(): """Récupère la liste de tous les cours de philosophie pour le menu déroulant.""" conn = create_connection() if not conn: return jsonify({"error": "Connexion à la base de données échouée."}), 503 try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT id, title 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 lors de la récupération des cours."}), 500 finally: if conn: conn.close() @app.route('/api/dissertations', methods=['GET']) def get_dissertations(): """Récupère toutes les dissertations avec pagination.""" page = int(request.args.get('page', 1)) per_page = int(request.args.get('per_page', 10)) search = request.args.get('search', '').strip() conn = create_connection() if not conn: return jsonify({"error": "Connexion à la base de données échouée."}), 503 try: with conn.cursor(cursor_factory=RealDictCursor) as cur: # Construire la requête avec recherche optionnelle base_query = """ SELECT d.*, COUNT(*) OVER() as total_count FROM dissertations d """ where_clause = "" params = [] if search: where_clause = " WHERE d.question ILIKE %s OR d.course_title ILIKE %s" params = [f"%{search}%", f"%{search}%"] query = base_query + where_clause + """ ORDER BY d.created_at DESC LIMIT %s OFFSET %s """ params.extend([per_page, (page - 1) * per_page]) cur.execute(query, params) results = cur.fetchall() total_count = results[0]['total_count'] if results else 0 dissertations = [] for row in results: dissertations.append({ 'id': row['id'], 'user_ip': row['user_ip'], 'user_agent': row['user_agent'], 'question': row['question'], 'dissertation_type': row['dissertation_type'], 'course_id': row['course_id'], 'course_title': row['course_title'], 'created_at': row['created_at'].isoformat() if row['created_at'] else None, 'pdf_generated': row['pdf_generated'], 'content_preview': row['generated_content'].get('introduction', '')[:200] + '...' if row['generated_content'] else '' }) return jsonify({ 'dissertations': dissertations, 'total': total_count, 'page': page, 'per_page': per_page, 'total_pages': (total_count + per_page - 1) // per_page }) except Exception as e: logging.error(f"Erreur lors de la récupération des dissertations : {e}") return jsonify({"error": "Erreur interne du serveur."}), 500 finally: if conn: conn.close() @app.route('/api/dissertations/', methods=['GET']) def get_dissertation_detail(dissertation_id): """Récupère le détail complet d'une dissertation.""" conn = create_connection() if not conn: return jsonify({"error": "Connexion à la base de données échouée."}), 503 try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT * FROM dissertations WHERE id = %s """, (dissertation_id,)) result = cur.fetchone() if not result: return jsonify({"error": "Dissertation non trouvée."}), 404 return jsonify({ 'id': result['id'], 'user_ip': result['user_ip'], 'user_agent': result['user_agent'], 'question': result['question'], 'dissertation_type': result['dissertation_type'], 'course_id': result['course_id'], 'course_title': result['course_title'], 'created_at': result['created_at'].isoformat() if result['created_at'] else None, 'pdf_generated': result['pdf_generated'], 'generated_content': result['generated_content'] }) except Exception as e: logging.error(f"Erreur lors de la récupération de la dissertation {dissertation_id} : {e}") return jsonify({"error": "Erreur interne du serveur."}), 500 finally: if conn: conn.close() @app.route('/api/dissertations/', methods=['DELETE']) def delete_dissertation(dissertation_id): """Supprime une dissertation.""" conn = create_connection() if not conn: return jsonify({"error": "Connexion à la base de données échouée."}), 503 try: with conn.cursor() as cur: cur.execute("DELETE FROM dissertations WHERE id = %s", (dissertation_id,)) if cur.rowcount == 0: return jsonify({"error": "Dissertation non trouvée."}), 404 conn.commit() return jsonify({"message": "Dissertation supprimée avec succès."}) except Exception as e: logging.error(f"Erreur lors de la suppression de la dissertation {dissertation_id} : {e}") return jsonify({"error": "Erreur interne du serveur."}), 500 finally: if conn: conn.close() @app.route('/api/generate_dissertation', methods=['POST']) def generate_dissertation_api(): if not client: return jsonify({"error": "Le service IA n'est pas correctement configuré."}), 503 data = request.json sujet = data.get('question', '').strip() dissertation_type = data.get('type', 'type1').strip() course_id = data.get('courseId') if not sujet: return jsonify({"error": "Le champ 'question' est obligatoire."}), 400 if dissertation_type not in ['type1', 'type2']: return jsonify({"error": "Type de méthodologie invalide."}), 400 # Récupérer le contenu du cours si un ID est fourni context_str = "" course_title = None if course_id: conn = create_connection() if not conn: return jsonify({"error": "Connexion à la base de données échouée pour récupérer le contexte."}), 503 try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT title, content FROM cours_philosophie WHERE id = %s", (course_id,)) result = cur.fetchone() if result: course_title = result['title'] if result.get('content'): context_str = f"\n\n--- EXTRAIT DE COURS À UTILISER COMME CONTEXTE PRINCIPAL ---\n{result['content']}" except Exception as e: logging.error(f"Erreur lors de la récupération du contexte du cours {course_id}: {e}") finally: if conn: conn.close() try: prompt_filename = f"philo_dissertation_{dissertation_type}.txt" prompt_template = load_prompt(prompt_filename) if "Erreur:" in prompt_template: logging.error(f"Fichier de prompt non trouvé : {prompt_filename}") return jsonify({"error": "Configuration du prompt introuvable pour ce type."}), 500 final_prompt = prompt_template.format(phi_prompt=sujet, context=context_str) config = types.GenerateContentConfig( safety_settings=SAFETY_SETTINGS, response_mime_type="application/json", response_schema=Dissertation, ) response = client.models.generate_content( model="gemini-2.5-flash", contents=final_prompt, config=config ) if response.parsed: # Sauvegarder en base de données user_ip = request.environ.get('HTTP_X_FORWARDED_FOR', request.environ.get('REMOTE_ADDR', 'unknown')) user_agent = request.headers.get('User-Agent', 'unknown') dissertation_id = save_dissertation( user_ip=user_ip, user_agent=user_agent, question=sujet, dissertation_type=dissertation_type, course_id=course_id, course_title=course_title, content=response.parsed.dict() ) result = response.parsed.dict() result['dissertation_id'] = dissertation_id return jsonify(result) else: logging.error(f"Erreur de parsing de la réponse structurée. Réponse brute : {response.text}") return jsonify({"error": "Le modèle n'a pas pu générer une structure valide."}), 500 except Exception as e: logging.error(f"Erreur de génération Gemini : {e}") return jsonify({"error": f"Une erreur est survenue avec le service IA : {e}"}), 500 if __name__ == '__main__': init_database() # Initialiser la base de données au démarrage app.run(debug=True, port=5001)