problems/views.py - main - tebo/BubblesCode

problems/views.py
408 строк15 KB
tebo
first_commit
03 июн 2026, 15:20
03 июн 2026, 15:20e3a5c0b
from django.shortcuts import render, get_object_or_404
from django.http import JsonResponse, HttpRequest, Http404
from django.contrib.auth.decorators import login_required
from django.views.generic import ListView
from django.db import models

import json
import subprocess
import sys
import ast
import os
import textwrap  


from .models import Problem, TestCase, Submission, UserCodeDraft
from theory.models import TheoryArticle
from .safe_runner import is_safe_code


PYTHON_INTERPRETER_PATH = sys.executable

class ProblemListView(ListView):
    """ 
    Displays a list of problems available to the current user. 
    Includes filtering by search, status, theory, and interactivity. 
    """
    model = Problem
    template_name = 'problems/problem_list.html'
    context_object_name = 'problems'
    ordering = ['id']

    def _base_queryset_for_user(self):
        user = self.request.user
        if user.is_staff:
            return Problem.objects.all()
        if not user.is_authenticated:
            return Problem.objects.filter(is_published=True, allowed_users=None)
        return Problem.objects.filter(is_published=True).filter(
            models.Q(allowed_users=None) | models.Q(allowed_users=user)
        ).distinct()

    def get_queryset(self):
        qs = self._base_queryset_for_user()
        user = self.request.user
        params = self.request.GET

        # Search
        query = params.get('q', '').strip()
        if query:
            qs = qs.filter(
                models.Q(title__icontains=query) |
                models.Q(description__icontains=query)
            )

        # Theory Filter
        theory_param = params.get('theory', '').strip()
        if theory_param:
            if theory_param == 'has':
                qs = qs.filter(theory_article__isnull=False)
            elif theory_param == 'none':
                qs = qs.filter(theory_article__isnull=True)
            elif theory_param != 'any':
                try:
                    theory_id = int(theory_param)
                    qs = qs.filter(theory_article_id=theory_id)
                except ValueError:
                    pass

        # Interactive Filter
        if params.get('interactive') in {'1', 'on', 'true', 'True'}:
            qs = qs.filter(use_interactive_runner=True)

        # Status Filter
        status_filter = params.get('status')
        if user.is_authenticated and status_filter in {'solved', 'attempted', 'unsolved'}:
            solved_ids = set(
                Submission.objects.filter(user=user, passed_all_tests=True)
                .values_list('problem_id', flat=True)
                .distinct()
            )
            attempted_ids = set(
                Submission.objects.filter(user=user)
                .values_list('problem_id', flat=True)
                .distinct()
            )
            if status_filter == 'solved':
                qs = qs.filter(id__in=solved_ids)
            elif status_filter == 'attempted':
                attempted_only = attempted_ids - solved_ids
                qs = qs.filter(id__in=attempted_only)
            elif status_filter == 'unsolved':
                qs = qs.exclude(id__in=solved_ids)

        # Sorting
        sort = params.get('sort', 'id_asc')
        ordering_map = {
            'id_asc': 'id',
            'id_desc': '-id',
            'title_asc': 'title',
            'title_desc': '-title',
        }
        qs = qs.order_by(ordering_map.get(sort, 'id'))

        return qs

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        params = self.request.GET
        user = self.request.user

        solved_ids = set()
        attempted_ids = set()
        if user.is_authenticated:
            solved_ids = set(
                Submission.objects.filter(user=user, passed_all_tests=True)
                .values_list('problem_id', flat=True)
            )
            attempted_ids = set(
                Submission.objects.filter(user=user)
                .values_list('problem_id', flat=True)
            )

        context.update({
            'filter_values': {
                'q': params.get('q', ''),
                'status': params.get('status', ''),
                'theory': params.get('theory', ''),
                'interactive': params.get('interactive') in {'1', 'on', 'true', 'True'},
                'sort': params.get('sort', 'id_asc'),
            },
            'solved_ids': solved_ids,
            'attempted_ids': attempted_ids,
            'theory_articles': TheoryArticle.objects.all().order_by('title'),
        })
        return context


def problem_detail(request: HttpRequest, pk: int):
    """ Displays the problem detail page, including editor and instructions. """
    problem = get_object_or_404(Problem, pk=pk)
    user = request.user

    is_staff = user.is_staff
    is_public_for_all = problem.is_published and not problem.allowed_users.exists()
    
    is_specifically_allowed = False
    if user.is_authenticated:
        is_specifically_allowed = problem.allowed_users.filter(pk=user.pk).exists()

    if not (is_staff or is_public_for_all or is_specifically_allowed):
        raise Http404("Задача не найдена или у вас нет прав доступа.")

    # загрузка черновика либо базового кода
    code_to_display = problem.initial_code
    if user.is_authenticated:
        try:
            draft = UserCodeDraft.objects.get(user=user, problem=problem)
            code_to_display = draft.code
        except UserCodeDraft.DoesNotExist:
            pass

    context = {
        'problem': problem,
        'can_access_theory_and_solution': user.is_authenticated,
        'code_to_display': code_to_display,
    }
    return render(request, 'problems/problem_detail.html', context)


@login_required
def execute_code(request: HttpRequest):
    """
    Запускает пользовательсикй код на тестовых случаях
    поддерживает функциональные решения и скриптовые тоже 
    """
    if request.method != 'POST':
        return JsonResponse({'error': 'Неверный метод запроса. Используйте POST.'}, status=405)

    try:
        data = json.loads(request.body)
        problem_id = data.get('problem_id')
        user_code = data.get('user_code')
        if not all([problem_id, user_code]):
            return JsonResponse({'error': 'Отсутствует problem_id или user_code.'}, status=400)

        # Проверка безопасности )(важно)
        is_safe, errors = is_safe_code(user_code)
        if not is_safe:
            return JsonResponse({'error': f"Ваш код не прошел проверку безопасности: {', '.join(errors)}"}, status=400)

        problem = get_object_or_404(Problem, pk=problem_id)
        test_cases = problem.test_cases.all()
    except (json.JSONDecodeError, Problem.DoesNotExist) as e:
        return JsonResponse({'error': f'Неверные данные запроса или задача не найдена: {e}'}, status=400)

    # Настройка окружения
    utf8_env = os.environ.copy()
    utf8_env['PYTHONUTF8'] = '1'

    # ловим ошибки
    overall_stdout = ""
    try:
        proc_for_stdout = subprocess.run(
            [PYTHON_INTERPRETER_PATH],
            input=user_code,
            capture_output=True, text=True, timeout=5, encoding='utf-8',
            env=utf8_env
        )
        overall_stdout = proc_for_stdout.stdout

        if proc_for_stdout.returncode != 0:
            return JsonResponse({
                'error': 'Ошибка выполнения или синтаксиса.',
                'overall_passed': False,
                'test_results': [],
                'stdout': overall_stdout,
                'stderr': proc_for_stdout.stderr
            })

    except subprocess.TimeoutExpired:
        return JsonResponse({'error': 'Превышен лимит времени при предварительном запуске.'}, status=400)

    # Определить режим: функция или скрипт
    results = []
    overall_passed = True

    # Если нет тест-кейсов, то просто помечаем задачу как решенную 
    if not test_cases.exists():
        Submission.objects.create(user=request.user, problem=problem, submitted_code=user_code, passed_all_tests=True)
        return JsonResponse({'overall_passed': True, 'test_results': [], 'stdout': overall_stdout})

    # Попытка найти определение функции
    function_name = None
    try:
        parsed_code = ast.parse(user_code)
        for node in ast.walk(parsed_code):
            if isinstance(node, ast.FunctionDef):
                function_name = node.name
                break
    except SyntaxError:
        return JsonResponse({'error': 'Синтаксическая ошибка.'}, status=400)

    # Запуск тест-кейсов
    for i, test in enumerate(test_cases):
        
        # Заготовки
        header = """# -*- coding: utf-8 -*-
import json, sys, io
from collections import deque
import math, random, datetime, functools, itertools, re, heapq

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next
        
class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right
"""

        if function_name:
            
            # Вызываем функцию напрямую
            runner_code = f"""{header}

{user_code}

try:
    result = {function_name}({test.input_data})
    print(json.dumps(result))
except Exception as e:
    print("Runtime Error: " + str(e), file=sys.stderr)
    sys.exit(1)
"""
        else:
            # Случай 2: Запуск скрипта
            # Исправляем отступы: пользовательский код должен быть сдвинут для блока try-except
            indented_code = textwrap.indent(user_code, '    ')
            
            # Обеспечиваем безопасный ввод (строковый тип + перенос строки для имитации Enter)
            safe_input_data = str(test.input_data)
            if not safe_input_data.endswith('\n'):
                safe_input_data += '\n'
                
            runner_code = f"""{header}
sys.stdin = io.StringIO({repr(safe_input_data)})

try:
{indented_code}
except Exception as e:
    print("Runtime Error: " + str(e), file=sys.stderr)
    sys.exit(1)
"""

        test_result = {
            'test_number': i + 1, 
            'input_data': test.input_data, 
            'expected_output': test.expected_output,
            'actual_output': None, 
            'passed': False, 
            'message': '', 
            'stderr': ''
        }

        try:
            proc = subprocess.run(
                [PYTHON_INTERPRETER_PATH],
                input=runner_code,
                capture_output=True, text=True, timeout=5, encoding='utf-8',
                env=utf8_env
            )
            test_result['stderr'] = proc.stderr
            
            if proc.returncode == 0:
                actual_output_str = proc.stdout.strip()
                
                # Логика сравнения
                try:
                    # 1. Парсим ожидаемый вывод
                    try:
                        expected = json.loads(test.expected_output)
                    except:
                        expected = ast.literal_eval(test.expected_output)

                    # 2.Парсим действительный вывод
                    try:
                        actual = json.loads(actual_output_str)
                    except json.JSONDecodeError:
                        try:
                            actual = ast.literal_eval(actual_output_str)
                        except:
                            actual = actual_output_str

                    test_result['actual_output'] = actual

                    # Сравнение
                    if actual == expected or str(actual).strip() == str(expected).strip():
                        test_result['passed'] = True
                        test_result['message'] = 'Тест пройден'
                    else:
                        test_result['passed'] = False
                        test_result['message'] = 'Неверный ответ'
                        overall_passed = False
                        
                except Exception as e:
                    if str(test.expected_output).strip() == actual_output_str:
                        test_result['passed'] = True
                        test_result['message'] = 'Тест пройден'
                        test_result['actual_output'] = actual_output_str
                    else:
                        test_result['passed'] = False
                        test_result['message'] = f'Ошибка проверки: {str(e)}'
                        test_result['actual_output'] = actual_output_str
                        overall_passed = False
            else:
                test_result['message'] = 'Ошибка выполнения'
                overall_passed = False
                
        except subprocess.TimeoutExpired:
            test_result['message'] = 'Превышен лимит времени'
            test_result['stderr'] = 'Timeout (5s)'
            overall_passed = False
            
        results.append(test_result)

    Submission.objects.create(
        user=request.user,
        problem=problem,
        submitted_code=user_code,
        passed_all_tests=overall_passed
    )

    return JsonResponse({
        'overall_passed': overall_passed,
        'test_results': results,
        'stdout': overall_stdout
    })


@login_required
def save_code_draft(request: HttpRequest):
    """ Сохранение черновика пользователя через AJAX-запрос """
    if request.method != 'POST':
        return JsonResponse({'status': 'error', 'message': 'Требуется POST-запрос'}, status=405)

    try:
        data = json.loads(request.body)
        problem_id = data.get('problem_id')
        user_code = data.get('code')

        if problem_id is None or user_code is None:
            return JsonResponse({'status': 'error', 'message': 'Отсутствуют problem_id или code'}, status=400)

        draft, created = UserCodeDraft.objects.update_or_create(
            user=request.user,
            problem_id=problem_id,
            defaults={'code': user_code}
        )

        return JsonResponse({'status': 'success', 'message': 'Черновик сохранен'})

    except json.JSONDecodeError:
        return JsonResponse({'status': 'error', 'message': 'Неверный формат JSON'}, status=400)
    except Exception as e:
        return JsonResponse({'status': 'error', 'message': str(e)}, status=500)