442 lines
15 KiB
Python
442 lines
15 KiB
Python
"""打手抢单前考试：按俱乐部独立题库与配置。"""
|
||
import logging
|
||
import os
|
||
import random
|
||
import uuid
|
||
|
||
from django.utils import timezone
|
||
|
||
from jituan.services.admin_context import build_admin_club_context
|
||
from jituan.constants import CLUB_ID_DEFAULT, 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 = '请切换到具体小程序/俱乐部后再操作'
|
||
EXAM_MANAGE_PERMS = frozenset({'dsks666', '8080a'})
|
||
OPTION_FIELDS = [f'option_{i}' for i in range(1, 7)]
|
||
CORRECT_FIELDS = [f'correct_{i}' for i in range(1, 7)]
|
||
|
||
|
||
def has_exam_manage_perm(permissions):
|
||
return any(code in permissions for code in EXAM_MANAGE_PERMS)
|
||
|
||
|
||
def resolve_exam_target_club_id(request, user, data=None):
|
||
"""
|
||
后台考试管理目标俱乐部。
|
||
优先 request 中的 target_club_id，须在用户可见俱乐部列表内。
|
||
集团身份可在页面内切换俱乐部，不依赖顶栏视图。
|
||
"""
|
||
payload = data if data is not None else (getattr(request, 'data', None) or {})
|
||
target = (payload.get('target_club_id') or payload.get('club_id') or '').strip()
|
||
ctx = build_admin_club_context(user)
|
||
visible_ids = {c['club_id'] for c in ctx.get('clubs', [])}
|
||
|
||
if target:
|
||
if target not in visible_ids:
|
||
return None, '无权管理该俱乐部的考试'
|
||
return target, None
|
||
|
||
if ctx.get('scope') == DATA_SCOPE_ALL:
|
||
default = (ctx.get('club_id') or CLUB_ID_DEFAULT).strip()
|
||
if default in visible_ids:
|
||
return default, None
|
||
if visible_ids:
|
||
return sorted(visible_ids)[0], None
|
||
return None, '请选择俱乐部'
|
||
|
||
club_id = resolve_club_id_from_request(request)
|
||
if club_id not in visible_ids:
|
||
return None, '无权管理该俱乐部的考试'
|
||
return club_id, None
|
||
|
||
|
||
def exam_admin_clubs_payload(user):
|
||
ctx = build_admin_club_context(user)
|
||
return {
|
||
'clubs': ctx.get('clubs', []),
|
||
'can_switch_club': bool(ctx.get('can_switch_club')),
|
||
'is_group_admin': bool(ctx.get('is_group_admin')),
|
||
}
|
||
|
||
|
||
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)
|
||
if not cfg.is_enabled:
|
||
return None, '本俱乐部未开启考试'
|
||
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, user, data=None):
|
||
clubs_info = exam_admin_clubs_payload(user)
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return {**clubs_info, 'scope_error': err}
|
||
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 {
|
||
**clubs_info,
|
||
'club_id': club_id,
|
||
'config': config_to_dict(cfg),
|
||
'pool_count': pool,
|
||
'enabled_pool_count': enabled_pool,
|
||
}
|
||
|
||
|
||
def save_admin_config(request, user, data):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
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, user, data=None):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return {**exam_admin_clubs_payload(user), 'list': [], 'scope_error': err}
|
||
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 {**exam_admin_clubs_payload(user), 'club_id': club_id, 'list': out}
|
||
|
||
|
||
def create_question(request, user, data):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
err = validate_question_payload(data)
|
||
if err:
|
||
return None, err
|
||
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, user, question_id, data):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
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, user, question_id, data=None):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
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, user, question_id, file_obj, data=None):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
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, user, image_id, data=None):
|
||
club_id, err = resolve_exam_target_club_id(request, user, data)
|
||
if err:
|
||
return None, err
|
||
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,
|
||
},
|
||
)
|