import json import requests from rest_framework.response import Response from rest_framework import status from concurrent.futures import ThreadPoolExecutor from utils.stringifiers import markdownToQuilldelta, closeJSON from utils.parsing import parse_syntactic, parse_parts from utils.formate import format_data, create_prompts from utils.mistral_api import call_mistral_api import os import uuid from datetime import datetime def getCourseMetadata(sampleText, user): content, error = call_mistral_api( user=user, content=sampleText, agent_id="ag:65468023:20250304:extract-meta:9cc0452c", max_tokens=240, response_format={"type": "json_object"} ) if error: return False try: data = json.loads(closeJSON(content)) return data except Exception as e: return False # Ici le but premier est d'identifier l'école # Pour les sujets étudiés, on peut se baser sur les titres des knowledges et autres qui seront générés après # Pour une recherche, on pourrait corréler ça aux titres des cours def generateFlashcard1(notion, user): # Génération de la réponse answer_content, error = call_mistral_api( user=user, content=notion, agent_id="ag:65468023:20240925:untitled-agent:d9055718", max_tokens=1024 ) if error: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb801"}]}} # Génération de la question basée sur la réponse question_content, error = call_mistral_api( user=user, content=answer_content, agent_id="ag:65468023:20241209:questionner:acbd2459", max_tokens=1024 ) if error: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb802"}]}} try: return { "question": markdownToQuilldelta(question_content), "answer": markdownToQuilldelta(answer_content) } except Exception as e: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb803"}]}} def generateFlashcard(notion, user, title=False): # Génération de la réponse content, error = call_mistral_api( user=user, content=notion, agent_id="ag:65468023:20250407:flashcard-generator:03671b80", max_tokens=1024, response_format={"type": "json_object"} ) if error: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb804"}]}} closed_json = closeJSON(content) try: data = json.loads(closed_json) if 'question' in data and 'answer' in data: flashcard = { 'question': markdownToQuilldelta(data['question']), 'answer': markdownToQuilldelta(data['answer']) } if title and isinstance(flashcard['question'], list) and len(flashcard['question']) > 0 and 'insert' in flashcard['question'][0]: flashcard['title'] = flashcard['question'][0]['insert'] return flashcard else: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb805"}]}} except Exception as e: return {"question": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la flashcard."}]}, "answer": {"ops": [{"insert": "Erreur Hb806"}]}} def generateExercise(notion, user, title=False): # Génération de l'exercice exercise_content, error = call_mistral_api( user=user, content=notion, agent_id="ag:65468023:20241209:exercise-designer:6eb4e1e9", max_tokens=2048 ) if error: return {"visibleAnswer": {"ops": [{"insert": "Une erreur est survenue lors de la génération de l'exercice."}]}, "instructions": {"ops": [{"insert": "Erreur Hb811"}]}} # Génération de la correction correction_content, error = call_mistral_api( user=user, content=exercise_content, agent_id="ag:65468023:20250407:corrector:49f29136", max_tokens=2048 ) if error: return {"visibleAnswer": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la correction."}]}, "instructions": {"ops": [{"insert": "Erreur Hb812"}]}} try: exercise = { "visibleAnswer": markdownToQuilldelta(correction_content), "instructions": markdownToQuilldelta(exercise_content) } if title and isinstance(exercise['visibleAnswer'], list) and len(exercise['visibleAnswer']) > 0 and 'insert' in exercise['visibleAnswer'][0]: exercise['title'] = exercise['visibleAnswer'][0]['insert'] return exercise except Exception as e: return {"visibleAnswer": {"ops": [{"insert": "Une erreur est survenue lors de la génération de la correction."}]}, "instructions": {"ops": [{"insert": "Erreur Hb813"}]}} def generateText(content, user, title=False): text_content, error = call_mistral_api( user=user, content=content, agent_id="ag:65468023:20240925:untitled-agent:d9055718", max_tokens=255 ) if error: return {"ops": [{"insert": "Une erreur est survenue lors de la génération du texte : erreur Hb820"}]}, error try: knowledge = { 'content': markdownToQuilldelta(text_content) } if title and isinstance(knowledge['content'], list) and len(knowledge['content']) > 0 and 'insert' in knowledge['content'][0]: knowledge['title'] = knowledge['content']['ops'][0]['insert'] return knowledge except Exception as e: return {"ops": [{"insert": "Une erreur est survenue lors de la génération du texte : erreur Hb821"}]}, error def extractStructFromPage(text, user, images=[], language=None): content, error = call_mistral_api( user=user, content=text, agent_id="ag:65468023:20241212:luminas:6750ec78", max_tokens=1023, images=images, language=language ) if error: return {'error': 1, 'response': error} return content def createCourseArchitecture(notions, user, language=None): result, error = call_mistral_api( user=user, content=notions, agent_id="ag:65468023:20250308:re-manager2:9a443cfb", max_tokens=9000, language=language, response_format={"type": "json_object"} ) if error: return None, error try: result_dict = json.loads(result) title = result_dict.get('title', '') architecture = result_dict.get('course', []) except Exception as e: return None, {"error": "Erreur lors du parsing JSON"} return title, architecture def getCourseV2(documents, user, language=None): # Création du dossier s'il n'existe pas output_dir = 'course_generated' if not os.path.exists(output_dir): os.makedirs(output_dir) # Création d'un fichier unique pour cette génération timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") file_id = str(uuid.uuid4())[:8] output_file = os.path.join(output_dir, f'{timestamp}_{file_id}.json') # Structure pour stocker toutes les données generation_data = { 'timestamp': timestamp, 'user_id': str(user.publicIdentifier), 'blocks_input': [], 'extracted_structures': [], 'architecture': None, 'generated_content': [] } course = [] compt_img = 0 compt_knwl = 0 max_img = 50 max_knwl = 400 pageCount = 0 entry = "" unordered_course = { 'texts': [], 'flashcards': [], 'exercises': [], } def check_page_size(page): if len(page['text']) > 10000: return False for image in page['images']: if len(image) > 10**7: return False return True max_documents = 20 blocks = [] if len(documents) > max_documents: course.append({'ops': [{'insert': f"A cause d'un trop grand nombre de documents, le cours n'a pas pu être généré correctement. Merci de fournir moins de {max_documents} documents. Erreur Gc100", "attributes": {'italic': True}}]}) return None, course, None, -1 for document in documents: blocks += format_data(document) # Sauvegarde des blocks d'entrée generation_data['blocks_input'] = blocks extracted_structures = [] with ThreadPoolExecutor(max_workers=4) as executor: futures = [ executor.submit( extractStructFromPage, block['text'], user, images=block.get('images', []), language=language ) for block in blocks ] unordered_course = "" for future in futures: try: result = future.result() unordered_course += result + '\n' extracted_structures.append(result) except Exception as e: print(f"[getCourseV2] Erreur dans l'exécution parallèle: {str(e)}") # Sauvegarde des structures extraites generation_data['extracted_structures'] = extracted_structures title, architecture = createCourseArchitecture(unordered_course, user, language=language) # Sauvegarde de l'architecture generation_data['architecture'] = { 'title': title, 'architecture': architecture } if not title: course.append({'ops': [{'insert': "Une erreur interne est survenue. Erreur Gc200", "attributes": {'italic': True}}]}) return None, course, None, -2 knowledges = create_prompts(title, architecture) course.append({'type': 'topic', 'data': markdownToQuilldelta(title)}) def generate_knowledge(knowledge): if knowledge['type'] == 'text': text = generateText(knowledge['prompt'], user, title=True) return {'type': 'text', 'data': text} elif knowledge['type'] == 'flashcard': fls = generateFlashcard(knowledge['prompt'], user, title=True) return {'type': 'flashcard', 'data': fls} elif knowledge['type'] == 'exercise': exercise = generateExercise(knowledge['prompt'], user, title=True) return {'type': 'exercise', 'data': exercise} elif knowledge['type'] == 'part': return {'type': 'text', 'data': {'title': knowledge['prompt'], 'content': {'ops': [{"insert": knowledge['prompt'], "attributes": {'bold': True}}, {"insert": "\n", "attributes": {'header': 1}}]}}} generated_contents = [] with ThreadPoolExecutor(max_workers=4) as executor: futures = [executor.submit(generate_knowledge, k) for k in knowledges] for future in futures: try: result = future.result() course.append(result) generated_contents.append(result) except Exception as e: print(f"[getCourseV2] Erreur dans le traitement parallèle: {e}") # Sauvegarde des contenus générés generation_data['generated_content'] = generated_contents # Écriture de toutes les données dans le fichier try: with open(output_file, 'w', encoding='utf-8') as f: json.dump(generation_data, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[getCourseV2] Erreur lors de la sauvegarde des données: {str(e)}") metadata = getCourseMetadata(blocks[0]['text'], user) return title, course, metadata, 0 def create_write_in_document(content): return 0 output_dir = 'course_generated' if not os.path.exists(output_dir): os.makedirs(output_dir) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") file_id = str(uuid.uuid4())[:8] safe_title = timestamp + "_" + file_id + ".txt" file_path = os.path.join(output_dir, safe_title) with open(file_path, 'w', encoding='utf-8') as f: json.dump(content, f, ensure_ascii=False, indent=4) return file_path