Константин Лебединский
614 lines
22 KiB
Python
614 lines
22 KiB
Python
"""Маршруты тестов (E1.2).
|
|
|
|
Покрытие Express → Flask:
|
|
- GET /api/tests/ — каталог + hidden by you
|
|
- POST /api/tests/ — создать тест (цепочку с версией 1)
|
|
- GET /api/tests/<id>/summary — краткая карточка
|
|
- GET /api/tests/<id>/versions — список версий + hasAttempts
|
|
- GET /api/tests/<id>/editor — контент редактора
|
|
- POST /api/tests/<id>/draft — saveTestDraft (fork если нужно)
|
|
- POST /api/tests/<id>/versions/<vid>/activate
|
|
- PATCH /api/tests/<id> — chainActive
|
|
- POST /api/tests/<id>/ai/generate-test
|
|
- POST /api/tests/<id>/ai/generate-question
|
|
|
|
UI-страницы:
|
|
- GET /tests — каталог
|
|
- GET /tests/<id>/edit — редактор (вызывает /api/tests/...)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid as _uuid
|
|
|
|
from flask import Blueprint, jsonify, render_template, request
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from ..auth.decorators import current_user, login_required
|
|
from ..db import get_session
|
|
from ..messages import RU
|
|
from ..models import Test, TestVersion, User
|
|
from ..services.ai_editor import (
|
|
HttpError as AiHttpError,
|
|
check_test_quality,
|
|
generate_full_test_by_shape,
|
|
generate_or_rephrase_question,
|
|
generate_test_by_title,
|
|
improve_test_full,
|
|
parse_and_validate_shape,
|
|
)
|
|
from ..services.document_extract import (
|
|
HttpError as DocExtractHttpError,
|
|
extract_text_from_file,
|
|
)
|
|
from ..services.document_gen import generation_for_import_document
|
|
from ..services.draft_validator import LlmError
|
|
from ..services.editor_content import HttpError as EditorHttpError, get_editor_content
|
|
from ..services.test_attempt import (
|
|
HttpError as AttemptHttpError,
|
|
check_question_for_attempt,
|
|
count_missing_hints,
|
|
generate_missing_hints_for_test,
|
|
get_attempt_review_for_user,
|
|
get_play_content,
|
|
list_test_attempts_for_author,
|
|
start_attempt,
|
|
submit_attempt,
|
|
)
|
|
from ..services.test_access import (
|
|
is_test_author,
|
|
is_test_edit_open,
|
|
list_hidden_by_author,
|
|
list_visible_tests,
|
|
user_has_test_access,
|
|
)
|
|
from ..services.test_chain import has_any_attempt_for_test
|
|
from ..services.test_draft import (
|
|
HttpError as DraftHttpError,
|
|
create_test_with_version,
|
|
save_test_draft,
|
|
)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
tests_bp = Blueprint('tests', __name__)
|
|
|
|
|
|
# ─── helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
def _stringify_uuids(d: dict) -> dict:
|
|
out = {}
|
|
for k, v in d.items():
|
|
if hasattr(v, 'hex') and not isinstance(v, (str, bytes)):
|
|
out[k] = str(v)
|
|
else:
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def _to_uuid(val) -> _uuid.UUID | None:
|
|
if isinstance(val, _uuid.UUID):
|
|
return val
|
|
try:
|
|
return _uuid.UUID(str(val))
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
|
|
|
|
def _check_test_author_or_404(test_id: str, user_id: str) -> Test:
|
|
session = get_session()
|
|
tid = _to_uuid(test_id)
|
|
test = session.get(Test, tid) if tid else None
|
|
if not test:
|
|
from werkzeug.exceptions import NotFound
|
|
raise NotFound(RU['notFound'])
|
|
if not is_test_author(test.created_by, user_id) and not is_test_edit_open():
|
|
from werkzeug.exceptions import Forbidden
|
|
raise Forbidden('Доступ запрещён.')
|
|
return test
|
|
|
|
|
|
# ─── JSON API ────────────────────────────────────────────────────────────────
|
|
|
|
@tests_bp.route('/api/tests/', methods=['GET'])
|
|
@tests_bp.route('/api/tests', methods=['GET'])
|
|
@login_required
|
|
def api_list_tests():
|
|
user = current_user()
|
|
visible = list_visible_tests(user.id)
|
|
hidden = list_hidden_by_author(user.id)
|
|
return jsonify(
|
|
tests=[_stringify_uuids(r) for r in visible],
|
|
hiddenByYou=[_stringify_uuids(r) for r in hidden],
|
|
)
|
|
|
|
|
|
@tests_bp.route('/api/tests/', methods=['POST'])
|
|
@tests_bp.route('/api/tests', methods=['POST'])
|
|
@login_required
|
|
def api_create_test():
|
|
user = current_user()
|
|
body = request.get_json(silent=True) or {}
|
|
title = body.get('title')
|
|
if not isinstance(title, str) or not title.strip():
|
|
return jsonify(error='Укажите название.'), 400
|
|
out = create_test_with_version(user.id, title=title.strip(), description=body.get('description'))
|
|
return jsonify(out), 201
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/summary', methods=['GET'])
|
|
@login_required
|
|
def api_test_summary(test_id):
|
|
user = current_user()
|
|
session = get_session()
|
|
tid = _to_uuid(test_id)
|
|
if not tid:
|
|
return jsonify(error=RU['notFound']), 404
|
|
|
|
test = (
|
|
session.query(Test)
|
|
.options(selectinload(Test.author), selectinload(Test.versions))
|
|
.filter(Test.id == tid)
|
|
.first()
|
|
)
|
|
if not test:
|
|
return jsonify(error=RU['notFound']), 404
|
|
|
|
is_author = is_test_author(test.created_by, user.id)
|
|
open_edit = is_test_edit_open()
|
|
if not test.is_active and not is_author and not open_edit:
|
|
return jsonify(error=RU['notFound']), 404
|
|
|
|
if not is_author and not open_edit:
|
|
acc = user_has_test_access(user.id, test_id)
|
|
if not acc.ok:
|
|
return jsonify(error=RU['notFound']), 404
|
|
|
|
active_version = next((v for v in test.versions if v.is_active), None)
|
|
has_attempts = has_any_attempt_for_test(session, tid)
|
|
|
|
return jsonify(
|
|
test={
|
|
'id': str(test.id),
|
|
'title': test.title,
|
|
'description': test.description,
|
|
'passingThreshold': test.passing_threshold,
|
|
'chainActive': test.is_active,
|
|
'activeVersionId': str(active_version.id) if active_version else None,
|
|
'version': active_version.version if active_version else None,
|
|
'createdAt': test.created_at.isoformat() if test.created_at else None,
|
|
'updatedAt': test.updated_at.isoformat() if test.updated_at else None,
|
|
'createdBy': str(test.created_by) if test.created_by else None,
|
|
'authorFullName': test.author.full_name if test.author else None,
|
|
'hasAttempts': bool(has_attempts),
|
|
},
|
|
isAuthor=is_author,
|
|
hasActiveVersion=active_version is not None,
|
|
)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/versions', methods=['GET'])
|
|
@login_required
|
|
def api_test_versions(test_id):
|
|
user = current_user()
|
|
session = get_session()
|
|
tid = _to_uuid(test_id)
|
|
if not tid:
|
|
return jsonify(error=RU['notFound']), 404
|
|
|
|
test = (
|
|
session.query(Test)
|
|
.options(selectinload(Test.author), selectinload(Test.versions))
|
|
.filter(Test.id == tid)
|
|
.first()
|
|
)
|
|
if not test:
|
|
return jsonify(error=RU['notFound']), 404
|
|
if not is_test_author(test.created_by, user.id) and not is_test_edit_open():
|
|
return jsonify(error='Доступ запрещён.'), 403
|
|
|
|
has_attempts = has_any_attempt_for_test(session, tid)
|
|
sorted_versions = sorted(test.versions, key=lambda v: v.version)
|
|
|
|
return jsonify(
|
|
test={
|
|
'id': str(test.id),
|
|
'title': test.title,
|
|
'description': test.description,
|
|
'chainActive': test.is_active,
|
|
'createdAt': test.created_at.isoformat() if test.created_at else None,
|
|
'updatedAt': test.updated_at.isoformat() if test.updated_at else None,
|
|
'createdBy': str(test.created_by) if test.created_by else None,
|
|
'authorFullName': test.author.full_name if test.author else None,
|
|
},
|
|
versions=[
|
|
{
|
|
'id': str(v.id),
|
|
'version': v.version,
|
|
'is_active': v.is_active,
|
|
'parent_id': str(v.parent_id) if v.parent_id else None,
|
|
'created_at': v.created_at.isoformat() if v.created_at else None,
|
|
}
|
|
for v in sorted_versions
|
|
],
|
|
hasAttempts=has_attempts,
|
|
)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/editor', methods=['GET'])
|
|
@login_required
|
|
def api_test_editor(test_id):
|
|
user = current_user()
|
|
try:
|
|
out = get_editor_content(user.id, test_id)
|
|
except EditorHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/draft', methods=['POST'])
|
|
@login_required
|
|
def api_save_draft(test_id):
|
|
user = current_user()
|
|
payload = request.get_json(silent=True) or {}
|
|
try:
|
|
out = save_test_draft(user.id, test_id, payload)
|
|
except DraftHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/versions/<version_id>/activate', methods=['POST'])
|
|
@login_required
|
|
def api_activate_version(test_id, version_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
|
|
session = get_session()
|
|
tid = _to_uuid(test_id)
|
|
vid = _to_uuid(version_id)
|
|
|
|
version = session.query(TestVersion).filter(
|
|
TestVersion.test_id == tid, TestVersion.id == vid
|
|
).first()
|
|
if not version:
|
|
return jsonify(error='Версия не найдена.'), 404
|
|
|
|
session.query(TestVersion).filter(TestVersion.test_id == tid).update(
|
|
{TestVersion.is_active: False}, synchronize_session='fetch'
|
|
)
|
|
version.is_active = True
|
|
session.commit()
|
|
return jsonify(ok=True, activeVersionId=str(version_id))
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>', methods=['PATCH'])
|
|
@login_required
|
|
def api_patch_test(test_id):
|
|
user = current_user()
|
|
body = request.get_json(silent=True) or {}
|
|
chain = body.get('chainActive', body.get('isActive'))
|
|
if not isinstance(chain, bool):
|
|
return jsonify(error='Передайте chainActive: true/false в теле запроса.'), 400
|
|
test = _check_test_author_or_404(test_id, user.id)
|
|
session = get_session()
|
|
test.is_active = chain
|
|
session.commit()
|
|
return jsonify(id=test_id, chainActive=chain)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts/start', methods=['POST'])
|
|
@login_required
|
|
def api_start_attempt(test_id):
|
|
user = current_user()
|
|
try:
|
|
out = start_attempt(None, user.id, test_id)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out), 201
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts/<attempt_id>/play', methods=['GET'])
|
|
@login_required
|
|
def api_attempt_play(test_id, attempt_id):
|
|
user = current_user()
|
|
try:
|
|
out = get_play_content(None, user.id, test_id, attempt_id)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts/<attempt_id>/submit', methods=['POST'])
|
|
@login_required
|
|
def api_attempt_submit(test_id, attempt_id):
|
|
user = current_user()
|
|
body = request.get_json(silent=True) or {}
|
|
try:
|
|
out = submit_attempt(None, user.id, test_id, attempt_id, body.get('answers'))
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/hints/status', methods=['GET'])
|
|
@login_required
|
|
def api_test_hints_status(test_id):
|
|
out = count_missing_hints(None, test_id)
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/hints/generate', methods=['POST'])
|
|
@login_required
|
|
def api_test_hints_generate(test_id):
|
|
user = current_user()
|
|
try:
|
|
out = generate_missing_hints_for_test(None, user.id, test_id)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts/<attempt_id>/check', methods=['POST'])
|
|
@login_required
|
|
def api_attempt_check_question(test_id, attempt_id):
|
|
user = current_user()
|
|
body = request.get_json(silent=True) or {}
|
|
qid = body.get('questionId')
|
|
sel = body.get('selectedOptionIds') or []
|
|
if not qid:
|
|
return jsonify(error='questionId обязателен.'), 400
|
|
if not isinstance(sel, list):
|
|
return jsonify(error='selectedOptionIds должен быть массивом.'), 400
|
|
try:
|
|
out = check_question_for_attempt(None, user.id, test_id, attempt_id, qid, sel)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts/<attempt_id>/review', methods=['GET'])
|
|
@login_required
|
|
def api_attempt_review(test_id, attempt_id):
|
|
user = current_user()
|
|
try:
|
|
out = get_attempt_review_for_user(None, user.id, test_id, attempt_id)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(out)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/attempts', methods=['GET'])
|
|
@login_required
|
|
def api_attempts_list(test_id):
|
|
user = current_user()
|
|
try:
|
|
rows = list_test_attempts_for_author(None, user.id, test_id)
|
|
except AttemptHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
return jsonify(
|
|
attempts=[
|
|
{
|
|
'id': str(r['id']),
|
|
'userId': str(r['user_id']),
|
|
'status': r['status'],
|
|
'attemptNumber': r['attempt_number'],
|
|
'startedAt': r['started_at'].isoformat() if r['started_at'] else None,
|
|
'completedAt': r['completed_at'].isoformat() if r['completed_at'] else None,
|
|
'correctCount': r['correct_count'],
|
|
'totalQuestions': r['total_questions'],
|
|
'passed': r['passed'],
|
|
'testVersion': r['test_version'],
|
|
'attempterName': r['attempter_name'],
|
|
'attempterLogin': r['attempter_login'],
|
|
}
|
|
for r in rows
|
|
]
|
|
)
|
|
|
|
|
|
# ─── AI ──────────────────────────────────────────────────────────────────────
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/generate-test', methods=['POST'])
|
|
@login_required
|
|
def api_ai_generate_test(test_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
body = request.get_json(silent=True) or {}
|
|
try:
|
|
shape = parse_and_validate_shape(body.get('shape'))
|
|
draft = generate_full_test_by_shape(
|
|
body.get('testTitle') or '',
|
|
body.get('testDescription') or '',
|
|
shape,
|
|
)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
return jsonify(ok=True, draft=draft)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/generate-question', methods=['POST'])
|
|
@login_required
|
|
def api_ai_generate_question(test_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
body = request.get_json(silent=True) or {}
|
|
try:
|
|
out = generate_or_rephrase_question(
|
|
body.get('testTitle') or '',
|
|
body.get('testDescription') or '',
|
|
body.get('questionText') or '',
|
|
body.get('optionsCount'),
|
|
bool(body.get('hasMultipleAnswers')),
|
|
mode=body.get('mode'),
|
|
existing_options=body.get('existingOptions'),
|
|
)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
return jsonify(ok=True, **out)
|
|
|
|
|
|
def _ai_error_response(e):
|
|
if isinstance(e, AiHttpError):
|
|
return jsonify(error=e.message), e.status
|
|
if isinstance(e, LlmError):
|
|
log.warning('LLM error: %s (%s)', e, e.code)
|
|
return (
|
|
jsonify(error=str(e), code=e.code, settingsUrl='/settings'),
|
|
e.status or 502,
|
|
)
|
|
raise e
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/generate-by-title', methods=['POST'])
|
|
@login_required
|
|
def api_ai_generate_by_title(test_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
body = request.get_json(silent=True) or {}
|
|
title = (body.get('testTitle') or '').strip()
|
|
if not title:
|
|
return jsonify(error='Заполните название теста.'), 400
|
|
try:
|
|
draft = generate_test_by_title(
|
|
title,
|
|
body.get('testDescription') or '',
|
|
int(body.get('questionsCount') or 10),
|
|
int(body.get('optionsCount') or 4),
|
|
bool(body.get('hasMultipleAnswers')),
|
|
)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
return jsonify(ok=True, draft=draft)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/check', methods=['POST'])
|
|
@login_required
|
|
def api_ai_check_test(test_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
body = request.get_json(silent=True) or {}
|
|
try:
|
|
review = check_test_quality(
|
|
body.get('testTitle') or '',
|
|
body.get('testDescription') or '',
|
|
body.get('questions') or [],
|
|
)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
return jsonify(ok=True, review=review)
|
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/ai/improve', methods=['POST'])
|
|
@login_required
|
|
def api_ai_improve_test(test_id):
|
|
user = current_user()
|
|
_check_test_author_or_404(test_id, user.id)
|
|
body = request.get_json(silent=True) or {}
|
|
try:
|
|
out = improve_test_full(
|
|
body.get('testTitle') or '',
|
|
body.get('testDescription') or '',
|
|
body.get('questions') or [],
|
|
)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
return jsonify(ok=True, **out)
|
|
|
|
|
|
# ─── Импорт документа (E1.3) ─────────────────────────────────────────────────
|
|
|
|
@tests_bp.route('/api/tests/import/document', methods=['POST'])
|
|
@login_required
|
|
def api_import_document():
|
|
"""Шаг 1: загрузить файл и извлечь текст. Генерация — отдельным запросом."""
|
|
f = request.files.get('file')
|
|
if f is None or not f.filename:
|
|
return jsonify(error='Прикрепите файл к полю file.'), 400
|
|
try:
|
|
extracted = extract_text_from_file(f.mimetype, f, f.filename)
|
|
except DocExtractHttpError as e:
|
|
return jsonify(error=e.message), e.status
|
|
except Exception:
|
|
log.exception('extract_text_from_file failed')
|
|
return jsonify(error='Не удалось разобрать файл.'), 500
|
|
|
|
return jsonify(
|
|
received=True,
|
|
originalName=f.filename,
|
|
mime=f.mimetype,
|
|
size=len(extracted.encode('utf-8')),
|
|
extractedText=extracted,
|
|
textLength=len(extracted),
|
|
)
|
|
|
|
|
|
@tests_bp.route('/api/tests/generate-from-extracted', methods=['POST'])
|
|
@login_required
|
|
def api_generate_from_extracted():
|
|
"""Шаг 2: сгенерировать черновик теста из ранее извлечённого текста + подсказки автора."""
|
|
body = request.get_json(silent=True) or {}
|
|
extracted = (body.get('extractedText') or '').strip()
|
|
user_hint = (body.get('userHint') or '').strip()
|
|
shape_raw = body.get('shape')
|
|
if not extracted:
|
|
return jsonify(error='Нет текста для генерации.'), 400
|
|
shape = None
|
|
if shape_raw:
|
|
try:
|
|
shape = parse_and_validate_shape(shape_raw)
|
|
except (AiHttpError, LlmError) as e:
|
|
return _ai_error_response(e)
|
|
generation = generation_for_import_document(extracted, user_hint=user_hint, shape=shape)
|
|
return jsonify(generation=generation)
|
|
|
|
|
|
# ─── UI (Jinja) ───────────────────────────────────────────────────────────────
|
|
|
|
@tests_bp.route('/tests', methods=['GET'])
|
|
@login_required
|
|
def tests_list_page():
|
|
user = current_user()
|
|
visible = list_visible_tests(user.id)
|
|
hidden = list_hidden_by_author(user.id)
|
|
return render_template(
|
|
'tests/list.html',
|
|
visible=[_stringify_uuids(r) for r in visible],
|
|
hidden=[_stringify_uuids(r) for r in hidden],
|
|
)
|
|
|
|
|
|
@tests_bp.route('/tests/<test_id>/edit', methods=['GET'])
|
|
@login_required
|
|
def tests_editor_page(test_id):
|
|
user = current_user()
|
|
try:
|
|
content = get_editor_content(user.id, test_id)
|
|
except EditorHttpError as e:
|
|
if e.status == 404:
|
|
return render_template('404.html'), 404
|
|
if e.status == 403:
|
|
return ('Доступ запрещён.', 403)
|
|
return render_template('500.html'), 500
|
|
return render_template('tests/editor.html', content=content, test_id=test_id)
|
|
|
|
|
|
@tests_bp.route('/tests/<test_id>/attempt/<attempt_id>', methods=['GET'])
|
|
@login_required
|
|
def tests_attempt_page(test_id, attempt_id):
|
|
return render_template('tests/attempt.html', test_id=test_id, attempt_id=attempt_id)
|
|
|
|
|
|
@tests_bp.route('/tests/<test_id>/attempts/<attempt_id>/review', methods=['GET'])
|
|
@login_required
|
|
def tests_attempt_review_page(test_id, attempt_id):
|
|
user = current_user()
|
|
try:
|
|
review = get_attempt_review_for_user(None, user.id, test_id, attempt_id)
|
|
except AttemptHttpError as e:
|
|
if e.status == 404:
|
|
return render_template('404.html'), 404
|
|
return (e.message, e.status)
|
|
return render_template('tests/attempt_review.html', test_id=test_id, review=review)
|