# app.py import os import logging import json from datetime import datetime 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 # --- 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") # Dossier pour stocker les données de gestion DATA_DIR = "data" DISSERTATIONS_FILE = os.path.join(DATA_DIR, "dissertations_log.json") # Créer le dossier data s'il n'existe pas os.makedirs(DATA_DIR, exist_ok=True) # --- 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 (de l'exemple) --- 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 # --- Helpers pour la gestion des données --- def save_dissertation_data(input_data, output_data, success=True, error_message=None): """Sauvegarde les données d'entrée et de sortie dans un fichier JSON.""" try: # Lire les données existantes if os.path.exists(DISSERTATIONS_FILE): with open(DISSERTATIONS_FILE, 'r', encoding='utf-8') as f: data = json.load(f) else: data = [] # Préparer le nouvel enregistrement record = { "timestamp": datetime.now().isoformat(), "input": { "question": input_data.get('question', ''), "type": input_data.get('type', ''), "courseId": input_data.get('courseId') }, "output": output_data if success else None, "success": success, "error": error_message, "id": len(data) + 1 } # Ajouter le nouvel enregistrement data.append(record) # Sauvegarder with open(DISSERTATIONS_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception as e: logging.error(f"Erreur lors de la sauvegarde des données: {e}") def load_dissertations_data(): """Charge toutes les données des dissertations depuis le fichier JSON.""" try: if os.path.exists(DISSERTATIONS_FILE): with open(DISSERTATIONS_FILE, 'r', encoding='utf-8') as f: return json.load(f) return [] except Exception as e: logging.error(f"Erreur lors du chargement des données: {e}") return [] # --- Route Principale --- @app.route('/') def philosophie(): return render_template("philosophie.html") # --- Route de Gestion --- @app.route('/gestion') def gestion(): return render_template("gestion.html") # --- API pour récupérer les données de gestion --- @app.route('/api/gestion/dissertations', methods=['GET']) def get_dissertations_data(): """Récupère toutes les données des dissertations générées.""" try: data = load_dissertations_data() return jsonify({ "success": True, "data": data, "total": len(data) }) except Exception as e: logging.error(f"Erreur lors de la récupération des données de gestion: {e}") return jsonify({ "success": False, "error": "Erreur lors de la récupération des données" }), 500 # --- API pour supprimer une entrée --- @app.route('/api/gestion/dissertations/', methods=['DELETE']) def delete_dissertation_record(record_id): """Supprime un enregistrement spécifique.""" try: data = load_dissertations_data() # Trouver l'index de l'enregistrement à supprimer record_index = None for i, record in enumerate(data): if record.get('id') == record_id: record_index = i break if record_index is None: return jsonify({ "success": False, "error": "Enregistrement non trouvé" }), 404 # Supprimer l'enregistrement deleted_record = data.pop(record_index) # Sauvegarder with open(DISSERTATIONS_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return jsonify({ "success": True, "message": "Enregistrement supprimé avec succès" }) except Exception as e: logging.error(f"Erreur lors de la suppression: {e}") return jsonify({ "success": False, "error": "Erreur lors de la suppression" }), 500 # --- API pour vider toutes les données --- @app.route('/api/gestion/dissertations/clear', methods=['DELETE']) def clear_all_dissertations(): """Vide toutes les données des dissertations.""" try: # Créer un fichier vide with open(DISSERTATIONS_FILE, 'w', encoding='utf-8') as f: json.dump([], f) return jsonify({ "success": True, "message": "Toutes les données ont été supprimées" }) except Exception as e: logging.error(f"Erreur lors de la suppression générale: {e}") return jsonify({ "success": False, "error": "Erreur lors de la suppression" }), 500 # --- NOUVELLE Route API pour lister les cours --- @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() # --- Route API pour la génération de dissertation (MODIFIÉE avec logging) --- @app.route('/api/generate_dissertation', methods=['POST']) def generate_dissertation_api(): if not client: error_msg = "Le service IA n'est pas correctement configuré." save_dissertation_data(request.json or {}, None, False, error_msg) return jsonify({"error": error_msg}), 503 data = request.json sujet = data.get('question', '').strip() dissertation_type = data.get('type', 'type1').strip() course_id = data.get('courseId') # Nouvel ID de cours optionnel if not sujet: error_msg = "Le champ 'question' est obligatoire." save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 400 if dissertation_type not in ['type1', 'type2']: error_msg = "Type de méthodologie invalide." save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 400 # Récupérer le contenu du cours si un ID est fourni context_str = "" if course_id: conn = create_connection() if not conn: error_msg = "Connexion à la base de données échouée pour récupérer le contexte." save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 503 try: 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.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}") # On continue sans le contexte en cas d'erreur DB 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: error_msg = "Configuration du prompt introuvable pour ce type." logging.error(f"Fichier de prompt non trouvé : {prompt_filename}") save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 500 # Injecter le sujet ET le contexte dans le prompt 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-flash-latest", contents=final_prompt, config=config ) if response.parsed: result = response.parsed.dict() # Sauvegarder les données avec succès save_dissertation_data(data, result, True) return jsonify(result) else: error_msg = "Le modèle n'a pas pu générer une structure valide." logging.error(f"Erreur de parsing de la réponse structurée. Réponse brute : {response.text}") save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 500 except Exception as e: error_msg = f"Une erreur est survenue avec le service IA : {e}" logging.error(f"Erreur de génération Gemini : {e}") save_dissertation_data(data, None, False, error_msg) return jsonify({"error": error_msg}), 500 if __name__ == '__main__': app.run(debug=True, port=5001)