You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
613 lines
22 KiB
613 lines
22 KiB
"""Маршруты тестов (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)
|
|
|