XingQue/Django
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 考试权限dsks666、页面内切俱乐部、商品板块仅集团可改

Browse Source
This commit is contained in:
XingQue
2026-06-26 06:07:52 +08:00
parent 34e79aa753
commit 85531da9eb
5 changed files with 125 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/view.py
View File

@@ -6872,6 +6872,9 @@ class HqbkxxView(APIView):
# 4. 检查是否有板块管理权限
if 'bankuai' not in permissions:
return Response({'code': 403, 'msg': '无板块管理权限'}, status=403)
scope_err = block_non_group_catalog_scope(request)
if scope_err:
return Response({'code': 400, 'msg': scope_err})
# 5. 查询所有板块
bankuai_list = []
@@ -6963,6 +6966,9 @@ class BkxgView(APIView):
# 4. 检查板块管理权限
if 'bankuai' not in permissions:
return Response({'code': 403, 'msg': '无板块管理权限'}, status=403)
scope_err = block_non_group_catalog_scope(request)
if scope_err:
return Response({'code': 400, 'msg': scope_err})
# 5. 获取操作类型
action = request.data.get('action')

6
jituan/management/commands/seed_gvsdsdk_permissions.py
View File

@@ -8,6 +8,12 @@ from gvsdsdk.models import Permission
# (PermCode, PermName, PermDesc, PermCategory)
EXTRA_PERMISSIONS = [
(
'dsks666',
'打手接单考试管理',
'配置各俱乐部考试开关、抽题规则与题库（小程序配置-打手接单考试）',
'小程序配置',
),
(
'sfbq666',
'身份装饰标签管理',

2
jituan/management/commands/seed_kefu_menu.py
View File

@@ -103,7 +103,7 @@ MENU_ROWS = [
{'page_id': 'miniapp.kaohe-tags', 'name': '抢单考核标签', 'path': '/miniapp/kaohe-tags', 'parent_id': 'miniapp', 'sort_order': 116,
'perm_codes': ['8080a']},
{'page_id': 'miniapp.dashou-exam', 'name': '打手接单考试', 'path': '/miniapp/dashou-exam', 'parent_id': 'miniapp', 'sort_order': 117,
'perm_codes': ['8080a']},
'perm_codes': ['dsks666', '8080a']},
# 店铺
{'page_id': 'shop', 'name': '店铺管理', 'path': '', 'parent_id': '', 'sort_order': 120},
{'page_id': 'shop.list', 'name': '店铺列表', 'path': '/shop/list', 'parent_id': 'shop', 'sort_order': 121,

115
jituan/services/dashou_exam.py
View File

@@ -6,7 +6,8 @@ import uuid
from django.utils import timezone
from jituan.constants import DATA_SCOPE_ALL
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,
@@ -26,10 +27,54 @@ 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
@@ -226,14 +271,16 @@ def build_group_exam_overview(request):
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)
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,
@@ -241,10 +288,10 @@ def build_admin_config_payload(request):
}
def save_admin_config(request, data):
if _scope_blocked(request):
return None, SCOPE_MSG
club_id = club_id_for_write(request)
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:
@@ -260,10 +307,10 @@ def save_admin_config(request, data):
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)
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:
@@ -274,16 +321,16 @@ def list_admin_questions(request):
).order_by('sort_order', 'id')
]
out.append(question_to_admin_dict(q, imgs))
return {'club_id': club_id, 'list': out}
return {**exam_admin_clubs_payload(user), 'club_id': club_id, 'list': out}
def create_question(request, data):
if _scope_blocked(request):
return None, SCOPE_MSG
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
club_id = club_id_for_write(request)
q = ClubDashouExamQuestion.query.create(
club_id=club_id,
stem=data.get('stem', '').strip(),
@@ -297,10 +344,10 @@ def create_question(request, data):
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)
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:
@@ -324,10 +371,10 @@ def update_question(request, question_id, data):
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)
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:
@@ -340,10 +387,10 @@ def delete_question(request, question_id):
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)
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)
@@ -364,10 +411,10 @@ def upload_question_image(request, question_id, file_obj):
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)
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:

60
jituan/views_dashou_exam.py
View File

@@ -5,8 +5,18 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from backend.utils import verify_kefu_permission
from jituan.services.dashou_exam import has_exam_manage_perm
EXAM_PERM = '8080a'
EXAM_PERM_HINT = '无权限（需 dsks666 打手接单考试管理，或 8080a 小程序配置）'
def _check_exam_perm(request, phone):
kefu, permissions = verify_kefu_permission(request, phone)
if kefu is None:
return None, permissions
if not has_exam_manage_perm(permissions):
return None, Response({'code': 403, 'msg': EXAM_PERM_HINT})
return request.user, None
class DashouExamStatusView(APIView):
@@ -55,11 +65,9 @@ class DashouExamConfigAdminView(APIView):
phone = (request.data.get('phone') or request.data.get('username') or '').strip()
if not phone:
return Response({'code': 401, 'msg': '缺少客服账号'})
kefu, permissions = verify_kefu_permission(request, phone)
if kefu is None:
return permissions
if EXAM_PERM not in permissions:
return Response({'code': 403, 'msg': '无权限'})
user, err_resp = _check_exam_perm(request, phone)
if err_resp:
return err_resp
action = (request.data.get('action') or 'get').strip()
from jituan.services.dashou_exam import (
build_admin_config_payload,
@@ -72,11 +80,11 @@ class DashouExamConfigAdminView(APIView):
return Response({'code': 400, 'msg': err})
return Response({'code': 0, 'data': data})
if action == 'save':
data, err = save_admin_config(request, request.data)
data, err = save_admin_config(request, user, request.data)
if err:
return Response({'code': 400, 'msg': err})
return Response({'code': 0, 'msg': '保存成功', 'data': {'config': data}})
payload = build_admin_config_payload(request)
payload = build_admin_config_payload(request, user, request.data)
return Response({'code': 0, 'data': payload})
@@ -89,13 +97,11 @@ class DashouExamQuestionListView(APIView):
phone = (request.data.get('phone') or request.data.get('username') or '').strip()
if not phone:
return Response({'code': 401, 'msg': '缺少客服账号'})
kefu, permissions = verify_kefu_permission(request, phone)
if kefu is None:
return permissions
if EXAM_PERM not in permissions:
return Response({'code': 403, 'msg': '无权限'})
user, err_resp = _check_exam_perm(request, phone)
if err_resp:
return err_resp
from jituan.services.dashou_exam import list_admin_questions
return Response({'code': 0, 'data': list_admin_questions(request)})
return Response({'code': 0, 'data': list_admin_questions(request, user, request.data)})
class DashouExamQuestionManageView(APIView):
@@ -107,11 +113,9 @@ class DashouExamQuestionManageView(APIView):
phone = (request.data.get('phone') or request.data.get('username') or '').strip()
if not phone:
return Response({'code': 401, 'msg': '缺少客服账号'})
kefu, permissions = verify_kefu_permission(request, phone)
if kefu is None:
return permissions
if EXAM_PERM not in permissions:
return Response({'code': 403, 'msg': '无权限'})
user, err_resp = _check_exam_perm(request, phone)
if err_resp:
return err_resp
action = (request.data.get('action') or '').strip()
qid = request.data.get('question_id') or request.data.get('id')
from jituan.services.dashou_exam import (
@@ -120,15 +124,15 @@ class DashouExamQuestionManageView(APIView):
update_question,
)
if action == 'create':
data, err = create_question(request, request.data)
data, err = create_question(request, user, request.data)
elif action == 'update':
if not qid:
return Response({'code': 400, 'msg': '缺少题目ID'})
data, err = update_question(request, int(qid), request.data)
data, err = update_question(request, user, int(qid), request.data)
elif action == 'delete':
if not qid:
return Response({'code': 400, 'msg': '缺少题目ID'})
data, err = delete_question(request, int(qid))
data, err = delete_question(request, user, int(qid), request.data)
else:
return Response({'code': 400, 'msg': '无效 action'})
if err:
@@ -145,24 +149,22 @@ class DashouExamImageManageView(APIView):
phone = (request.data.get('phone') or request.data.get('username') or '').strip()
if not phone:
return Response({'code': 401, 'msg': '缺少客服账号'})
kefu, permissions = verify_kefu_permission(request, phone)
if kefu is None:
return permissions
if EXAM_PERM not in permissions:
return Response({'code': 403, 'msg': '无权限'})
user, err_resp = _check_exam_perm(request, phone)
if err_resp:
return err_resp
action = (request.data.get('action') or 'upload').strip()
from jituan.services.dashou_exam import delete_question_image, upload_question_image
if action == 'delete':
image_id = request.data.get('image_id')
if not image_id:
return Response({'code': 400, 'msg': '缺少 image_id'})
data, err = delete_question_image(request, int(image_id))
data, err = delete_question_image(request, user, int(image_id), request.data)
else:
qid = request.data.get('question_id')
file_obj = request.FILES.get('file')
if not qid or not file_obj:
return Response({'code': 400, 'msg': '缺少题目或图片'})
data, err = upload_question_image(request, int(qid), file_obj)
data, err = upload_question_image(request, user, int(qid), file_obj, request.data)
if err:
return Response({'code': 400, 'msg': err})
return Response({'code': 0, 'msg': '成功', 'data': data})