Mireya Cueto Garrido
946f16a633
Introduce a full Vite/React UI for exams, auth, materials, images, generation, and export. Adapt backend for Sinbad2IA chat API, bcrypt passwords, CORS on port 5173, and schema migrations.
259 lines
11 KiB
Python
259 lines
11 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import ForbiddenError, NotFoundError
|
|
from app.core.security import clean_text
|
|
from app.models.exam import ExamTemplate, ExportFormat, ExportJob, ExportStatus, Question
|
|
from app.schemas.exam import (
|
|
ExamHistoryItem,
|
|
ExamTemplateCreate,
|
|
ExamTemplateRead,
|
|
ExportResponse,
|
|
ParsedQuestionsResponse,
|
|
ParseRequest,
|
|
PromptResponse,
|
|
QuestionCreate,
|
|
QuestionRead,
|
|
)
|
|
from app.services.image_service import ImageService
|
|
from app.services.llm import LLMClient
|
|
from app.services.material_service import MaterialService
|
|
from app.services.moodle_exporter import MoodleXMLExporter
|
|
from app.services.parser import AIQuestionParser
|
|
from app.services.prompt_builder import PromptBuilder
|
|
|
|
|
|
class ExamService:
|
|
def __init__(
|
|
self,
|
|
db: Session,
|
|
prompt_builder: PromptBuilder | None = None,
|
|
parser: AIQuestionParser | None = None,
|
|
exporter: MoodleXMLExporter | None = None,
|
|
material_service: MaterialService | None = None,
|
|
image_service: ImageService | None = None,
|
|
) -> None:
|
|
self.db = db
|
|
self.prompt_builder = prompt_builder or PromptBuilder()
|
|
self.parser = parser or AIQuestionParser()
|
|
self.exporter = exporter or MoodleXMLExporter()
|
|
self.material_service = material_service
|
|
self.image_service = image_service
|
|
|
|
def create_template(self, user_id: uuid.UUID, payload: ExamTemplateCreate) -> ExamTemplateRead:
|
|
template = ExamTemplate(
|
|
user_id=user_id,
|
|
title=clean_text(payload.title, max_length=200),
|
|
subject=clean_text(payload.subject, max_length=200),
|
|
educational_level=clean_text(payload.educational_level, max_length=120),
|
|
language=clean_text(payload.language, max_length=20),
|
|
settings=payload.settings.model_dump(mode="json"),
|
|
difficulty_profile=payload.difficulty_profile.model_dump(mode="json"),
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
self.db.refresh(template)
|
|
return self._template_read(template)
|
|
|
|
def list_templates(self, user_id: uuid.UUID) -> list[ExamTemplateRead]:
|
|
templates = self.db.scalars(
|
|
select(ExamTemplate)
|
|
.where(ExamTemplate.user_id == user_id)
|
|
.order_by(ExamTemplate.created_at.desc())
|
|
).all()
|
|
return [self._template_read(template) for template in templates]
|
|
|
|
def list_history(self, user_id: uuid.UUID) -> list[ExamHistoryItem]:
|
|
templates = self.db.scalars(
|
|
select(ExamTemplate)
|
|
.where(ExamTemplate.user_id == user_id)
|
|
.order_by(ExamTemplate.updated_at.desc())
|
|
).all()
|
|
history: list[ExamHistoryItem] = []
|
|
for template in templates:
|
|
export_jobs = sorted(template.export_jobs, key=lambda job: job.created_at, reverse=True)
|
|
history.append(
|
|
ExamHistoryItem(
|
|
id=template.id,
|
|
title=template.title,
|
|
subject=template.subject,
|
|
educational_level=template.educational_level,
|
|
language=template.language,
|
|
question_count=len(template.questions),
|
|
export_count=len(export_jobs),
|
|
last_export_at=export_jobs[0].created_at if export_jobs else None,
|
|
created_at=template.created_at,
|
|
updated_at=template.updated_at,
|
|
)
|
|
)
|
|
return history
|
|
|
|
def get_template(self, user_id: uuid.UUID, template_id: uuid.UUID) -> ExamTemplateRead:
|
|
return self._template_read(self._get_user_template_or_404(user_id, template_id))
|
|
|
|
def list_questions(self, user_id: uuid.UUID, template_id: uuid.UUID) -> list[QuestionRead]:
|
|
template = self._get_user_template_or_404(user_id, template_id)
|
|
questions = sorted(template.questions, key=lambda q: q.created_at)
|
|
return [self.to_question_read(question) for question in questions]
|
|
|
|
def get_owned_template(self, user_id: uuid.UUID, template_id: uuid.UUID) -> ExamTemplate:
|
|
return self._get_user_template_or_404(user_id, template_id)
|
|
|
|
def build_prompt(
|
|
self,
|
|
user_id: uuid.UUID,
|
|
template_id: uuid.UUID,
|
|
topic_prompt: str,
|
|
material_ids: list[uuid.UUID] | None = None,
|
|
) -> PromptResponse:
|
|
template = self._get_user_template_or_404(user_id, template_id)
|
|
reference_context = self._reference_context(template_id, material_ids)
|
|
images_catalog = self._images_catalog(template_id)
|
|
prompt = self.prompt_builder.build_prompt(
|
|
template,
|
|
topic_prompt,
|
|
reference_context,
|
|
images_catalog,
|
|
)
|
|
return PromptResponse(template_id=template.id, prompt=prompt)
|
|
|
|
async def generate_with_llm(
|
|
self,
|
|
user_id: uuid.UUID,
|
|
template_id: uuid.UUID,
|
|
topic_prompt: str,
|
|
llm_client: LLMClient,
|
|
material_ids: list[uuid.UUID] | None = None,
|
|
) -> ParsedQuestionsResponse:
|
|
template = self._get_user_template_or_404(user_id, template_id)
|
|
reference_context = self._reference_context(template_id, material_ids)
|
|
images_catalog = self._images_catalog(template_id)
|
|
prompt = self.prompt_builder.build_prompt(
|
|
template,
|
|
topic_prompt,
|
|
reference_context,
|
|
images_catalog,
|
|
)
|
|
raw_output = await llm_client.generate(prompt)
|
|
questions = self.parser.parse_json(raw_output)
|
|
return self._persist_questions(template.id, questions)
|
|
|
|
def parse_and_persist(self, user_id: uuid.UUID, payload: ParseRequest) -> ParsedQuestionsResponse:
|
|
self._get_user_template_or_404(user_id, payload.template_id)
|
|
questions = self.parser.parse(payload.raw_output, payload.input_format)
|
|
return self._persist_questions(payload.template_id, questions)
|
|
|
|
def export(self, user_id: uuid.UUID, template_id: uuid.UUID, export_format: ExportFormat) -> ExportResponse:
|
|
template = self._get_user_template_or_404(user_id, template_id)
|
|
questions = list(template.questions)
|
|
if not questions:
|
|
raise NotFoundError("Template does not contain questions to export")
|
|
|
|
image_map = self._image_map(template.id)
|
|
if export_format == ExportFormat.XML:
|
|
content = self.exporter.export_xml(questions, image_map)
|
|
elif export_format == ExportFormat.TXT:
|
|
content = self.exporter.export_txt(questions)
|
|
else:
|
|
content = self.exporter.export_json(questions)
|
|
|
|
self.db.add(
|
|
ExportJob(
|
|
template_id=template.id,
|
|
status=ExportStatus.COMPLETED,
|
|
format=export_format,
|
|
content=content,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
return ExportResponse(template_id=template.id, format=export_format, content=content)
|
|
|
|
def get_owned_question(self, user_id: uuid.UUID, question_id: uuid.UUID) -> tuple[Question, ExamTemplate]:
|
|
question = self.db.get(Question, question_id)
|
|
if question is None:
|
|
raise NotFoundError("Question not found")
|
|
template = self._get_user_template_or_404(user_id, question.template_id)
|
|
if question.template_id != template.id:
|
|
raise NotFoundError("Question not found")
|
|
return question, template
|
|
|
|
def to_question_read(self, question: Question) -> QuestionRead:
|
|
read = QuestionRead.model_validate(question)
|
|
if question.image_id:
|
|
return read.model_copy(update={"image_url": f"/exam/images/{question.image_id}/content"})
|
|
return read
|
|
|
|
def _persist_questions(self, template_id: uuid.UUID, questions: list[QuestionCreate]) -> ParsedQuestionsResponse:
|
|
persisted: list[Question] = []
|
|
for payload in questions:
|
|
image_id = payload.image_id
|
|
if image_id is not None:
|
|
if self.image_service is None:
|
|
raise NotFoundError("Image service is not available")
|
|
self.image_service.get_image_for_template(template_id, image_id)
|
|
|
|
question = Question(
|
|
template_id=template_id,
|
|
question_type=payload.question_type,
|
|
statement=clean_text(payload.statement),
|
|
correct_answers=[clean_text(answer, max_length=1_000) for answer in payload.correct_answers],
|
|
wrong_answers=[clean_text(answer, max_length=1_000) for answer in payload.wrong_answers],
|
|
matching_pairs=[pair.model_dump() for pair in payload.matching_pairs],
|
|
image_id=image_id,
|
|
difficulty=payload.difficulty,
|
|
score=payload.score,
|
|
penalty=payload.penalty,
|
|
options=payload.options,
|
|
)
|
|
self.db.add(question)
|
|
persisted.append(question)
|
|
|
|
self.db.commit()
|
|
for question in persisted:
|
|
self.db.refresh(question)
|
|
|
|
return ParsedQuestionsResponse(questions=[self.to_question_read(question) for question in persisted])
|
|
|
|
def _reference_context(
|
|
self,
|
|
template_id: uuid.UUID,
|
|
material_ids: list[uuid.UUID] | None,
|
|
) -> str:
|
|
if self.material_service is None:
|
|
return ""
|
|
return self.material_service.build_reference_context(template_id, material_ids)
|
|
|
|
def _images_catalog(self, template_id: uuid.UUID) -> str:
|
|
if self.image_service is None:
|
|
return ""
|
|
return self.image_service.images_catalog(template_id)
|
|
|
|
def _image_map(self, template_id: uuid.UUID) -> dict[uuid.UUID, object]:
|
|
if self.image_service is None:
|
|
return {}
|
|
return self.image_service.build_image_map(template_id)
|
|
|
|
def _get_user_template_or_404(self, user_id: uuid.UUID, template_id: uuid.UUID) -> ExamTemplate:
|
|
template = self.db.get(ExamTemplate, template_id)
|
|
if template is None:
|
|
raise NotFoundError("Exam template not found")
|
|
if template.user_id != user_id:
|
|
raise ForbiddenError("You do not have access to this exam template")
|
|
return template
|
|
|
|
def _template_read(self, template: ExamTemplate) -> ExamTemplateRead:
|
|
return ExamTemplateRead(
|
|
id=template.id,
|
|
title=template.title,
|
|
subject=template.subject,
|
|
educational_level=template.educational_level,
|
|
language=template.language,
|
|
settings=template.settings,
|
|
difficulty_profile=template.difficulty_profile,
|
|
created_at=template.created_at,
|
|
updated_at=template.updated_at,
|
|
question_count=len(template.questions),
|
|
)
|