import os import logging import json import tempfile from datetime import datetime from flask import Flask, jsonify, render_template, request, send_file 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 weasyprint import HTML import io import mimetypes # --- 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 --- 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 (ou le type de texte analysé).") 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 # --- 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: if os.path.exists(DISSERTATIONS_FILE): with open(DISSERTATIONS_FILE, 'r', encoding='utf-8') as f: data = json.load(f) else: data = [] question_logged = input_data.get('question', 'N/A (Image upload)') record = { "timestamp": datetime.now().isoformat(), "input": { "question": question_logged, "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 } data.append(record) 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 [] # --- Routes --- @app.route('/') def philosophie(): return render_template("philosophie.html") @app.route('/gestion') def gestion(): return render_template("gestion.html") # --- API de gestion --- @app.route('/api/gestion/dissertations', methods=['GET']) def get_dissertations_data(): 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 @app.route('/api/gestion/dissertations/', methods=['DELETE']) def delete_dissertation_record(record_id): try: data = load_dissertations_data() record_index = next((i for i, record in enumerate(data) if record.get('id') == record_id), None) if record_index is None: return jsonify({"success": False, "error": "Enregistrement non trouvé"}), 404 data.pop(record_index) 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 @app.route('/api/gestion/dissertations/clear', methods=['DELETE']) def clear_all_dissertations(): try: 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 # --- API pour lister les cours --- @app.route('/api/philosophy/courses', methods=['GET']) def get_philosophy_courses(): 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() # --- API pour la génération de dissertation (MISE À JOUR avec File Upload API) --- @app.route('/api/generate_dissertation', methods=['POST']) def generate_dissertation_api(): if not client: error_msg = "Le service IA n'est pas correctement configuré." return jsonify({"error": error_msg}), 503 # Détecter le type de requête et extraire les données is_file_upload = 'image' in request.files data_for_log = {} uploaded_file_ref = None # Pour stocker la référence du fichier uploadé if is_file_upload: dissertation_type = request.form.get('type', 'type3').strip() data_for_log = {'type': dissertation_type, 'courseId': None} if dissertation_type != 'type3': error_msg = "Le type 3 nécessite un fichier image." save_dissertation_data(data_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 400 else: data = request.json if not data: error_msg = "Requête JSON vide ou format incorrect." save_dissertation_data({'type': 'unknown'}, None, False, error_msg) return jsonify({"error": error_msg}), 400 dissertation_type = data.get('type', 'type1').strip() sujet = data.get('question', '').strip() course_id = data.get('courseId') data_for_log = {'type': dissertation_type, 'question': sujet, 'courseId': course_id} if dissertation_type not in ['type1', 'type2']: error_msg = "Type de méthodologie invalide." save_dissertation_data(data_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 400 if not sujet: error_msg = "Le champ 'question' est obligatoire." save_dissertation_data(data_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 400 try: prompt_filename = f"philo_dissertation_{dissertation_type}.txt" prompt_template = load_prompt(prompt_filename) if "Erreur:" in prompt_template: error_msg = f"Configuration du prompt introuvable pour le type '{dissertation_type}'." logging.error(f"Fichier de prompt non trouvé : {prompt_filename}") save_dissertation_data(data_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 500 # --- Préparation du contenu pour Gemini --- contents = [] if dissertation_type == 'type3': image_file = request.files['image'] # Mise à jour du log pour le commentaire de texte data_for_log['question'] = f"Image: {image_file.filename}" # UPLOAD DU FICHIER VIA L'API FILE UPLOAD DE GEMINI # Sauvegarder temporairement le fichier with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(image_file.filename)[1]) as temp_file: image_file.save(temp_file.name) temp_file_path = temp_file.name try: # Upload du fichier vers Gemini File API logging.info(f"Upload du fichier {image_file.filename} vers Gemini File API...") uploaded_file_ref = client.files.upload(file=temp_file_path) logging.info(f"Fichier uploadé avec succès. URI: {uploaded_file_ref.uri}") # Contenu: le prompt + la référence du fichier uploadé contents = [prompt_template, uploaded_file_ref] finally: # Supprimer le fichier temporaire try: os.unlink(temp_file_path) except Exception as e: logging.warning(f"Impossible de supprimer le fichier temporaire {temp_file_path}: {e}") # Utiliser le modèle multimodal model_name = "models/gemini-flash-latest" else: # Type 1 et 2 (texte uniquement) context_str = "" # Récupération du contexte du cours if data_for_log.get('courseId'): conn = create_connection() if conn: try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute("SELECT content FROM cours_philosophie WHERE id = %s", (data_for_log['courseId'],)) 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 {data_for_log['courseId']}: {e}") finally: conn.close() final_prompt = prompt_template.format(phi_prompt=data_for_log['question'], context=context_str) contents = [final_prompt] model_name = "models/gemini-flash-latest" # --- Appel à l'IA --- config = types.GenerateContentConfig( safety_settings=SAFETY_SETTINGS, response_mime_type="application/json", response_schema=Dissertation, ) response = client.models.generate_content( model=model_name, contents=contents, config=config ) # --- Traitement de la réponse --- if response.text: try: result = json.loads(response.text) if dissertation_type == 'type3' and 'sujet' in result: data_for_log['question'] = result['sujet'] save_dissertation_data(data_for_log, result, True) return jsonify(result) except json.JSONDecodeError: error_msg = "Le modèle n'a pas pu générer une structure JSON valide." logging.error(f"Erreur JSON Decode: Réponse brute : {response.text}") save_dissertation_data(data_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 500 else: error_msg = "Le modèle n'a retourné aucune réponse textuelle." save_dissertation_data(data_for_log, 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_for_log, None, False, error_msg) return jsonify({"error": error_msg}), 500 finally: # Nettoyer le fichier uploadé sur Gemini (optionnel mais recommandé) if uploaded_file_ref: try: client.files.delete(name=uploaded_file_ref.name) logging.info(f"Fichier {uploaded_file_ref.name} supprimé de Gemini File API") except Exception as e: logging.warning(f"Impossible de supprimer le fichier {uploaded_file_ref.name}: {e}") # --- ROUTE API POUR LA GÉNÉRATION DE PDF --- @app.route('/api/generate_pdf', methods=['POST']) def generate_pdf_api(): """Génère un PDF à partir des données JSON de la dissertation.""" dissertation_data = request.json if not dissertation_data: return jsonify({"error": "Aucune donnée de dissertation fournie."}), 400 try: html_string = render_template('dissertation_pdf.html', dissertation=dissertation_data) html = HTML(string=html_string) pdf_bytes = html.write_pdf() return send_file( io.BytesIO(pdf_bytes), mimetype='application/pdf', as_attachment=True, download_name='dissertation-philosophie.pdf' ) except Exception as e: logging.error(f"Erreur lors de la génération du PDF : {e}") return jsonify({"error": "Une erreur est survenue lors de la création du PDF."}), 500 if __name__ == '__main__': app.run(debug=True, port=5001)