Files
2026-08-27 19:53:56 +08:00

372 lines
17 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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/<int:paper_id>', 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/<int:paper_id>', 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/<int:paper_id>', 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/<int:paper_id>', 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/<int:paper_id>', 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/<int:paper_id>', 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,
})