LIMUXI/zlqy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mobile app chat contest_detail exam_detail exam_list themes

Browse Source
This commit is contained in:
limuxi
2026-03-09 18:50:22 +08:00
parent 7dec929e29
commit 6ccfe9b107
5 changed files with 174 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
app.py
View File

@@ -151,10 +151,15 @@ def decrypt_questions(encrypted_str):
return decrypted.decode('utf-8')
def get_exam_questions(exam):
"""统一获取考试题目（自动处理加密）"""
"""统一获取考试题目（自动处理加密，确保每题有id）"""
if exam.is_encrypted and exam.encrypted_questions:
return json.loads(decrypt_questions(exam.encrypted_questions))
return exam.get_questions()
questions = json.loads(decrypt_questions(exam.encrypted_questions))
else:
questions = exam.get_questions()
for i, q in enumerate(questions):
if 'id' not in q:
q['id'] = i + 1
return questions
def teacher_required(f):
@wraps(f)
@@ -580,9 +585,11 @@ def exam_list():
Exam.creator_id == user.get('id')
)
# 杯赛负责人和杯赛老师可见其杯赛下的所有考试（用于批改）
cup_member_contest_ids = [m.contest_id for m in ContestMembership.query.filter_by(
user_contest_memberships = ContestMembership.query.filter_by(
user_id=user.get('id')).filter(
ContestMembership.role.in_(['owner', 'teacher'])).all()]
ContestMembership.role.in_(['owner', 'teacher'])).all()
cup_member_contest_ids = [m.contest_id for m in user_contest_memberships]
cup_owner_contest_ids = [m.contest_id for m in user_contest_memberships if m.role == 'owner']
if cup_member_contest_ids:
base_filter = or_(base_filter, Exam.contest_id.in_(cup_member_contest_ids))
query = query.filter(base_filter)
@@ -595,6 +602,15 @@ def exam_list():
)
all_exams = query.all()
now = datetime.now()
changed = False
for exam in all_exams:
if exam.status == 'available' and exam.scheduled_end and now > exam.scheduled_end:
exam.status = 'closed'
changed = True
if changed:
db.session.commit()
all_subjects = db.session.query(Exam.subject).distinct().all()
all_subjects = [s[0] for s in all_subjects if s[0]]
@@ -612,7 +628,9 @@ def exam_list():
user_submissions=user_submissions,
search_query=search_query,
subject_filter=subject_filter,
all_subjects=all_subjects
all_subjects=all_subjects,
now=now,
cup_owner_contest_ids=cup_owner_contest_ids
)
@app.route('/exams/create')
@@ -655,12 +673,15 @@ def exam_detail(exam_id):
draft = Draft.query.filter_by(exam_id=exam_id, user_id=user.get('id')).first()
questions = get_exam_questions(exam)
# 检查预定时间
now = datetime.utcnow()
now = datetime.now()
schedule_status = 'available' # available, not_started, ended
if exam.scheduled_start and now < exam.scheduled_start:
schedule_status = 'not_started'
if exam.scheduled_end and now > exam.scheduled_end:
schedule_status = 'ended'
if exam.status == 'available':
exam.status = 'closed'
db.session.commit()
return render_template('exam_detail.html', exam=exam, questions=questions,
existing_submission=existing, draft=draft,
schedule_status=schedule_status, need_password=False)
@@ -677,7 +698,7 @@ def exam_result(exam_id):
return redirect(url_for('exam_detail', exam_id=exam_id))
# 检查成绩公布时间
score_hidden = False
if exam.score_release_time and datetime.utcnow() < exam.score_release_time:
if exam.score_release_time and datetime.now() < exam.score_release_time:
if user.get('role') != 'teacher' and user.get('role') != 'admin':
score_hidden = True
questions = get_exam_questions(exam)
@@ -1315,6 +1336,11 @@ def api_notifications():
item['contest_name'] = contest.name if contest else ''
applicant = User.query.get(ta.user_id)
item['applicant_name'] = applicant.name if applicant else ''
if n.type == 'exam_submission' and n.post_id:
sub = Submission.query.get(n.post_id)
if sub:
item['exam_id'] = sub.exam_id
item['submission_id'] = sub.id
if n.type == 'friend_request' and n.post_id:
fr = Friend.query.get(n.post_id)
if fr:
@@ -1362,7 +1388,7 @@ def api_delete_notification(nid):
@app.route('/notifications')
@login_required
def notifications_page():
return redirect(url_for('chat'))
return redirect(url_for('chat_page'))
# ========== 系统公告管理 API ==========
@app.route('/api/system-notifications')
@@ -2753,7 +2779,7 @@ def api_submit_exam(exam_id):
if exam.status == 'closed':
return jsonify({'success': False, 'message': '该考试已关闭'}), 400
# 检查预定时间
now = datetime.utcnow()
now = datetime.now()
if exam.scheduled_start and now < exam.scheduled_start:
return jsonify({'success': False, 'message': '考试尚未开始'}), 400
if exam.scheduled_end and now > exam.scheduled_end:
@@ -2761,73 +2787,78 @@ def api_submit_exam(exam_id):
if Submission.query.filter_by(exam_id=exam_id, user_id=user.get('id')).first():
return jsonify({'success': False, 'message': '您已提交过该试卷'}), 400
data = request.get_json(force=True, silent=True)
if not data:
return jsonify({'success': False, 'message': '无效的请求数据'}), 400
answers = data.get('answers', {})
score = 0
auto_graded = True
question_scores = {}
questions = get_exam_questions(exam)
# AI批改客观题
for q in questions:
qid = str(q['id'])
if q['type'] == 'choice':
# 选择题：精确匹配
if answers.get(qid) == q.get('answer'):
score += q.get('score', 0)
question_scores[qid] = q.get('score', 0)
try:
score = 0
auto_graded = True
question_scores = {}
questions = get_exam_questions(exam)
for q in questions:
qid = str(q.get('id', ''))
q_score = q.get('score', 0) or 0
if q.get('type') == 'choice':
if answers.get(qid) == q.get('answer'):
score += q_score
question_scores[qid] = q_score
else:
question_scores[qid] = 0
elif q.get('type') == 'fill':
student_answer = (answers.get(qid) or '').strip()
correct_answer = (q.get('answer') or '').strip()
correct_answers = [a.strip() for a in correct_answer.split('|')] if correct_answer else []
if correct_answers and student_answer in correct_answers:
score += q_score
question_scores[qid] = q_score
elif correct_answers and student_answer.lower() in [a.lower() for a in correct_answers]:
score += q_score
question_scores[qid] = q_score
else:
auto_graded = False
question_scores[qid] = 0
elif q.get('type') == 'judge':
if answers.get(qid) == q.get('answer'):
score += q_score
question_scores[qid] = q_score
else:
question_scores[qid] = 0
else:
auto_graded = False
question_scores[qid] = 0
elif q['type'] == 'fill':
# 填空题：使用AI智能判断
student_answer = answers.get(qid, '').strip()
correct_answer = q.get('answer', '').strip()
# 先尝试精确匹配（支持多个正确答案，用|分隔）
correct_answers = [a.strip() for a in correct_answer.split('|')]
if student_answer in correct_answers:
score += q.get('score', 0)
question_scores[qid] = q.get('score', 0)
else:
# 使用AI判断答案是否正确
ai_score = ai_grade_fill_question(q.get('content', ''), correct_answer, student_answer, q.get('score', 0))
score += ai_score
question_scores[qid] = ai_score
elif q['type'] == 'judge':
# 判断题：精确匹配
if answers.get(qid) == q.get('answer'):
score += q.get('score', 0)
question_scores[qid] = q.get('score', 0)
else:
question_scores[qid] = 0
else:
# 主观题需要人工批改
auto_graded = False
question_scores[qid] = 0
sub = Submission(
exam_id=exam_id,
user_id=user.get('id'),
score=score,
graded=auto_graded,
graded_by='AI自动批改' if auto_graded else ''
)
sub.set_answers(answers)
sub.set_question_scores(question_scores)
db.session.add(sub)
Draft.query.filter_by(exam_id=exam_id, user_id=user.get('id')).delete()
db.session.commit()
sub = Submission(
exam_id=exam_id,
user_id=user.get('id'),
score=score,
graded=auto_graded,
graded_by='AI自动批改' if auto_graded else ''
)
sub.set_answers(answers)
sub.set_question_scores(question_scores)
db.session.add(sub)
Draft.query.filter_by(exam_id=exam_id, user_id=user.get('id')).delete()
db.session.commit()
except Exception as e:
db.session.rollback()
app.logger.error(f'考试提交失败: exam_id={exam_id}, user_id={user.get("id")}, error={str(e)}')
return jsonify({'success': False, 'message': '提交处理失败，请重试'}), 500
# 如果有主观题，通知老师批改
if not auto_graded and exam.contest_id:
contest = Contest.query.get(exam.contest_id)
if contest:
# 通知杯赛负责人和老师
memberships = ContestMembership.query.filter_by(contest_id=exam.contest_id).all()
for m in memberships:
if m.role in ('owner', 'teacher'):
add_notification(m.user_id, 'exam_submission',
f'考生 {user.get("name")} 提交了考试「{exam.title}」，请批改主观题。',
from_user=user.get('name', ''), post_id=sub.id)
try:
if not auto_graded and exam.contest_id:
contest = Contest.query.get(exam.contest_id)
if contest:
memberships = ContestMembership.query.filter_by(contest_id=exam.contest_id).all()
for m in memberships:
if m.role in ('owner', 'teacher'):
add_notification(m.user_id, 'exam_submission',
f'考生 {user.get("name")} 提交了考试「{exam.title}」，请批改主观题。',
from_user=user.get('name', ''), post_id=sub.id)
except Exception:
pass
return jsonify({'success': True, 'message': '提交成功', 'submission_id': sub.id})
@@ -2861,13 +2892,23 @@ def api_grade_submission(exam_id, sub_id):
@app.route('/api/exams/<int:exam_id>/status', methods=['POST'])
def api_update_exam_status(exam_id):
user = session.get('user')
if not user or user.get('role') not in ('teacher', 'admin'):
if not user:
return jsonify({'success': False, 'message': '无权限'}), 403
exam = Exam.query.get(exam_id)
if not exam:
return jsonify({'success': False, 'message': '试卷不存在'}), 404
if exam.creator_id != user.get('id'):
return jsonify({'success': False, 'message': '只能修改自己创建的试卷'}), 403
has_permission = False
if user.get('role') == 'admin':
has_permission = True
elif user.get('role') == 'teacher' and exam.creator_id == user.get('id'):
has_permission = True
elif exam.contest_id:
membership = ContestMembership.query.filter_by(
user_id=user['id'], contest_id=exam.contest_id).first()
if membership and membership.role == 'owner':
has_permission = True
if not has_permission:
return jsonify({'success': False, 'message': '无权限修改该试卷状态'}), 403
data = request.get_json(force=True, silent=True)
new_status = data.get('status', '')
if new_status not in ('available', 'closed'):
@@ -2915,13 +2956,23 @@ def api_update_exam_visibility(exam_id):
@app.route('/api/exams/<int:exam_id>', methods=['DELETE'])
def api_delete_exam(exam_id):
user = session.get('user')
if not user or user.get('role') not in ('teacher', 'admin'):
if not user:
return jsonify({'success': False, 'message': '无权限'}), 403
exam = Exam.query.get(exam_id)
if not exam:
return jsonify({'success': False, 'message': '试卷不存在'}), 404
if exam.creator_id != user.get('id'):
return jsonify({'success': False, 'message': '只能删除自己创建的试卷'}), 403
has_permission = False
if user.get('role') == 'admin':
has_permission = True
elif user.get('role') == 'teacher' and exam.creator_id == user.get('id'):
has_permission = True
elif exam.contest_id:
membership = ContestMembership.query.filter_by(
user_id=user['id'], contest_id=exam.contest_id).first()
if membership and membership.role == 'owner':
has_permission = True
if not has_permission:
return jsonify({'success': False, 'message': '无权限删除该试卷'}), 403
Submission.query.filter_by(exam_id=exam_id).delete()
Draft.query.filter_by(exam_id=exam_id).delete()
db.session.delete(exam)
@@ -2993,6 +3044,14 @@ def api_contest_exams(contest_id):
if user:
membership = ContestMembership.query.filter_by(user_id=user['id'], contest_id=contest_id).first()
is_owner = (membership and membership.role == 'owner') or user.get('role') == 'admin'
now = datetime.now()
changed = False
for e in exams:
if e.status == 'available' and e.scheduled_end and now > e.scheduled_end:
e.status = 'closed'
changed = True
if changed:
db.session.commit()
data = []
for e in exams:
item = {
@@ -3234,11 +3293,14 @@ def api_create_exam_from_bank(contest_id):
return jsonify({'success': False, 'message': '请填写标题并选择题目'}), 400
# 从题库获取题目并转换为考试题目格式
questions = []
q_idx = 0
for qid in question_ids:
item = QuestionBankItem.query.get(qid)
if not item or item.contest_id != contest_id:
continue
q_idx += 1
q = {
'id': q_idx,
'type': item.type,
'content': item.content,
'score': item.score,