393 lines
13 KiB
Python
393 lines
13 KiB
Python
"""打手抢单前考试：按俱乐部独立题库与配置。"""
|
|
import logging
|
|
import os
|
|
import random
|
|
import uuid
|
|
|
|
from django.utils import timezone
|
|
|
|
from jituan.constants import DATA_SCOPE_ALL
|
|
from jituan.models import (
|
|
Club,
|
|
ClubDashouExamConfig,
|
|
ClubDashouExamPass,
|
|
ClubDashouExamQuestion,
|
|
ClubDashouExamQuestionImage,
|
|
)
|
|
from jituan.services.club_context import (
|
|
club_id_for_write,
|
|
resolve_club_id_from_request,
|
|
resolve_club_scope,
|
|
resolve_effective_club_id,
|
|
)
|
|
from products.models import Huiyuangoumai
|
|
from utils.oss_utils import delete_from_oss, upload_to_oss, validate_image
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SCOPE_MSG = '请切换到具体小程序/俱乐部后再操作'
|
|
OPTION_FIELDS = [f'option_{i}' for i in range(1, 7)]
|
|
CORRECT_FIELDS = [f'correct_{i}' for i in range(1, 7)]
|
|
|
|
|
|
def _scope_blocked(request):
|
|
return resolve_club_scope(request) == DATA_SCOPE_ALL
|
|
|
|
|
|
def get_or_create_config(club_id):
|
|
row, _ = ClubDashouExamConfig.query.get_or_create(
|
|
club_id=club_id,
|
|
defaults={
|
|
'is_enabled': False,
|
|
'draw_count': 10,
|
|
'pass_count': 10,
|
|
'require_member': True,
|
|
},
|
|
)
|
|
return row
|
|
|
|
|
|
def config_to_dict(row):
|
|
return {
|
|
'club_id': row.club_id,
|
|
'is_enabled': bool(row.is_enabled),
|
|
'draw_count': int(row.draw_count or 10),
|
|
'pass_count': int(row.pass_count or 10),
|
|
'require_member': bool(row.require_member),
|
|
'max_wrong': max(0, int(row.draw_count or 0) - int(row.pass_count or 0)),
|
|
}
|
|
|
|
|
|
def user_has_active_member(yonghuid, club_id):
|
|
for r in Huiyuangoumai.query.filter(yonghu_id=yonghuid, club_id=club_id):
|
|
if r.huiyuan_zhuangtai == 1 and not r.jiance_shifou_daoqi():
|
|
return True
|
|
return False
|
|
|
|
|
|
def user_exam_passed(club_id, yonghuid):
|
|
return ClubDashouExamPass.query.filter(club_id=club_id, yonghuid=yonghuid).exists()
|
|
|
|
|
|
def validate_question_payload(data, for_update=False):
|
|
stem = (data.get('stem') or '').strip()
|
|
if not stem:
|
|
return '题干不能为空'
|
|
qtype = int(data.get('question_type') or 1)
|
|
if qtype not in (1, 2):
|
|
return '题目类型无效'
|
|
options = []
|
|
correct_slots = []
|
|
for i in range(1, 7):
|
|
text = (data.get(f'option_{i}') or '').strip()
|
|
is_corr = bool(data.get(f'correct_{i}'))
|
|
if text:
|
|
options.append(i)
|
|
if is_corr:
|
|
correct_slots.append(i)
|
|
if len(options) < 2:
|
|
return '至少需要两个有效选项'
|
|
if not correct_slots:
|
|
return '至少标记一个正确答案'
|
|
if qtype == 1 and len(correct_slots) != 1:
|
|
return '单选题只能有一个正确答案'
|
|
return None
|
|
|
|
|
|
def question_to_admin_dict(q, images=None):
|
|
item = {
|
|
'id': q.id,
|
|
'stem': q.stem,
|
|
'question_type': q.question_type,
|
|
'explanation': q.explanation or '',
|
|
'is_enabled': bool(q.is_enabled),
|
|
'sort_order': q.sort_order,
|
|
'images': images or [],
|
|
}
|
|
for i in range(1, 7):
|
|
item[f'option_{i}'] = getattr(q, f'option_{i}', '') or ''
|
|
item[f'correct_{i}'] = bool(getattr(q, f'correct_{i}', False))
|
|
return item
|
|
|
|
|
|
def build_question_bundle_item(q, image_urls):
|
|
options = []
|
|
correct_slots = []
|
|
for i in range(1, 7):
|
|
text = (getattr(q, f'option_{i}', '') or '').strip()
|
|
if not text:
|
|
continue
|
|
options.append({'slot': i, 'text': text})
|
|
if getattr(q, f'correct_{i}', False):
|
|
correct_slots.append(i)
|
|
return {
|
|
'id': q.id,
|
|
'stem': q.stem,
|
|
'question_type': q.question_type,
|
|
'images': image_urls,
|
|
'options': options,
|
|
'correct_slots': correct_slots,
|
|
'explanation': q.explanation or '',
|
|
}
|
|
|
|
|
|
def get_exam_status(request, user):
|
|
club_id = resolve_effective_club_id(request, user)
|
|
cfg = get_or_create_config(club_id)
|
|
yonghuid = user.UserUID
|
|
has_member = user_has_active_member(yonghuid, club_id)
|
|
passed = user_exam_passed(club_id, yonghuid)
|
|
enabled = bool(cfg.is_enabled)
|
|
pool_count = ClubDashouExamQuestion.query.filter(club_id=club_id, is_enabled=True).count()
|
|
return {
|
|
'club_id': club_id,
|
|
'exam_enabled': enabled,
|
|
'exam_passed': passed,
|
|
'has_member': has_member,
|
|
'can_take_exam': has_member if cfg.require_member else True,
|
|
'exam_required': enabled and not passed,
|
|
'pool_count': pool_count,
|
|
'config': config_to_dict(cfg),
|
|
}
|
|
|
|
|
|
def build_exam_bundle(request, user):
|
|
club_id = resolve_effective_club_id(request, user)
|
|
cfg = get_or_create_config(club_id)
|
|
yonghuid = user.UserUID
|
|
has_member = user_has_active_member(yonghuid, club_id)
|
|
if cfg.require_member and not has_member:
|
|
return None, '须先开通会员才能参加考试'
|
|
|
|
qs = ClubDashouExamQuestion.query.filter(club_id=club_id, is_enabled=True)
|
|
all_ids = list(qs.values_list('id', flat=True))
|
|
draw = min(int(cfg.draw_count or 10), len(all_ids))
|
|
if draw < 1:
|
|
return None, '题库暂无可用题目，请联系管理员'
|
|
|
|
selected_ids = random.sample(all_ids, draw)
|
|
questions = []
|
|
for qid in selected_ids:
|
|
q = ClubDashouExamQuestion.query.get(id=qid)
|
|
imgs = [
|
|
r.image_url for r in ClubDashouExamQuestionImage.query.filter(
|
|
club_id=club_id, question_id=qid,
|
|
).order_by('sort_order', 'id')
|
|
]
|
|
questions.append(build_question_bundle_item(q, imgs))
|
|
|
|
return {
|
|
'club_id': club_id,
|
|
'config': config_to_dict(cfg),
|
|
'has_member': has_member,
|
|
'already_passed': user_exam_passed(club_id, yonghuid),
|
|
'questions': questions,
|
|
}, None
|
|
|
|
|
|
def mark_exam_passed(request, user):
|
|
club_id = resolve_effective_club_id(request, user)
|
|
cfg = get_or_create_config(club_id)
|
|
if not cfg.is_enabled:
|
|
return None, '本俱乐部未开启考试'
|
|
yonghuid = user.UserUID
|
|
if cfg.require_member and not user_has_active_member(yonghuid, club_id):
|
|
return None, '须先开通会员'
|
|
ClubDashouExamPass.query.update_or_create(
|
|
club_id=club_id,
|
|
yonghuid=yonghuid,
|
|
defaults={'passed_at': timezone.now()},
|
|
)
|
|
return {'club_id': club_id, 'passed': True}, None
|
|
|
|
|
|
def build_group_exam_overview(request):
|
|
"""集团汇总视图：各俱乐部考试配置一览。"""
|
|
if not _scope_blocked(request):
|
|
return None, '请在集团汇总视图查看各俱乐部概览'
|
|
rows = []
|
|
for club in Club.query.filter(status=1).order_by('club_id'):
|
|
cid = club.club_id
|
|
cfg = get_or_create_config(cid)
|
|
pool = ClubDashouExamQuestion.query.filter(club_id=cid).count()
|
|
enabled_pool = ClubDashouExamQuestion.query.filter(club_id=cid, is_enabled=True).count()
|
|
passed_count = ClubDashouExamPass.query.filter(club_id=cid).count()
|
|
rows.append({
|
|
'club_id': cid,
|
|
'club_name': club.name or cid,
|
|
'is_enabled': bool(cfg.is_enabled),
|
|
'draw_count': int(cfg.draw_count or 10),
|
|
'pass_count': int(cfg.pass_count or 10),
|
|
'require_member': bool(cfg.require_member),
|
|
'pool_count': pool,
|
|
'enabled_pool_count': enabled_pool,
|
|
'passed_user_count': passed_count,
|
|
})
|
|
return {'list': rows}, None
|
|
|
|
|
|
def build_admin_config_payload(request):
|
|
if _scope_blocked(request):
|
|
return {'scope_error': SCOPE_MSG}
|
|
club_id = resolve_club_id_from_request(request)
|
|
cfg = get_or_create_config(club_id)
|
|
pool = ClubDashouExamQuestion.query.filter(club_id=club_id).count()
|
|
enabled_pool = ClubDashouExamQuestion.query.filter(club_id=club_id, is_enabled=True).count()
|
|
return {
|
|
'club_id': club_id,
|
|
'config': config_to_dict(cfg),
|
|
'pool_count': pool,
|
|
'enabled_pool_count': enabled_pool,
|
|
}
|
|
|
|
|
|
def save_admin_config(request, data):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
club_id = club_id_for_write(request)
|
|
draw = int(data.get('draw_count') or 10)
|
|
pass_cnt = int(data.get('pass_count') or draw)
|
|
if draw < 1:
|
|
return None, '抽题数量至少为1'
|
|
if pass_cnt < 1 or pass_cnt > draw:
|
|
return None, '及格题数须在 1 到抽题数量之间'
|
|
cfg = get_or_create_config(club_id)
|
|
cfg.is_enabled = bool(data.get('is_enabled'))
|
|
cfg.draw_count = draw
|
|
cfg.pass_count = pass_cnt
|
|
cfg.require_member = bool(data.get('require_member', True))
|
|
cfg.save(update_fields=['is_enabled', 'draw_count', 'pass_count', 'require_member', 'UpdateTime'])
|
|
return config_to_dict(cfg), None
|
|
|
|
|
|
def list_admin_questions(request):
|
|
if _scope_blocked(request):
|
|
return {'list': [], 'scope_error': SCOPE_MSG}
|
|
club_id = resolve_club_id_from_request(request)
|
|
rows = ClubDashouExamQuestion.query.filter(club_id=club_id).order_by('-sort_order', 'id')
|
|
out = []
|
|
for q in rows:
|
|
imgs = [
|
|
{'id': img.id, 'image_url': img.image_url, 'sort_order': img.sort_order}
|
|
for img in ClubDashouExamQuestionImage.query.filter(
|
|
club_id=club_id, question_id=q.id,
|
|
).order_by('sort_order', 'id')
|
|
]
|
|
out.append(question_to_admin_dict(q, imgs))
|
|
return {'club_id': club_id, 'list': out}
|
|
|
|
|
|
def create_question(request, data):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
err = validate_question_payload(data)
|
|
if err:
|
|
return None, err
|
|
club_id = club_id_for_write(request)
|
|
q = ClubDashouExamQuestion.query.create(
|
|
club_id=club_id,
|
|
stem=data.get('stem', '').strip(),
|
|
question_type=int(data.get('question_type') or 1),
|
|
explanation=(data.get('explanation') or '').strip(),
|
|
is_enabled=bool(data.get('is_enabled', True)),
|
|
sort_order=int(data.get('sort_order') or 0),
|
|
**{f'option_{i}': (data.get(f'option_{i}') or '').strip() for i in range(1, 7)},
|
|
**{f'correct_{i}': bool(data.get(f'correct_{i}')) for i in range(1, 7)},
|
|
)
|
|
return {'id': q.id}, None
|
|
|
|
|
|
def update_question(request, question_id, data):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
club_id = club_id_for_write(request)
|
|
try:
|
|
q = ClubDashouExamQuestion.query.get(id=question_id, club_id=club_id)
|
|
except ClubDashouExamQuestion.DoesNotExist:
|
|
return None, '题目不存在'
|
|
err = validate_question_payload(data, for_update=True)
|
|
if err:
|
|
return None, err
|
|
q.stem = data.get('stem', q.stem).strip()
|
|
q.question_type = int(data.get('question_type') or q.question_type)
|
|
q.explanation = (data.get('explanation') or '').strip()
|
|
if 'is_enabled' in data:
|
|
q.is_enabled = bool(data.get('is_enabled'))
|
|
if 'sort_order' in data:
|
|
q.sort_order = int(data.get('sort_order') or 0)
|
|
for i in range(1, 7):
|
|
if f'option_{i}' in data:
|
|
q.__setattr__(f'option_{i}', (data.get(f'option_{i}') or '').strip())
|
|
if f'correct_{i}' in data:
|
|
q.__setattr__(f'correct_{i}', bool(data.get(f'correct_{i}')))
|
|
q.save()
|
|
return {'id': q.id}, None
|
|
|
|
|
|
def delete_question(request, question_id):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
club_id = club_id_for_write(request)
|
|
try:
|
|
q = ClubDashouExamQuestion.query.get(id=question_id, club_id=club_id)
|
|
except ClubDashouExamQuestion.DoesNotExist:
|
|
return None, '题目不存在'
|
|
for img in ClubDashouExamQuestionImage.query.filter(club_id=club_id, question_id=q.id):
|
|
if img.image_url:
|
|
delete_from_oss(img.image_url)
|
|
ClubDashouExamQuestionImage.query.filter(club_id=club_id, question_id=q.id).delete()
|
|
q.delete()
|
|
return {'id': question_id}, None
|
|
|
|
|
|
def upload_question_image(request, question_id, file_obj):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
club_id = club_id_for_write(request)
|
|
if not ClubDashouExamQuestion.query.filter(id=question_id, club_id=club_id).exists():
|
|
return None, '题目不存在'
|
|
ok, msg = validate_image(file_obj)
|
|
if not ok:
|
|
return None, msg
|
|
ext = os.path.splitext(file_obj.name)[1].lower() or '.png'
|
|
if ext not in ('.png', '.jpg', '.jpeg', '.webp'):
|
|
ext = '.png'
|
|
oss_path = f'club/{club_id}/exam/{question_id}_{uuid.uuid4().hex}{ext}'
|
|
if not upload_to_oss(file_obj, oss_path):
|
|
return None, '图片上传失败'
|
|
row = ClubDashouExamQuestionImage.query.create(
|
|
club_id=club_id,
|
|
question_id=question_id,
|
|
image_url=oss_path,
|
|
sort_order=int(request.data.get('sort_order') or 0),
|
|
)
|
|
return {'id': row.id, 'image_url': oss_path}, None
|
|
|
|
|
|
def delete_question_image(request, image_id):
|
|
if _scope_blocked(request):
|
|
return None, SCOPE_MSG
|
|
club_id = club_id_for_write(request)
|
|
try:
|
|
img = ClubDashouExamQuestionImage.query.get(id=image_id, club_id=club_id)
|
|
except ClubDashouExamQuestionImage.DoesNotExist:
|
|
return None, '图片不存在'
|
|
if img.image_url:
|
|
delete_from_oss(img.image_url)
|
|
img.delete()
|
|
return {'id': image_id}, None
|
|
|
|
|
|
def copy_exam_config(template_club_id, new_club_id):
|
|
tpl = ClubDashouExamConfig.query.filter(club_id=template_club_id).first()
|
|
if tpl:
|
|
ClubDashouExamConfig.query.update_or_create(
|
|
club_id=new_club_id,
|
|
defaults={
|
|
'is_enabled': tpl.is_enabled,
|
|
'draw_count': tpl.draw_count,
|
|
'pass_count': tpl.pass_count,
|
|
'require_member': tpl.require_member,
|
|
},
|
|
)
|