import logging from datetime import timedelta from django.utils import timezone from django.utils.translation import gettext as _ from django.utils.encoding import escape_uri_path from django.core.files.storage import default_storage from django.db import transaction from rest_framework.response import Response from rest_framework.permissions import IsAdminUser from rest_framework_csv import renderers as csv_r # from rest_framework.exceptions import NotFound from university.api.base import Endpoint from university.api.bases import AdminOnlineExamEndpoint, AdminOnlineExamExamineeEndpoint from university.models import ( OnlineExam, UserOnlineExam, OnlineExamEvent, UserOnlineExamFace, OnlineExamRegistrate, OnlineExamPromotionCode ) from university.forms.admin_online_exam import ( AdminMessageForm, OnlineExamEditForm, OnlineExamUploadCertForm ) logger = logging.getLogger(__name__) class ExamListEndpoint(Endpoint): permission_classes = (IsAdminUser, ) def get(self, request, *args, **kwargs): return Response({ 'detail': _('success'), 'data': [{ 'id': e.id, 'name': e.name, 'slug': e.slug, 'registration_deadline': timezone.localtime(e.registration_deadline).strftime('%Y-%m-%d/%H:%M'), 'start_at': timezone.localtime(e.start_at).strftime('%Y-%m-%d/%H:%M'), 'end_at': timezone.localtime(e.end_at).strftime('%Y-%m-%d/%H:%M'), 'status_distrubution': e.get_status_distribution(), 'total_score': e.total_score, 'passing_score': e.passing_score, 'number_of_questions': e.number_of_questions } for e in OnlineExam.objects.all().order_by('-start_at')] }) class ExamDetaiEndpoint(AdminOnlineExamEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, online_exam, *args, **kwargs): return Response({ 'detail': _('success'), 'data': { 'id': online_exam.id, 'name': online_exam.name, 'slug': online_exam.slug, 'registration_deadline': timezone.localtime(online_exam.registration_deadline).strftime('%Y-%m-%d %H:%M') if online_exam.registration_deadline else None, 'start_at': timezone.localtime(online_exam.start_at).strftime('%Y-%m-%d %H:%M') if online_exam.start_at else None, 'end_at': timezone.localtime(online_exam.end_at).strftime('%Y-%m-%d %H:%M') if online_exam.end_at else None, 'status_distrubution': online_exam.get_status_distribution(), 'total_score': online_exam.total_score, 'passing_score': online_exam.passing_score, 'number_of_questions': online_exam.number_of_questions, 'short_description': online_exam.short_description, 'detailed_description': online_exam.detailed_description, 'rules': online_exam.rules, 'allowed_cheating_times': online_exam.allowed_cheating_times, 'alert_wechat_webhook': online_exam.alert_wechat_webhook } }) def put(self, request, online_exam, *args, **kwargs): form = OnlineExamEditForm(data=request.data, instance=online_exam) if not form.is_valid(): return Response({ 'detail': _('params error'), 'errors': form.errors }, status=400) form.save() return Response({ 'detail': _('success') }) class ExamineeeListEndpoint(AdminOnlineExamEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, online_exam, *args, **kwargs): user_online_exams = online_exam.useronlineexam_set\ .select_related('user', 'online_exam').all() data = [] for u_exam in user_online_exams: registration, created = OnlineExamRegistrate.objects.get_or_create( user=u_exam.user, online_exam=online_exam) data.append({ 'user': { 'id': u_exam.user.id, 'username': u_exam.user.username, 'phone': u_exam.user.phone, 'email': u_exam.user.email, 'exam_real_name': u_exam.examinee_name }, 'id': u_exam.id, 'status': u_exam.status, 'status_display': u_exam.get_status_display(), 'face_identity_passed': u_exam.face_identity_passed, 'score': u_exam.score, 'passed': u_exam.passed, 'cert_url': u_exam.cert.private_url if u_exam.cert else None, 'start_at': timezone.localtime(u_exam.start_at).strftime('%Y-%m-%d/%H:%M') if u_exam.start_at else None, 'end_at': timezone.localtime(u_exam.end_at).strftime('%Y-%m-%d/%H:%M') if u_exam.end_at else None, 'is_force_submitted': u_exam.is_force_submitted, 'force_submit_reason': u_exam.force_submit_reason, 'payment_method': registration.get_payment_method_display(), 'promotion_code': registration.promotion_code.code if registration.promotion_code else None, }) return Response({ 'detail': _('success'), 'data': data }) class ExamineeeListDownloadEndpoint(AdminOnlineExamEndpoint): permission_classes = (IsAdminUser, ) renderer_classes = (csv_r.CSVRenderer, ) valid_fields = {} filename = '' def get_renderer_context(self): context = super().get_renderer_context() context['header'] = self.valid_fields.keys() context['labels'] = self.valid_fields context['encoding'] = 'utf-8_sig' return context def get(self, request, online_exam, *args, **kwargs): self.filename = escape_uri_path('{}-考生列表.csv'.format(online_exam.name)) self.valid_fields = { 'user_openid': '用户 ID', 'username': '用户名', 'name': '姓名', 'chinese_id': '身份证号', 'phone': '手机号', 'email': '邮箱', 'company': '公司名称', 'city': '所在城市', 'status': '考试状态', 'start_at': '考试开始时间', 'end_at': '考试结束时间', 'score': '成绩', 'passed': '是否通过', 'is_force_submitted': '是否强制交卷', 'force_submit_reason': '强制交卷原因', 'payment_method': '报名付款方式', 'promotion_code': '优惠码', 'need_receipt': '是否需要发票', 'receipt_header': '发票抬头', 'receipt_tin': '纳税人识别号', } user_online_exams = online_exam.useronlineexam_set\ .select_related('user', 'online_exam').all() data = [] for u_e in user_online_exams: registration, created = OnlineExamRegistrate.objects.get_or_create( user=u_e.user, online_exam=online_exam) data.append({ 'user_openid': u_e.user.singlesignonrecord_set.first().external_id, 'username': u_e.user.username, 'name': u_e.examinee_name, 'chinese_id': u_e.examinee_chinese_id, 'phone': registration.phone, 'email': registration.email, 'company': registration.company, 'city': registration.city, 'status': u_e.get_status_display(), 'passed': 'Y' if u_e.passed else 'N', 'score': u_e.score, 'cert_url': u_e.cert.private_url if u_e.cert else None, 'start_at': timezone.localtime(u_e.start_at).strftime('%Y-%m-%d/%H:%M') if u_e.start_at else None, 'end_at': timezone.localtime(u_e.end_at).strftime('%Y-%m-%d/%H:%M') if u_e.end_at else None, 'is_force_submitted': 'Y' if u_e.is_force_submitted else 'N', 'force_submit_reason': u_e.force_submit_reason, 'payment_method': registration.get_payment_method_display(), 'promotion_code': registration.promotion_code.code if registration.promotion_code else None, 'need_receipt': 'Y' if registration.need_receipt else 'N', 'receipt_header': registration.receipt_header, 'receipt_tin': registration.receipt_tin }) return Response(data) def finalize_response(self, request, response, *args, **kwargs): resp = super(ExamineeeListDownloadEndpoint, self).finalize_response( request, response, *args, **kwargs) resp['Content-Disposition'] = "attachment; filename={}".format(self.filename) return resp class ExamineeeDetailEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): answerd_question_num = sum(1 if value else 0 for value in user_online_exam.draft.values()) registration, created = OnlineExamRegistrate.objects.get_or_create( user=exam_user, online_exam=online_exam) registration_data = { 'name': registration.name, 'chinese_id': registration.chinese_id, 'phone': registration.phone, 'company': registration.company, 'city': registration.city, 'payment_method': registration.payment_method, 'payment_method_display': registration.get_payment_method_display(), 'online_payment_success': registration.online_payment_success } return Response({ 'detail': _('success'), 'data': { 'user': { 'id': exam_user.id, 'username': exam_user.username, 'phone': exam_user.phone, 'email': exam_user.email }, 'exam': { 'total_score': online_exam.total_score, 'passing_score': online_exam.passing_score, 'number_of_questions': online_exam.number_of_questions }, 'status': user_online_exam.status, 'status_display': user_online_exam.get_status_display(), 'score': user_online_exam.score, 'examinee_name': user_online_exam.examinee_name, 'examinee_chinese_id': user_online_exam.examinee_chinese_id, 'face_identity_passed': user_online_exam.face_identity_passed, 'identity_face_image_url': user_online_exam.identity_face.face.private_url if user_online_exam.identity_face else None, 'start_at': timezone.localtime(user_online_exam.start_at).strftime('%Y-%m-%d/%H:%M') if user_online_exam.start_at else None, 'end_at': timezone.localtime(user_online_exam.end_at).strftime('%Y-%m-%d/%H:%M') if user_online_exam.end_at else None, 'is_force_submitted': user_online_exam.is_force_submitted, 'force_submit_reason': user_online_exam.force_submit_reason, 'answerd_question_num': answerd_question_num, 'cheating_record': user_online_exam.cheating_record, 'registration': registration_data } }) def delete(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): admin = request.user examinee = user_online_exam.user registration, created = OnlineExamRegistrate.objects.get_or_create( user=examinee, online_exam=online_exam) promotion_code = OnlineExamPromotionCode.objects.filter(registration=registration) action_log = 'Admin(user_id: {}, username: {}) Delete User Online Exam, ' 'online_exam_id: {}, examinee_id: {}, examinee_name: {},' 'examinee_company: {}'.format( admin.id, admin.username, online_exam.id, examinee.id, registration.name, registration.company ) with transaction.atomic(): user_online_exam.delete() promotion_code.update(registration=None) registration.delete() logger.info(action_log) return Response() class ExamineeeCaptureListEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): user_exam_faces = user_online_exam.useronlineexamface_set\ .select_related('face').all().order_by('-created_at') return Response({ 'detail': _('success'), 'data': [{ 'id': f.id, 'face_image_url': f.face.private_url, 'face_type': f.face_type, 'face_type_display': f.get_face_type_display(), 'status': f.status, 'status_display': f.get_status_display(), 'comment': f.comment, 'created_at': timezone.localtime(f.created_at).strftime('%Y-%m-%d/%H:%M') } for f in user_exam_faces] }) class ExamineeeEventListEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): events = user_online_exam.onlineexamevent_set.all().order_by('-created_at') return Response({ 'detail': _('success'), 'data': [{ 'id': e.id, 'event_type': e.event_type, 'event_type_display': e.get_event_type_display(), 'comment': e.comment, 'ip': e.ip, 'user_agent': e.user_agent, 'created_at': timezone.localtime(e.created_at).strftime('%Y-%m-%d/%H:%M') } for e in events] }) class ExamineeeConfirmRegistrationEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): if user_online_exam.status == UserOnlineExam.STATUS_REGISTE_WAIT_CONFIRM: user_online_exam.status = UserOnlineExam.STATUS_REGISTERED user_online_exam.save(update_fields=['status', 'updated_at']) return Response({ 'detail': _('success'), }) class ExamineeeUploadCertEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): form = OnlineExamUploadCertForm(user_online_exam, files=request.FILES) if not form.is_valid(): return Response({ 'detail': _('params error'), 'errors': form.errors }, status=400) form.save() return Response({ 'detail': _('success'), }) class ExamineeeForceSubmitEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): form = AdminMessageForm( user_online_exam=user_online_exam, data=request.data ) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors }, status=400) else: form.send_force_submit() return Response({'detail': 'success'}) class ExamineeeSendWarningMsgEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): form = AdminMessageForm( user_online_exam=user_online_exam, data=request.data ) if not form.is_valid(): return Response({ 'detail': _('wrong params'), 'errors': form.errors }, status=400) else: form.send_warning_message() return Response({'detail': 'success'}) class ExamineeeForceRestEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): if user_online_exam.status == UserOnlineExam.STATUS_UNDERWAY: return Response({'detail': 'status != 2'}, status=400) user_online_exam.status = UserOnlineExam.STATUS_WAIT user_online_exam.is_force_submitted = False user_online_exam.force_submit_reason = None user_online_exam.face_identity_num = 0 user_online_exam.face_identity_passed = False user_online_exam.cheating_record = {} user_online_exam.save() return Response({'detail': 'success'}) class ExamineeeSkipFaceidCheckEndpoint(AdminOnlineExamExamineeEndpoint): permission_classes = (IsAdminUser, ) def post(self, request, user_online_exam, online_exam, exam_user, *args, **kwargs): if user_online_exam.status == UserOnlineExam.STATUS_UNDERWAY: return Response({'detail': 'status != 2'}, status=400) user_online_exam.status = UserOnlineExam.STATUS_UNDERWAY user_online_exam.face_identity_passed = True user_online_exam.face_identity_num = 0 user_online_exam.save() return Response({'detail': 'success'}) class ExamInvigilationEndpoint(AdminOnlineExamEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, online_exam, *args, **kwargs): try: per_page = int(self.request.GET.get('per_page', 30)) cur_page = int(self.request.GET.get('cur_page', 1)) except Exception: return Response({ 'detail': _('params error') }, status=400) a_minute_ago = timezone.now() - timedelta(minutes=1) user_online_exams = online_exam.useronlineexam_set.select_related('user')\ .filter(status=UserOnlineExam.STATUS_UNDERWAY) total_num = user_online_exams.count() user_online_exams = user_online_exams[(cur_page - 1) * per_page:cur_page * per_page] user_online_exams_ids = list(user_online_exams.values_list('id', flat=True)) captures = online_exam.useronlineexamface_set.filter( user_online_exam_id__in=user_online_exams_ids, user_online_exam__status=UserOnlineExam.STATUS_UNDERWAY, created_at__gte=a_minute_ago ).values( 'face__path', 'user_id', 'status', 'comment', 'created_at' ).order_by('created_at') events = online_exam.onlineexamevent_set.filter( user_online_exam_id__in=user_online_exams_ids, user_online_exam__status=UserOnlineExam.STATUS_UNDERWAY, created_at__gte=a_minute_ago ).exclude( event_type=OnlineExamEvent.EVENT_SAVE_DRAFT ).values( 'user_id', 'comment', 'created_at', 'event_type' ).order_by('created_at') registrations = OnlineExamRegistrate.objects.filter( online_exam=online_exam).values('phone', 'email', 'user_id') captures = {d['user_id']: d for d in captures} events = {d['user_id']: d for d in events} registrations = {d['user_id']: d for d in registrations} data = [] for u_exam in user_online_exams: user_id = u_exam.user_id _registration = registrations.get(user_id, {}) _user_data = { 'real_name': u_exam.examinee_name, 'chinese_id': u_exam.examinee_chinese_id, 'phone': _registration.get('phone'), 'email': _registration.get('email'), 'username': u_exam.user.username, 'id': user_id } _recent_capture, _recent_event = None, None if user_id in captures: _recent_capture = { 'private_url': default_storage.sign_url(captures[user_id]['face__path'], 60), 'path': captures[user_id]['face__path'], 'status': captures[user_id]['status'], 'status_display': dict(UserOnlineExamFace.STATUS_CHOICES).get( captures[user_id]['status'], '未知'), 'comment': captures[user_id]['comment'], 'created_at': timezone.localtime( captures[user_id]['created_at'],).strftime('%H:%M:%S') } if user_id in events: _recent_event = { 'comment': events[user_id]['comment'], 'event_type': events[user_id]['event_type'], 'event_type_display': dict(OnlineExamEvent.EVENT_CHOICES).get( events[user_id]['event_type'], '未知事件'), 'created_at': timezone.localtime( events[user_id]['created_at'],).strftime('%H:%M:%S') } answerd_question_num = sum(1 if value else 0 for value in u_exam.draft.values()) answerd_question_percent = round( answerd_question_num / online_exam.number_of_questions * 100, 2) data.append({ 'answerd_question_num': answerd_question_num, 'answerd_question_percent': answerd_question_percent, 'cheating_record': u_exam.cheating_record, 'user': _user_data, 'recent_capture': _recent_capture, 'recent_event': _recent_event }) return Response({ 'detail': _('success'), 'data': data, 'total_num': total_num, 'per_page': per_page, 'cur_page': cur_page }) class ExamInvigilationReckonScoreEndpoint(AdminOnlineExamEndpoint): permission_classes = (IsAdminUser, ) def get(self, request, online_exam, *args, **kwargs): user_online_exams = UserOnlineExam.objects.select_related('user').filter( online_exam=online_exam, status__in=[UserOnlineExam.STATUS_FINISHED, UserOnlineExam.STATUS_UNDERWAY] ) user_count = user_online_exams.count() passed_count = 0 score_collection = [] for user_exam in user_online_exams: _score = 0 _passed = False _user_answers = None if user_exam.status == UserOnlineExam.STATUS_UNDERWAY: _user_answers = user_exam.draft else: _user_answers = user_exam.answers _answers = {k: (v, _user_answers.get(k)) for k, v in online_exam.answers.items()} for k, v in _answers.items(): _flag = False if isinstance(v[0], str): if isinstance(v[1], str) and v[0] == v[1]: _flag = True if isinstance(v[1], list) and len(v[1]) == 1 and v[0] == v[1][0]: _flag = True if isinstance(v[0], list) and isinstance(v[1], list) and (set(v[0]) == set(v[1])): _flag = True if _flag: _score += 1 if _score >= online_exam.passing_score: _passed = True passed_count += 1 score_collection.append({ 'username': user_exam.user.username, 'user_id': user_exam.user.id, 'real_name': user_exam.examinee_name, 'score': _score, 'passed': _passed, 'status': user_exam.status, 'status_display': user_exam.get_status_display(), }) return Response({ 'detail': _('success'), 'data': { 'total_count': user_count, 'passed_count': passed_count, 'reckon_results': score_collection } })