import logging from rest_framework import viewsets, status, permissions from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.views import APIView from django.utils import timezone from django.http import HttpResponse from django.template.loader import render_to_string from .models import Formation, Lesson, UserProgress, UserFormationStatus, LessonChatMessage, LessonQuizAttempt from .serializers import FormationSerializer, FormationListSerializer, LessonSerializer, AdminFormationSerializer, AdminLessonSerializer from .ai_service import generate_quiz_questions, get_chat_response from .pdf_utils import extract_text_from_pdf from .certificate import build_certificate_pdf, get_certificate_context logger = logging.getLogger(__name__) def get_lesson_content(lesson): """ Retourne le texte du cours pour l'IA : content_text si rempli, sinon extraction du PDF (et sauvegarde dans content_text pour les prochaines fois). """ if lesson.content_text and lesson.content_text.strip(): return lesson.content_text.strip() if lesson.pdf_file: try: text = extract_text_from_pdf(lesson.pdf_file) if text and text.strip(): lesson.content_text = text.strip() lesson.save(update_fields=["content_text"]) logger.info("[get_lesson_content] Texte extrait du PDF pour leçon %s (%s signes)", lesson.id, len(text)) return lesson.content_text except Exception as e: logger.exception("[get_lesson_content] Erreur extraction PDF leçon %s: %s", lesson.id, e) fallback = f"Leçon : {lesson.title}. (Aucun contenu texte disponible pour le moment.)" return fallback class FormationViewSet(viewsets.ReadOnlyModelViewSet): """Affiche la liste des formations et le détail (avec verrou N+1).""" queryset = Formation.objects.all() serializer_class = FormationSerializer permission_classes = [permissions.IsAuthenticated] def get_serializer_class(self): if self.action == 'list': return FormationListSerializer return FormationSerializer def _can_access_formation(self, user, formation: Formation) -> bool: """ Règle d'or : accès à la formation N+1 seulement si N est validée. Basé sur le champ order. """ if formation.order <= 1: return True prev = Formation.objects.filter(order=formation.order - 1).first() if not prev: return True return UserFormationStatus.objects.filter( user=user, formation=prev, is_completed=True ).exists() def retrieve(self, request, *args, **kwargs): formation = self.get_object() if not self._can_access_formation(request.user, formation): return Response( {"error": "Vous devez d'abord valider la formation précédente pour accéder à celle-ci."}, status=status.HTTP_403_FORBIDDEN, ) return super().retrieve(request, *args, **kwargs) @action(detail=True, methods=['get']) def final_quiz(self, request, pk=None): """Quiz final : si formation déjà validée, retourne le quiz déjà passé ; sinon cooldown ou nouveau quiz.""" formation = self.get_object() status_obj = UserFormationStatus.objects.filter(user=request.user, formation=formation).first() # Déjà validé : retourner le dernier quiz enregistré, ou score seul si ancienne validation if status_obj and status_obj.is_completed: return Response({ "status": "already_passed", "questions": getattr(status_obj, 'last_questions', None) or [], "user_answers": getattr(status_obj, 'last_user_answers', None) or [], "score": status_obj.final_score, }) if status_obj and not status_obj.is_completed and status_obj.last_attempt_at: from datetime import timedelta if timezone.now() < status_obj.last_attempt_at + timedelta(hours=24): return Response({ "cooldown": True, "can_retry_after": (status_obj.last_attempt_at + timedelta(hours=24)).isoformat(), "message": "Réessayez dans 24h.", }) lessons = formation.lessons.all().order_by('order') full_context_parts = [] for lesson in lessons: full_context_parts.append(get_lesson_content(lesson)) full_context = "\n\n".join(full_context_parts) questions = generate_quiz_questions(full_context, num_questions=15) return Response(questions) @action(detail=True, methods=['post']) def validate_final_quiz(self, request, pk=None): """Valide le quiz final. Enregistre questions/réponses pour pouvoir les revoir.""" formation = self.get_object() try: score = int(request.data.get('score', 0)) except (TypeError, ValueError): score = 0 passed = request.data.get('passed') in (True, 'true', 1, '1') passed = passed and score >= 7 questions = request.data.get('questions', []) user_answers = request.data.get('user_answers', []) status_obj, _ = UserFormationStatus.objects.get_or_create(user=request.user, formation=formation) now = timezone.now() status_obj.final_score = score status_obj.last_attempt_at = now if questions: status_obj.last_questions = questions if user_answers is not None: status_obj.last_user_answers = user_answers if passed: status_obj.is_completed = True status_obj.completed_at = now status_obj.save() return Response({"passed": True, "score": score, "message": "Formation validée !"}) status_obj.is_completed = False status_obj.completed_at = None status_obj.save() from datetime import timedelta can_retry_after = now + timedelta(hours=24) return Response({ "passed": False, "score": score, "message": "Score insuffisant. Réessayez dans 24h.", "can_retry_after": can_retry_after.isoformat(), }) @action(detail=True, methods=['get'], url_path='certificate') def certificate(self, request, pk=None): """ Certificat de la formation : template HTML (joli) ou PDF (téléchargement). Accessible uniquement si l'utilisateur a validé la formation. ?download=1 → PDF en pièce jointe ; sinon → HTML (template certificate/index.html). """ formation = self.get_object() status_obj = UserFormationStatus.objects.filter( user=request.user, formation=formation, is_completed=True ).first() if not status_obj: return Response( {"error": "Vous devez valider cette formation pour obtenir le certificat."}, status=status.HTTP_403_FORBIDDEN, ) want_pdf = request.query_params.get("download") in ("1", "true") if want_pdf: try: pdf_bytes = build_certificate_pdf(request.user, formation, status_obj) except Exception as e: logger.exception("Certificate PDF generation failed: %s", e) return Response( {"error": "Impossible de générer le certificat PDF."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) filename = f"certificat_{formation.title.replace(' ', '_')}_{request.user.id}.pdf" response = HttpResponse(pdf_bytes, content_type="application/pdf") response["Content-Disposition"] = f'attachment; filename="{filename}"' return response # Rendu du template HTML (template certificate/index.html) try: context = get_certificate_context(request.user, formation, status_obj) html = render_to_string("certificate/index.html", context) return HttpResponse(html, content_type="text/html; charset=utf-8") except Exception as e: logger.exception("Certificate HTML render failed: %s", e) return Response( {"error": "Impossible d'afficher le certificat."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) class LessonViewSet(viewsets.ReadOnlyModelViewSet): serializer_class = LessonSerializer permission_classes = [permissions.IsAuthenticated] def get_queryset(self): # On peut filtrer par formation via ?formation_id=X formation_id = self.request.query_params.get('formation_id') if formation_id: return Lesson.objects.filter(formation_id=formation_id) return Lesson.objects.all() @action(detail=True, methods=['get']) def get_quiz(self, request, pk=None): """Retourne le dernier quiz si déjà joué (résultat ou cooldown), sinon génère un nouveau quiz.""" lesson = self.get_object() from datetime import timedelta if lesson.order > 1: prev_lesson = Lesson.objects.filter(formation=lesson.formation, order=lesson.order - 1).first() if prev_lesson and not UserProgress.objects.filter(user=request.user, lesson=prev_lesson, is_completed=True).exists(): return Response({"error": "Leçon précédente non terminée !"}, status=status.HTTP_403_FORBIDDEN) last = LessonQuizAttempt.objects.filter(user=request.user, lesson=lesson).order_by('-attempted_at').first() if last: if last.passed: return Response({ "status": "passed", "score": last.score, "questions": last.questions, "user_answers": last.user_answers, "message": "Leçon déjà validée.", }) retry_after = last.attempted_at + timedelta(hours=24) if timezone.now() < retry_after: return Response({ "status": "cooldown", "can_retry_after": retry_after.isoformat(), "message": "Réessayez dans 24h.", }) context = get_lesson_content(lesson) questions = generate_quiz_questions(context, num_questions=3) return Response(questions) @action(detail=True, methods=['post']) def validate_quiz(self, request, pk=None): """Enregistre la tentative (questions/réponses si fournis) et valide ou cooldown 24h.""" lesson = self.get_object() if 'score' not in request.data or 'passed' not in request.data: return Response( {"error": "Les champs 'score' et 'passed' sont requis."}, status=status.HTTP_400_BAD_REQUEST ) try: score = int(request.data.get('score', 0)) passed = request.data.get('passed', False) if isinstance(passed, str): passed = passed.lower() in ('true', '1', 'yes') except (ValueError, TypeError): return Response( {"error": "Le score doit être un nombre entier."}, status=status.HTTP_400_BAD_REQUEST ) questions = request.data.get('questions', []) user_answers = request.data.get('user_answers', []) LessonQuizAttempt.objects.create( user=request.user, lesson=lesson, questions=questions, user_answers=user_answers, score=score, passed=passed, ) if passed: progress, _ = UserProgress.objects.get_or_create(user=request.user, lesson=lesson) progress.is_completed = True progress.score_quiz = score progress.completed_at = timezone.now() progress.save() return Response({ "message": "Leçon validée ! La suivante est débloquée.", "passed": True, "score": score, }) from datetime import timedelta retry_after = timezone.now() + timedelta(hours=24) return Response({ "message": "Échec. Réessayez dans 24h.", "passed": False, "score": score, "can_retry_after": retry_after.isoformat(), }, status=status.HTTP_200_OK) @action(detail=True, methods=['post']) def mark_complete(self, request, pk=None): """ Marquer la leçon comme terminée sans passer le quiz (optionnel). Débloque la leçon suivante. Le quiz du cours reste optionnel ; seul le quiz final est obligatoire. """ lesson = self.get_object() progress, _ = UserProgress.objects.get_or_create(user=request.user, lesson=lesson) if not progress.is_completed: progress.is_completed = True progress.completed_at = timezone.now() progress.save(update_fields=['is_completed', 'completed_at']) return Response({ "message": "Leçon marquée comme terminée. Vous pouvez passer à la suivante.", "lesson_id": lesson.id, }, status=status.HTTP_200_OK) @action(detail=True, methods=['get']) def chat_history(self, request, pk=None): """Retourne l'historique des messages du chat pour cette leçon.""" lesson = self.get_object() messages = LessonChatMessage.objects.filter( user=request.user, lesson=lesson ).order_by('created_at')[:50] data = [ {"role": "user" if m.role == "USER" else "bot", "content": m.content} for m in messages ] return Response(data) @action(detail=True, methods=['post']) def ask_bot(self, request, pk=None): """ Endpoint pour le chatbot. Payload attendu : { "message": "Peux-tu résumer ce cours ?" } """ lesson = self.get_object() user_message = request.data.get('message', '') logger.info("[ask_bot] lesson_id=%s message=%s", lesson.id, (user_message or "")[:100]) if not user_message: return Response({"error": "Message vide"}, status=status.HTTP_400_BAD_REQUEST) context = get_lesson_content(lesson) logger.info("[ask_bot] context length=%s", len(context)) user_info = { "first_name": getattr(request.user, "first_name", "") or "", "last_name": getattr(request.user, "last_name", "") or "", "email": getattr(request.user, "email", "") or "", } # Construire l'historique à partir de la BDD previous_messages = LessonChatMessage.objects.filter(user=request.user, lesson=lesson).order_by('created_at')[:20] chat_history = [ { "role": "user" if msg.role == "USER" else "assistant", "content": msg.content, } for msg in previous_messages ] try: response_text = get_chat_response(context, user_message, chat_history=chat_history, user_info=user_info) logger.info("[ask_bot] réponse length=%s", len(response_text or "")) # Sauvegarde de l'historique (question + réponse) LessonChatMessage.objects.create( user=request.user, lesson=lesson, role="USER", content=user_message, ) LessonChatMessage.objects.create( user=request.user, lesson=lesson, role="BOT", content=response_text or "", ) return Response({ "response": response_text, "lesson_id": lesson.id }) except Exception as e: logger.exception("[ask_bot] Erreur: %s", e) return Response( {"error": str(e), "response": "Désolé, une erreur est survenue côté serveur."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) # --- VUES ADMIN POUR CRÉER/MODIFIER/SUPPRIMER --- class AdminFormationView(APIView): """Vue admin pour créer/modifier/supprimer des formations.""" permission_classes = [permissions.IsAdminUser] def get(self, request, pk=None): """Lister toutes les formations ou récupérer une formation (admin : avec image_url pour édition).""" if pk: try: formation = Formation.objects.prefetch_related('lessons').get(pk=pk) # Admin a besoin de image + image_url pour le formulaire d'édition serializer = AdminFormationSerializer(formation, context={'request': request}) return Response(serializer.data) except Formation.DoesNotExist: return Response({"error": "Formation introuvable."}, status=status.HTTP_404_NOT_FOUND) else: formations = Formation.objects.all().order_by('order', 'created_at') serializer = AdminFormationSerializer(formations, many=True, context={'request': request}) return Response(serializer.data) def post(self, request): """Créer une nouvelle formation.""" serializer = AdminFormationSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def put(self, request, pk=None): """Modifier une formation existante.""" try: formation = Formation.objects.get(pk=pk) except Formation.DoesNotExist: return Response({"error": "Formation introuvable."}, status=status.HTTP_404_NOT_FOUND) serializer = AdminFormationSerializer(formation, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def delete(self, request, pk=None): """Supprimer une formation.""" try: formation = Formation.objects.get(pk=pk) except Formation.DoesNotExist: return Response({"error": "Formation introuvable."}, status=status.HTTP_404_NOT_FOUND) formation.delete() return Response({"message": "Formation supprimée avec succès."}, status=status.HTTP_200_OK) class AdminLessonView(APIView): """Vue admin pour créer/modifier/supprimer des leçons.""" permission_classes = [permissions.IsAdminUser] def get(self, request): """Lister les leçons (filtrées par formation_id si fourni).""" formation_id = request.query_params.get('formation_id') if formation_id: lessons = Lesson.objects.filter(formation_id=formation_id).order_by('order') else: lessons = Lesson.objects.all().order_by('formation', 'order') serializer = AdminLessonSerializer(lessons, many=True) return Response(serializer.data) def post(self, request): """Créer une nouvelle leçon.""" # Vérifier que le PDF est fourni if 'pdf_file' not in request.FILES: return Response( {"error": "Le fichier PDF est obligatoire."}, status=status.HTTP_400_BAD_REQUEST ) serializer = AdminLessonSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def put(self, request, pk=None): """Modifier une leçon existante.""" try: lesson = Lesson.objects.get(pk=pk) except Lesson.DoesNotExist: return Response({"error": "Leçon introuvable."}, status=status.HTTP_404_NOT_FOUND) serializer = AdminLessonSerializer(lesson, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def delete(self, request, pk=None): """Supprimer une leçon.""" try: lesson = Lesson.objects.get(pk=pk) except Lesson.DoesNotExist: return Response({"error": "Leçon introuvable."}, status=status.HTTP_404_NOT_FOUND) lesson.delete() return Response({"message": "Leçon supprimée avec succès."}, status=status.HTTP_200_OK)