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.
465 lines
17 KiB
465 lines
17 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 |
|
|
|
from flask import Blueprint, jsonify, render_template, request |
|
from sqlalchemy import text |
|
|
|
from ..auth.decorators import current_user, login_required |
|
from ..db import get_engine |
|
from ..messages import RU |
|
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_access import is_test_author, list_hidden_by_author, list_visible_tests |
|
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: |
|
"""Преобразует UUID-поля в строки для безопасной JSON-сериализации.""" |
|
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 _check_test_author_or_404(test_id: str, user_id: str) -> dict: |
|
"""Загружает {id, created_by}; 404 если нет, 403 если не автор.""" |
|
eng = get_engine() |
|
with eng.connect() as conn: |
|
row = conn.execute( |
|
text('SELECT id, created_by FROM tests WHERE id = :id'), |
|
{'id': test_id}, |
|
).mappings().first() |
|
if not row: |
|
from werkzeug.exceptions import NotFound |
|
|
|
raise NotFound(RU['notFound']) |
|
if not is_test_author(row['created_by'], user_id): |
|
from werkzeug.exceptions import Forbidden |
|
|
|
raise Forbidden('Доступ запрещён.') |
|
return dict(row) |
|
|
|
|
|
# ─── 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() |
|
eng = get_engine() |
|
with eng.connect() as conn: |
|
row = conn.execute( |
|
text( |
|
""" |
|
SELECT t.id, t.title, t.description, t.passing_threshold, t.is_active AS chain_active, |
|
t.created_by, t.created_at, t.updated_at, |
|
tv.id AS active_version_id, tv.version, |
|
u.full_name AS author_full_name |
|
FROM tests t |
|
LEFT JOIN test_versions tv ON tv.test_id = t.id AND tv.is_active = true |
|
LEFT JOIN users u ON u.id = t.created_by |
|
WHERE t.id = :id |
|
""" |
|
), |
|
{'id': test_id}, |
|
).mappings().first() |
|
if not row: |
|
return jsonify(error=RU['notFound']), 404 |
|
|
|
is_author = is_test_author(row['created_by'], user.id) |
|
if not row['chain_active'] and not is_author: |
|
return jsonify(error=RU['notFound']), 404 |
|
|
|
if not is_author: |
|
from ..services.test_access import user_has_test_access |
|
|
|
acc = user_has_test_access(user.id, test_id) |
|
if not acc.ok: |
|
return jsonify(error=RU['notFound']), 404 |
|
|
|
return jsonify( |
|
test={ |
|
'id': str(row['id']), |
|
'title': row['title'], |
|
'description': row['description'], |
|
'passingThreshold': row['passing_threshold'], |
|
'chainActive': row['chain_active'], |
|
'activeVersionId': str(row['active_version_id']) if row['active_version_id'] else None, |
|
'version': row['version'], |
|
'createdAt': row['created_at'].isoformat() if row['created_at'] else None, |
|
'updatedAt': row['updated_at'].isoformat() if row['updated_at'] else None, |
|
'createdBy': str(row['created_by']) if row['created_by'] else None, |
|
'authorFullName': row['author_full_name'], |
|
}, |
|
isAuthor=is_author, |
|
hasActiveVersion=row['active_version_id'] is not None, |
|
) |
|
|
|
|
|
@tests_bp.route('/api/tests/<test_id>/versions', methods=['GET']) |
|
@login_required |
|
def api_test_versions(test_id): |
|
user = current_user() |
|
eng = get_engine() |
|
with eng.connect() as conn: |
|
t = conn.execute( |
|
text( |
|
""" |
|
SELECT t.id, t.title, t.created_by, t.is_active, t.created_at, t.updated_at, |
|
t.description, u.full_name AS author_full_name |
|
FROM tests t |
|
INNER JOIN users u ON u.id = t.created_by |
|
WHERE t.id = :id |
|
""" |
|
), |
|
{'id': test_id}, |
|
).mappings().first() |
|
if not t: |
|
return jsonify(error=RU['notFound']), 404 |
|
if not is_test_author(t['created_by'], user.id): |
|
return jsonify(error='Доступ запрещён.'), 403 |
|
|
|
rows = conn.execute( |
|
text( |
|
'SELECT id, version, is_active, parent_id, created_at ' |
|
'FROM test_versions WHERE test_id = :id ORDER BY version' |
|
), |
|
{'id': test_id}, |
|
).mappings().all() |
|
has_attempts = has_any_attempt_for_test(conn, test_id) |
|
|
|
return jsonify( |
|
test={ |
|
'id': str(t['id']), |
|
'title': t['title'], |
|
'description': t['description'], |
|
'chainActive': t['is_active'], |
|
'createdAt': t['created_at'].isoformat() if t['created_at'] else None, |
|
'updatedAt': t['updated_at'].isoformat() if t['updated_at'] else None, |
|
'createdBy': str(t['created_by']) if t['created_by'] else None, |
|
'authorFullName': t['author_full_name'], |
|
}, |
|
versions=[ |
|
{ |
|
'id': str(r['id']), |
|
'version': r['version'], |
|
'is_active': r['is_active'], |
|
'parent_id': str(r['parent_id']) if r['parent_id'] else None, |
|
'created_at': r['created_at'].isoformat() if r['created_at'] else None, |
|
} |
|
for r in rows |
|
], |
|
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) |
|
eng = get_engine() |
|
with eng.begin() as conn: |
|
v = conn.execute( |
|
text('SELECT id FROM test_versions WHERE test_id = :t AND id = :v'), |
|
{'t': test_id, 'v': version_id}, |
|
).first() |
|
if not v: |
|
return jsonify(error='Версия не найдена.'), 404 |
|
conn.execute( |
|
text('UPDATE test_versions SET is_active = false WHERE test_id = :t'), |
|
{'t': test_id}, |
|
) |
|
conn.execute( |
|
text('UPDATE test_versions SET is_active = true WHERE id = :v'), |
|
{'v': version_id}, |
|
) |
|
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 |
|
_check_test_author_or_404(test_id, user.id) |
|
eng = get_engine() |
|
with eng.begin() as conn: |
|
conn.execute( |
|
text( |
|
'UPDATE tests SET is_active = :v, updated_at = CURRENT_TIMESTAMP WHERE id = :id' |
|
), |
|
{'v': chain, 'id': test_id}, |
|
) |
|
return jsonify(id=test_id, chainActive=chain) |
|
|
|
|
|
# ─── 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')), |
|
) |
|
except (AiHttpError, LlmError) as e: |
|
return _ai_error_response(e) |
|
return jsonify(ok=True, **out) |
|
|
|
|
|
# ─── AI v2 (E1.8) ──────────────────────────────────────────────────── |
|
|
|
def _ai_error_response(e): |
|
"""Единый JSON-формат ошибки для AI-эндпоинтов.""" |
|
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(): |
|
"""PDF/DOCX/TXT/MD → извлечённый текст + AI-черновик (если задан LLM-ключ). |
|
|
|
Ограничения: размер файла — `MAX_CONTENT_LENGTH = 16 МБ` (см. фабрику). |
|
""" |
|
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 |
|
|
|
generation = generation_for_import_document(extracted) |
|
return jsonify( |
|
received=True, |
|
originalName=f.filename, |
|
mime=f.mimetype, |
|
size=len(extracted.encode('utf-8')), |
|
extractedText=extracted, |
|
textLength=len(extracted), |
|
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)
|
|
|