import json import logging import urllib.parse from datetime import timedelta from odoo import http, fields from odoo.http import request from odoo.exceptions import UserError, AccessError from ..controllers.paper_pdf_reportlab import render_paper_pdf _logger = logging.getLogger(__name__) class ExamController(http.Controller): """学生答题卡式考试页面控制器。""" def _check_paper_access(self, paper_id, states=None): """校验当前登录用户是否为该答卷的学员本人。""" paper = request.env['yuthon.school.exam.paper'].sudo().browse(paper_id) if not paper.exists(): raise http.NotFound() if paper.user_id.id != request.env.user.id: raise AccessError('您无权访问该答卷。') if states and paper.state not in states: raise UserError('当前答卷状态不允许该操作。') return paper # ------------------------------------------------------------------ # 1. 开始考试：初始化答题记录并跳转到答题页面 # ------------------------------------------------------------------ @http.route('/exam/start/', type='http', auth='user') def exam_start(self, paper_id, **kw): paper = self._check_paper_access(paper_id, states=('pending', 'in_progress')) try: paper.with_user(request.env.user).action_start() except UserError as e: return request.render('college.exam_error_page', {'message': str(e)}) return request.redirect('/exam/page/%s' % paper_id) # ------------------------------------------------------------------ # 2. 答题主页面 # ------------------------------------------------------------------ @http.route('/exam/page/', type='http', auth='user', website=False) def exam_page(self, paper_id, **kw): paper = self._check_paper_access(paper_id) if paper.state == 'pending': # 未开始则自动开始 try: paper.with_user(request.env.user).action_start() except UserError as e: return request.render('college.exam_error_page', {'message': str(e)}) if paper.state not in ('in_progress', 'submitted', 'graded'): return request.render('college.exam_error_page', { 'message': '该答卷当前状态为 %s，无法进入答题页面。' % dict(paper._fields['state'].selection).get(paper.state) }) answers = paper.answer_ids.sorted(lambda a: (a.sequence, a.id)) exam = paper.exam_id # 计算剩余时间（秒） remaining_seconds = 0 if paper.start_time and exam.duration_minutes: end_dt = paper.start_time + timedelta(minutes=exam.duration_minutes) remaining_seconds = max(0, int((end_dt - fields.Datetime.now()).total_seconds())) values = { 'paper': paper, 'exam': exam, 'course': paper.course_id, 'chapters': paper.chapter_ids, 'student': paper.student_id, 'answers': answers, 'answer_json': json.dumps([self._answer_to_dict(a) for a in answers]), 'remaining_seconds': remaining_seconds, 'duration_minutes': exam.duration_minutes or 0, 'question_types': dict(request.env['yuthon.school.exam.question']._fields['question_type'].selection), } return request.render('college.exam_take_page', values) def _answer_to_dict(self, answer): """将答题记录转为前端 JSON 可用的字典。""" q = answer.question_id return { 'id': answer.id, 'sequence': answer.sequence or 0, 'question_type': answer.question_type, 'question_text': answer.question_text or '', 'question_text_short': answer.question_text_short or '', 'max_score': answer.max_score, 'options': [{ 'id': opt.id, 'label': opt.label, 'content': opt.content or '', 'text': opt.name or '%s. %s' % (opt.label, opt._strip_html(opt.content or '')), } for opt in q.option_ids.sorted(lambda o: (o.sequence, o.id))], # 当前作答 'selected_option_id': answer.selected_option_id.id or False, 'selected_option_ids': answer.selected_option_ids.ids, 'answer_choice': answer.answer_choice or False, 'answer_text': answer.answer_text or '', 'is_answered': bool( answer.selected_option_id or answer.selected_option_ids or answer.answer_choice or (answer.answer_text and answer._strip_html(answer.answer_text)) ), } # ------------------------------------------------------------------ # 3. AJAX 保存单题答案 # ------------------------------------------------------------------ @http.route('/exam/save', type='json', auth='user', methods=['POST']) def exam_save_answer(self, **kw): # 注意：type='json' 路由的参数在 request.params（JSON-RPC 内层 params）， # 而非 request.jsonrequest（那是 envelope {jsonrpc, method, params, id}）。 params = request.params or {} paper_id = params.get('paper_id') answer_id = params.get('answer_id') data = params.get('data', {}) paper = self._check_paper_access(paper_id, states=('in_progress',)) answer = request.env['yuthon.school.exam.answer'].sudo().browse(int(answer_id)) if not answer.exists() or answer.paper_id.id != paper.id: return {'status': 'error', 'message': '题目不属于当前答卷。'} qtype = answer.question_type vals = {} if qtype == 'single_choice': vals['selected_option_id'] = int(data.get('selected_option_id') or 0) or False if vals['selected_option_id']: vals['selected_option_ids'] = [(6, 0, [vals['selected_option_id']])] else: vals['selected_option_ids'] = [(5,)] elif qtype == 'multiple_choice': opt_ids = [int(x) for x in data.get('selected_option_ids', []) if x] vals['selected_option_ids'] = [(6, 0, opt_ids)] vals['selected_option_id'] = opt_ids[0] if opt_ids else False elif qtype == 'true_false': vals['answer_choice'] = data.get('answer_choice') or False elif qtype in ('fill_blank', 'essay'): vals['answer_text'] = data.get('answer_text') or False answer.write(vals) return { 'status': 'ok', 'is_answered': bool( answer.selected_option_id or answer.selected_option_ids or answer.answer_choice or (answer.answer_text and answer._strip_html(answer.answer_text)) ), } # ------------------------------------------------------------------ # 4. AJAX 提交整卷 # ------------------------------------------------------------------ @http.route('/exam/submit', type='json', auth='user', methods=['POST']) def exam_submit(self, **kw): params = request.params or {} paper_id = params.get('paper_id') paper = self._check_paper_access(paper_id, states=('in_progress',)) try: paper.with_user(request.env.user).action_submit() except UserError as e: return {'status': 'error', 'message': str(e)} return { 'status': 'ok', 'redirect': '/exam/result/%s' % paper.id, } # ------------------------------------------------------------------ # 5. 结果页面 # ------------------------------------------------------------------ @http.route('/exam/result/', type='http', auth='user', website=False) def exam_result(self, paper_id, **kw): paper = self._check_paper_access(paper_id) if paper.state not in ('submitted', 'graded'): return request.redirect('/exam/page/%s' % paper_id) answers = paper.answer_ids.sorted(lambda a: (a.sequence, a.id)) values = { 'paper': paper, 'exam': paper.exam_id, 'answers': answers, 'question_types': dict(request.env['yuthon.school.exam.question']._fields['question_type'].selection), } return request.render('college.exam_result_page', values) # ------------------------------------------------------------------ # 6. 老师评阅页面：仿学生结果页，在详情页中给简答题打分 # ------------------------------------------------------------------ def _check_review_access(self, paper_id, allow_student_readonly=False): """校验当前登录用户是否为该答卷对应考试的发布老师或监督老师。 当 allow_student_readonly=True 时，也允许学生只读查看自己的已评阅卷子。 """ paper = request.env['yuthon.school.exam.paper'].sudo().browse(paper_id) if not paper.exists(): raise http.NotFound() user = request.env.user # 老师和系统管理员保持原有权限（可评阅/可查看） if user.has_group('college.group_teacher') or user.has_group('base.group_system'): exam = paper.exam_id if user.id != exam.teacher_id.id and user.id not in exam.supervisor_teacher_ids.ids: raise AccessError('您不是该考试的发布老师或监督老师，无权评阅此答卷。') if paper.state not in ('submitted', 'graded'): raise UserError('当前答卷状态为 %s，不能评阅。' % dict(paper._fields['state'].selection).get(paper.state)) return paper # 学生只读访问：只能查看自己的已评阅卷子 if allow_student_readonly and user.has_group('college.group_student'): if paper.user_id.id != user.id: raise AccessError('您只能查看自己的答卷。') if paper.state != 'graded': raise UserError('当前答卷尚未评阅，无法查看结果。') return paper raise AccessError('仅老师可访问评阅页面。') @http.route('/exam/review/', type='http', auth='user', website=False) def exam_review(self, paper_id, **kw): paper = self._check_review_access(paper_id, allow_student_readonly=(kw.get('mode') == 'readonly')) readonly = kw.get('mode') == 'readonly' answers = paper.answer_ids.sorted(lambda a: (a.sequence, a.id)) values = { 'paper': paper, 'exam': paper.exam_id, 'answers': answers, 'readonly': readonly, 'question_types': dict(request.env['yuthon.school.exam.question']._fields['question_type'].selection), } return request.render('college.exam_review_page', values) @http.route('/exam/grade_answer', type='json', auth='user', methods=['POST']) def exam_grade_answer(self, **kw): params = request.params or {} paper_id = params.get('paper_id') answer_id = params.get('answer_id') score = params.get('score') comment = params.get('comment', '') paper = self._check_review_access(paper_id) answer = request.env['yuthon.school.exam.answer'].sudo().browse(int(answer_id)) if not answer.exists() or answer.paper_id.id != paper.id: return {'status': 'error', 'message': '题目不属于当前答卷。'} try: score_val = float(score) if score is not None else 0.0 except (ValueError, TypeError): return {'status': 'error', 'message': '分数必须是数字。'} if score_val < 0 or score_val > answer.max_score: return {'status': 'error', 'message': '分数必须在 0 到 %s 之间。' % answer.max_score} answer.write({'score': score_val, 'teacher_comment': comment or ''}) return {'status': 'ok', 'paper_score': paper.score} @http.route('/exam/finish_review', type='json', auth='user', methods=['POST']) def exam_finish_review(self, **kw): params = request.params or {} paper_id = params.get('paper_id') answer_items = params.get('answers', []) or [] paper = self._check_review_access(paper_id) # 先批量保存简答题分数和评语 for item in answer_items: answer_id = item.get('answer_id') score = item.get('score') comment = item.get('comment', '') if score is None or score == '': continue answer = request.env['yuthon.school.exam.answer'].sudo().browse(int(answer_id)) if not answer.exists() or answer.paper_id.id != paper.id: return {'status': 'error', 'message': '题目不属于当前答卷。'} try: score_val = float(score) except (ValueError, TypeError): return {'status': 'error', 'message': '分数必须是数字。'} if score_val < 0 or score_val > answer.max_score: return {'status': 'error', 'message': '第 %s 题分数必须在 0 到 %s 之间。' % (answer.sequence, answer.max_score)} answer.write({'score': score_val, 'teacher_comment': comment or ''}) try: paper.with_user(request.env.user).action_save_grade() except UserError as e: return {'status': 'error', 'message': str(e)} return {'status': 'ok', 'state': paper.state, 'paper_score': paper.score} # ------------------------------------------------------------------ # 7. 下载已批改答卷 PDF # ------------------------------------------------------------------ def _render_paper_pdf(self, paper): # 使用 reportlab 直接生成 PDF，绕过 Windows 版 wkhtmltopdf 无法加载中文字体的问题 pdf = render_paper_pdf(paper) ascii_name = 'exam_paper_%s.pdf' % paper.id cn_name = '答卷_%s_%s.pdf' % (paper.exam_id.name or 'exam', paper.student_id.name or 'student') disposition = "attachment; filename=%s; filename*=UTF-8''%s" % ( ascii_name, urllib.parse.quote(cn_name)) return request.make_response(pdf, headers=[ ('Content-Type', 'application/pdf'), ('Content-Length', str(len(pdf))), ('Content-Disposition', disposition), ]) @http.route('/exam/download/', type='http', auth='user') def exam_download(self, paper_id, **kw): paper = self._check_paper_access(paper_id) if paper.state not in ('submitted', 'graded'): return request.redirect('/exam/page/%s' % paper_id) return self._render_paper_pdf(paper) @http.route('/exam/review/download/', type='http', auth='user') def exam_review_download(self, paper_id, **kw): # 学生只读查看页里的「下载试卷 PDF」也走这里，需允许学生下载自己的已评阅卷子 paper = self._check_review_access(paper_id, allow_student_readonly=True) return self._render_paper_pdf(paper) @http.route('/exam/review/download/batch', type='http', auth='user') def exam_review_download_batch(self, **kw): """批量下载引导页：根据 ids 逐份校验权限，列出有权下载的答卷， 前端依次触发每份独立 PDF 下载（浏览器单请求只能落一个文件，故不做合集）。 学生只能下载自己已评阅的答卷；老师/管理员只能下载自己任教/监督考试的答卷。 单份（ids 仅一个）时自动直接下载该份，无需二次点击。 """ user = request.env.user is_staff = user.has_group('college.group_teacher') or user.has_group('base.group_system') ids_raw = (request.params.get('ids') or '').strip() paper_ids = [int(x) for x in ids_raw.split(',') if x.strip().isdigit()] if not paper_ids: return request.render('college.exam_error_page', {'message': '请先勾选要下载的答卷。'}) papers = request.env['yuthon.school.exam.paper'].sudo().browse(paper_ids) papers = papers.filtered(lambda p: p.exists()) # 逐份校验权限，过滤掉无权的卷子 allowed = [] for p in papers: if is_staff: exam = p.exam_id if user.id != exam.teacher_id.id and user.id not in exam.supervisor_teacher_ids.ids: continue if p.state not in ('submitted', 'graded'): continue else: # 学生：只能下载自己的已评阅卷子 if p.user_id.id != user.id: continue if p.state != 'graded': continue allowed.append(p) if not allowed: return request.render('college.exam_error_page', {'message': '您没有权限下载所选的答卷，或所选答卷尚未评阅。'}) if len(allowed) == 1: # 单份：直接下载该份独立 PDF return self._render_paper_pdf(allowed[0]) paper_infos = [{ 'id': p.id, 'name': '%s - %s' % (p.exam_id.name or '考试', p.student_id.name or '学员'), 'url': '/exam/review/download/%s' % p.id, } for p in allowed] return request.render('college.exam_batch_download_page', { 'count': len(allowed), 'papers': paper_infos, })