import time import base64 import hashlib import logging from io import BytesIO from random import shuffle from datetime import timedelta from django.utils.translation import gettext as _ from django.utils import timezone from django.core.files.storage import default_storage from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from university.api.base import Endpoint from university.api.bases import MyOnlineExamEndpoint from university.models import ( UserOnlineExam, OnlineExamEvent, UserOnlineExamFace, OnlineExamRegistrate, Upload, OnlineExam ) from university.file_storage.utils import generate_random_file_path from university.forms import ( FaceIdentificationForm, ExamineeCaptureForm, ExamineeInfoForm, ExamEnrollUpdateForm ) from university.utils.requests import ( get_client_ip, get_client_agent, check_ip_in_allowlist ) from university.tasks.online_exam import async_check_examinee_face logger = logging.getLogger(__name__) class MyExamListEndpoint(Endpoint): permission_classes = (IsAuthenticated, ) def get(self, request, *args, **kwargs): my_exams = [{ 'name': u_exam.online_exam.name, 'slug': u_exam.online_exam.slug, 'short_description': u_exam.online_exam.short_description, 'rules': u_exam.online_exam.rules, 'start_at': timezone.localtime(u_exam.online_exam.start_at).strftime('%Y-%m-%d %H:%M:%S'), 'end_at': timezone.localtime(u_exam.online_exam.end_at).strftime('%Y-%m-%d %H:%M:%S'), 'status': u_exam.status, 'status_display': u_exam.get_status_display(), 'number_of_questions': u_exam.online_exam.number_of_questions, 'total_score': u_exam.online_exam.total_score, 'passing_score': u_exam.online_exam.passing_score, 'score': u_exam.score, 'passed': u_exam.passed, 'cover_url': u_exam.online_exam.cover_image.url, 'cert_url': u_exam.cert.private_url if u_exam.cert else None, } for u_exam in request.user.useronlineexam_set.exclude( status=UserOnlineExam.STATUS_REGISTE_WAIT_CONFIRM).order_by('-created_at')] return Response({ 'detail': _('success'), 'data': my_exams }) class MyExamDetailEndpoint(MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def get(self, request, my_onlineexam, *args, **kwargs): return Response({ 'detail': _('success'), 'data': { 'name': my_onlineexam.online_exam.name, 'slug': my_onlineexam.online_exam.slug, 'short_description': my_onlineexam.online_exam.short_description, 'rules': my_onlineexam.online_exam.rules, 'start_at': timezone.localtime(my_onlineexam.online_exam.start_at).strftime('%Y-%m-%d %H:%M:%S'), 'end_at': timezone.localtime(my_onlineexam.online_exam.end_at).strftime('%Y-%m-%d %H:%M:%S'), 'status': my_onlineexam.status, 'status_display': my_onlineexam.get_status_display(), 'number_of_questions': my_onlineexam.online_exam.number_of_questions, 'total_score': my_onlineexam.online_exam.total_score, 'passing_score': my_onlineexam.online_exam.passing_score, 'score': my_onlineexam.score, 'passed': my_onlineexam.passed, 'is_force_submitted': my_onlineexam.is_force_submitted, 'force_submit_reason': my_onlineexam.force_submit_reason, 'cover_url': my_onlineexam.online_exam.cover_image.url, 'cert_url': my_onlineexam.cert.private_url if my_onlineexam.cert else None, } }) class MyExamRegistrationEndpoint(MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def get(self, request, my_onlineexam, *args, **kwargs): registration, created = OnlineExamRegistrate.objects.get_or_create( user=request.user, online_exam=my_onlineexam.online_exam ) online_exam = my_onlineexam.online_exam available_sessions = [] now = timezone.now() for exam in OnlineExam.objects.filter( flag=online_exam.flag, registration_deadline__gt=now): exam_start_at = timezone.localtime(exam.start_at) exam_end_at = timezone.localtime(exam.end_at) available_sessions.append({ 'display': '{} ~ {}'.format( exam_start_at.strftime('%Y年%m月%d日 %H:%M'), exam_end_at.strftime('%H:%M')), 'start_at': exam_start_at.strftime('%Y-%m-%d %H:%M:%S%z'), 'end_at': exam_end_at.strftime('%Y-%m-%d %H:%M:%S%z'), 'exam_id': exam.id }) return Response({ 'detail': _('success'), 'data': { 'name': registration.name, 'chinese_id': registration.chinese_id, 'phone': registration.phone, 'email': registration.email, 'company': registration.company, 'city': registration.city, 'tug_role': registration.tug_role, 'obtained_dba_certs': registration.obtained_dba_certs, 'exam_modification_remaining': registration.exam_modification_remaining, 'can_update_exam': registration.can_update_exam, 'available_sessions': available_sessions, 'exam': { 'id': online_exam.id, 'slug': online_exam.slug, 'flag': online_exam.flag, 'time_display': '{} ~ {}'.format( timezone.localtime(online_exam.start_at).strftime('%Y年%m月%d日 %H:%M'), timezone.localtime(online_exam.end_at).strftime('%H:%M')), } } }) def post(self, request, my_onlineexam, *args, **kwargs): registration, created = OnlineExamRegistrate.objects.get_or_create( user=request.user, online_exam=my_onlineexam.online_exam) form = ExamEnrollUpdateForm( user_online_exam=my_onlineexam, registration=registration, data=request.data) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors, }, status=400) target_exam = form.save() return Response({ 'detail': _('success'), 'data': { 'exam': { 'id': target_exam.id, 'slug': target_exam.slug } } }) class ExamPaperMixin: def check_permission(self, my_onlineexam, ip): if not check_ip_in_allowlist(ip, my_onlineexam.online_exam.ip_allowlist): return False, 0, '考试设置了 IP 白名单, 你必须在指定环境下参加考试' exam_start_at = my_onlineexam.online_exam.start_at exam_end_at = my_onlineexam.online_exam.end_at if timezone.now() < exam_start_at or timezone.now() > (exam_end_at + timedelta(minutes=3)): return False, 1, '请在指定时间范围内参加考试' if my_onlineexam.status == UserOnlineExam.STATUS_FINISHED: return False, 2, '已交卷' if my_onlineexam.status == UserOnlineExam.STATUS_WAIT: return False, 3, '请先进行人脸识别，再继续参加考试' if my_onlineexam.status != UserOnlineExam.STATUS_UNDERWAY: return False, 4, '考试状态错误' return True, None, None class MyExamEnvCheckEndpoint(MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def get(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) return Response({ 'detail': _('success') }) def check_permission(self, my_onlineexam, ip): if not check_ip_in_allowlist(ip, my_onlineexam.online_exam.ip_allowlist): return False, 0, '考试设置了 IP 白名单, 你必须在指定环境下参加考试' exam_start_at = my_onlineexam.online_exam.start_at exam_end_at = my_onlineexam.online_exam.end_at if timezone.now() < exam_start_at or timezone.now() > (exam_end_at + timedelta(minutes=3)): return False, 1, '请在指定时间范围内参加考试' if my_onlineexam.status == UserOnlineExam.STATUS_FINISHED: return False, 2, '已交卷' return True, None, None class MyExamFaceIdentificationEndpoint(MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def post(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) form = FaceIdentificationForm( user=request.user, user_online_exam=my_onlineexam, data=request.data ) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors }, status=400) else: form.save() return Response({'detail': 'success'}) def check_permission(self, my_onlineexam, ip): if not check_ip_in_allowlist(ip, my_onlineexam.online_exam.ip_allowlist): return False, 0, '考试设置了 IP 白名单, 你必须在指定环境下参加考试' exam_start_at = my_onlineexam.online_exam.start_at exam_end_at = my_onlineexam.online_exam.end_at if timezone.now() < exam_start_at or timezone.now() > (exam_end_at + timedelta(minutes=3)): return False, 1, '请在指定时间范围内参加考试' if my_onlineexam.status == UserOnlineExam.STATUS_FINISHED: return False, 2, '已交卷' return True, None, None class MyExamExamineeInfoEndpoint(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def post(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) form = ExamineeInfoForm( user=request.user, user_online_exam=my_onlineexam, data=request.data ) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors }, status=400) else: form.save() return Response({'detail': 'success'}) class MyExamPaper(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def get(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) now = timezone.localtime() now_str = now.strftime('%Y-%m-%d %H:%M:%S') now_timestamp = time.mktime(now.timetuple()) start_at = timezone.localtime(my_onlineexam.online_exam.start_at) end_at = timezone.localtime(my_onlineexam.online_exam.end_at) start_at_str = start_at.strftime('%Y-%m-%d %H:%M:%S') end_at_str = end_at.strftime('%Y-%m-%d %H:%M:%S') start_at_timestamp = time.mktime(start_at.timetuple()) end_at_timestamp = time.mktime(end_at.timetuple()) structure = my_onlineexam.online_exam.structure for section in structure: shuffle(section['questions']) for q in section['questions']: _options = q['options'] _temp = _options[:-1] shuffle(_temp) _temp.append(_options[-1]) q['options'] = _temp OnlineExamEvent.objects.create( user=request.user, online_exam=my_onlineexam.online_exam, user_online_exam=my_onlineexam, event_type=OnlineExamEvent.EVENT_GET_PAPER, ip=get_client_ip(request), user_agent=get_client_agent(request), ) return Response({ 'detail': _('success'), 'data': { 'name': my_onlineexam.online_exam.name, 'slug': my_onlineexam.online_exam.slug, 'short_description': my_onlineexam.online_exam.short_description, 'rules': my_onlineexam.online_exam.rules, 'detailed_description': my_onlineexam.online_exam.detailed_description, 'start_at': start_at_str, 'end_at': end_at_str, 'start_at_timestamp': start_at_timestamp, 'end_at_timestamp': end_at_timestamp, 'now_timestamp': now_timestamp, 'now': now_str, 'number_of_questions': my_onlineexam.online_exam.number_of_questions, 'total_score': my_onlineexam.online_exam.total_score, 'passing_score': my_onlineexam.online_exam.passing_score, 'structure': my_onlineexam.online_exam.structure, 'draft': my_onlineexam.draft } }) class MyExamDraftEndpoint(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def post(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) my_onlineexam.draft = request.data my_onlineexam.save() OnlineExamEvent.objects.create( user=request.user, online_exam=my_onlineexam.online_exam, user_online_exam=my_onlineexam, event_type=OnlineExamEvent.EVENT_SAVE_DRAFT, ip=get_client_ip(request), user_agent=get_client_agent(request) ) return Response({ 'detail': _('success') }) class MyExamSubmitEndpoint(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def post(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) data = request.data user_answers = data.get('answers', {}) my_onlineexam.answers = user_answers my_onlineexam.status = UserOnlineExam.STATUS_FINISHED my_onlineexam.end_at = timezone.now() my_onlineexam.save() OnlineExamEvent.objects.create( user=request.user, online_exam=my_onlineexam.online_exam, user_online_exam=my_onlineexam, event_type=OnlineExamEvent.EVENT_SUBMIT, ip=get_client_ip(request), user_agent=get_client_agent(request), ) return Response({ 'detail': _('success') }) class MyExamExamineeCaptureEndpoint(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def post(self, request, my_onlineexam, *args, **kwargs): valid, error_code, error_detail = self.check_permission( my_onlineexam, get_client_ip(request)) if not valid: return Response({ 'detail': error_detail, 'error_code': error_code }, status=403) form = ExamineeCaptureForm( user=request.user, data=request.data ) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors }, status=400) image = form.cleaned_data['image'] content_type, base64string = image.split(':', 1)[1].split(';', 1) file_suffix = content_type.split('/')[1] base64string = base64string.split(',', 1)[1] img = base64.b64decode(base64string) content_hash = hashlib.sha1(img).hexdigest() file_obj = BytesIO() file_obj.write(img) file_obj.seek(0) file_path = '{}.{}'.format(generate_random_file_path(), file_suffix) default_storage.save(file_path, file_obj) upload = Upload.objects.create( user=request.user, original_filename=file_path, filesize=len(img), path=file_path, sha1=content_hash, extension=content_type, is_private=True ) exam_face = UserOnlineExamFace.objects.create( user=request.user, online_exam=my_onlineexam.online_exam, user_online_exam=my_onlineexam, face=upload, face_type=UserOnlineExamFace.FACE_TYPE_CAPTURE ) async_check_examinee_face.delay(exam_face.id) return Response({ 'detail': 'success' }) class MyExamEventEndpoint(ExamPaperMixin, MyOnlineExamEndpoint): permission_classes = (IsAuthenticated, ) def get(self, request, my_onlineexam, *args, **kwargs): event_type = request.GET.get('type') force_submit = False if event_type not in [ OnlineExamEvent.EVENT_EXIT_FULLSCREEN, OnlineExamEvent.EVENT_LOSTFOCUS]: return Response({ 'detail': 'invalid event_type' }, status=400) OnlineExamEvent.objects.create( user=request.user, online_exam=my_onlineexam.online_exam, user_online_exam=my_onlineexam, event_type=event_type, ip=get_client_ip(request), user_agent=get_client_agent(request) ) event_num = my_onlineexam.cheating_record.get(event_type, 0) event_num += 1 my_onlineexam.cheating_record[event_type] = event_num my_onlineexam.save(update_fields=['cheating_record']) if sum(my_onlineexam.cheating_record.values()) > my_onlineexam.online_exam.allowed_cheating_times: my_onlineexam.is_force_submitted = True my_onlineexam.force_submit_reason = '违规操作超过上限，您无法继续进行考试，如有疑问，请扫描下方二维码，联系工作人员' my_onlineexam.save(update_fields=['is_force_submitted', 'force_submit_reason']) force_submit = True return Response({ 'detail': _('success'), "data": { "force_submit": force_submit } }) class MyExamDebugReset(Endpoint): permission_classes = (IsAuthenticated, ) def get(self, request, *args, **kwargs): user = request.user try: user_exam = UserOnlineExam.objects.get( user=user, online_exam_id=3 ) user_exam.identity_face = None user_exam.save() UserOnlineExamFace.objects.filter(user_online_exam=user_exam).delete() OnlineExamEvent.objects.filter(user_online_exam=user_exam).delete() user_exam.delete() except Exception as e: print(e) UserOnlineExam.objects.create( user=user, online_exam_id=3, status=UserOnlineExam.STATUS_WAIT ) return Response({ 'detail': _('success') }) class SetFaceIdResultDebug(Endpoint): permission_classes = (IsAuthenticated, ) def post(self, request, *args, **kwargs): user = request.user user.face_audit_status = request.data['status'] user.save(update_fields=['face_audit_status']) return Response({ 'detail': _('success') })