diff --git a/apps/case/mobile.py b/apps/case/mobile.py index f04a675..90197ee 100644 --- a/apps/case/mobile.py +++ b/apps/case/mobile.py @@ -1,233 +1,233 @@ -"""移动端病例列表（首页 / 病例页 5 个入口）。 - -均读 `case_base`，只取**已发布**病例（publish_status=2 & status=1 & is_deleted=0）： - -- 5.1 推荐病例（个性化） GET /api/case/mobile/recommended/ -- 5.2 科室专项 GET /api/case/mobile/specialty/ -- 5.3 薄弱环节 GET /api/case/mobile/weak/ -- 5.4 教学互动 GET /api/case/mobile/teaching/ -- 5.5 教师任务 GET /api/case/mobile/teacher-task/（CMS 暂无指派，先同教学互动） - -个性化/薄弱口径读只读表 training_record（user_id=当前用户、status='completed'）， -分数按 score_type 归一百分制后比较，与《个人中心》一致。 -""" -from django.db.models import Q -from rest_framework import serializers -from rest_framework.decorators import api_view, permission_classes -from rest_framework.permissions import IsAuthenticated -from rest_framework.pagination import PageNumberPagination -from rest_framework.response import Response -from drf_spectacular.utils import extend_schema - -from apps.training.models import TrainingRecord -from .models import CaseBase - -# 薄弱阈值：归一百分制下，最好成绩低于该分数视为「薄弱病例」 -WEAK_SCORE_THRESHOLD = 70 - - -class MobilePagination(PageNumberPagination): - page_size = 20 - page_size_query_param = 'page_size' - max_page_size = 100 - - -class MobileCaseSerializer(serializers.ModelSerializer): - """移动端病例列表元素（统一字段）。""" - case_type_display = serializers.CharField(source='get_case_type_display', read_only=True) - department_name = serializers.CharField(source='department.name', read_only=True) - # 仅「薄弱环节」会注入；其余接口为 None - my_best_score = serializers.SerializerMethodField() - my_train_count = serializers.SerializerMethodField() - - class Meta: - model = CaseBase - fields = [ - 'id', 'title', 'case_type', 'case_type_display', 'difficulty', - 'difficulty_score', 'department', 'department_name', - 'chief_complaint', 'description', 'patient_age', 'patient_gender', - 'tags', 'competency_tags', 'estimated_minutes', 'osce_enabled', - 'created_at', 'my_best_score', 'my_train_count', - ] - - def _stat(self, obj): - return (self.context.get('case_stats') or {}).get(obj.id) - - def get_my_best_score(self, obj): - s = self._stat(obj) - return s['best_score'] if s else None - - def get_my_train_count(self, obj): - s = self._stat(obj) - return s['train_count'] if s else None - - -# ── 公共工具 ────────────────────────────────────────────────────────────── - -def _published(): - """已发布病例基集（已过滤 is_deleted）。""" - return CaseBase.objects.filter(publish_status=2, status=1) - - -def _apply_common_filters(qs, request, *, allow_department=True): - """通用查询参数：search / case_type / difficulty / department。""" - p = request.query_params - search = (p.get('search') or '').strip() - if search: - qs = qs.filter( - Q(title__icontains=search) - | Q(chief_complaint__icontains=search) - | Q(tags__icontains=search) - ) - if p.get('case_type'): - qs = qs.filter(case_type=p['case_type']) - if p.get('difficulty'): - qs = qs.filter(difficulty=p['difficulty']) - if allow_department and p.get('department'): - qs = qs.filter(department_id=p['department']) - return qs - - -def _completed_records(user): - """当前用户已完成训练记录（只读表，不 JOIN，仅取 case_id/分数）。""" - return TrainingRecord.objects.filter(user_id=user.id, status='completed') - - -def _norm_score(score, score_type): - if score is None: - return None - return float(score) * 20 if score_type == 'five_point' else float(score) - - -def _user_case_stats(user): - """{case_id: {best_score, train_count}}（归一百分制）。""" - stats = {} - rows = _completed_records(user).values('case_id', 'total_score', 'score_type') - for r in rows: - cid = r['case_id'] - if cid is None: - continue - s = _norm_score(r['total_score'], r['score_type']) - cur = stats.setdefault(cid, {'best_score': None, 'train_count': 0}) - cur['train_count'] += 1 - if s is not None and (cur['best_score'] is None or s > cur['best_score']): - cur['best_score'] = round(s, 1) - return stats - - -def _paginate(qs, request, *, case_stats=None): - paginator = MobilePagination() - page = paginator.paginate_queryset(qs, request) - ctx = {'request': request, 'case_stats': case_stats or {}} - data = MobileCaseSerializer(page, many=True, context=ctx).data - return paginator.get_paginated_response(data) - - -# ── 5.1 推荐病例（个性化）──────────────────────────────────────────────────── - -def _recommend_score(case, *, dept_id, trained_ids, weak_dims): - """个性化推荐分：未训练 > 同科室 > 命中薄弱能力标签。""" - score = 0 - if case.id not in trained_ids: - score += 2 - if dept_id and case.department_id == dept_id: - score += 1 - if weak_dims and case.competency_tags: - tags = ' '.join(str(t) for t in case.competency_tags) - if any(w and w in tags for w in weak_dims): - score += 1 - return score - - -@extend_schema(summary='推荐病例（个性化）', tags=['病例列表']) -@api_view(['GET']) -@permission_classes([IsAuthenticated]) -def recommended(request): - user = request.user - qs = _apply_common_filters(_published(), request) - cases = list(qs.select_related('department')) - - trained_ids = set( - _completed_records(user).values_list('case_id', flat=True) - ) - weak_dims = [str(w) for w in (user.weak_dimensions or [])] - dept_id = user.department_id - - cases.sort( - key=lambda c: ( - _recommend_score(c, dept_id=dept_id, trained_ids=trained_ids, weak_dims=weak_dims), - c.created_at or c.id, - ), - reverse=True, - ) - return _paginate(cases, request) - - -# ── 5.2 科室专项 ──────────────────────────────────────────────────────────── - -@extend_schema(summary='科室专项病例', tags=['病例列表']) -@api_view(['GET']) -@permission_classes([IsAuthenticated]) -def specialty(request): - qs = _apply_common_filters(_published(), request, allow_department=False) - dept = request.query_params.get('department') or request.user.department_id - if dept: - qs = qs.filter(department_id=dept) - qs = qs.select_related('department').order_by('-created_at', '-id') - return _paginate(qs, request) - - -# ── 5.3 薄弱环节 ──────────────────────────────────────────────────────────── - -@extend_schema(summary='薄弱环节病例', tags=['病例列表']) -@api_view(['GET']) -@permission_classes([IsAuthenticated]) -def weak(request): - user = request.user - stats = _user_case_stats(user) - weak_ids = [cid for cid, s in stats.items() - if s['best_score'] is not None and s['best_score'] < WEAK_SCORE_THRESHOLD] - - base = _apply_common_filters(_published(), request).select_related('department') - if weak_ids: - cases = list(base.filter(id__in=weak_ids)) - # worst-first：最好成绩越低越靠前 - cases.sort(key=lambda c: stats[c.id]['best_score']) - return _paginate(cases, request, case_stats=stats) - - # 冷启动：无低分记录 → 回退到命中用户薄弱能力标签的已发布病例（competency_tags 为 JSON，按 Python 匹配） - weak_dims = [str(w) for w in (user.weak_dimensions or []) if w] - cases = [] - if weak_dims: - for c in base.order_by('-created_at', '-id'): - tags = ' '.join(str(t) for t in (c.competency_tags or [])) - if any(w in tags for w in weak_dims): - cases.append(c) - return _paginate(cases, request, case_stats=stats) - - -# ── 5.4 教学互动 ──────────────────────────────────────────────────────────── - -@extend_schema(summary='教学互动病例', tags=['病例列表']) -@api_view(['GET']) -@permission_classes([IsAuthenticated]) -def teaching(request): - qs = _apply_common_filters( - _published().filter(case_type='teaching'), request - ).select_related('department').order_by('-created_at', '-id') - return _paginate(qs, request) - - -# ── 5.5 教师任务（暂同教学互动）────────────────────────────────────────────── - -@extend_schema( - summary='教师任务病例', tags=['病例列表'], - description='CMS 暂无「指派病例」功能，先与教学互动一致（已发布 case_type=teaching）。', -) -@api_view(['GET']) -@permission_classes([IsAuthenticated]) -def teacher_task(request): - qs = _apply_common_filters( - _published().filter(case_type='teaching'), request - ).select_related('department').order_by('-created_at', '-id') - return _paginate(qs, request) +"""移动端病例列表（首页 / 病例页 5 个入口）。 + +均读 `case_base`，只取**已发布**病例（publish_status=2 & status=1 & is_deleted=0）： + +- 5.1 推荐病例（个性化） GET /api/case/mobile/recommended/ +- 5.2 科室专项 GET /api/case/mobile/specialty/ +- 5.3 薄弱环节 GET /api/case/mobile/weak/ +- 5.4 教学互动 GET /api/case/mobile/teaching/ +- 5.5 教师任务 GET /api/case/mobile/teacher-task/（CMS 暂无指派，先同教学互动） + +个性化/薄弱口径读只读表 training_record（user_id=当前用户、status='completed'）， +分数按 score_type 归一百分制后比较，与《个人中心》一致。 +""" +from django.db.models import Q +from rest_framework import serializers +from rest_framework.decorators import api_view, permission_classes +from rest_framework.permissions import IsAuthenticated +from rest_framework.pagination import PageNumberPagination +from rest_framework.response import Response +from drf_spectacular.utils import extend_schema + +from apps.training.models import TrainingRecord +from .models import CaseBase + +# 薄弱阈值：归一百分制下，最好成绩低于该分数视为「薄弱病例」 +WEAK_SCORE_THRESHOLD = 70 + + +class MobilePagination(PageNumberPagination): + page_size = 20 + page_size_query_param = 'page_size' + max_page_size = 100 + + +class MobileCaseSerializer(serializers.ModelSerializer): + """移动端病例列表元素（统一字段）。""" + case_type_display = serializers.CharField(source='get_case_type_display', read_only=True) + department_name = serializers.CharField(source='department.name', read_only=True) + # 仅「薄弱环节」会注入；其余接口为 None + my_best_score = serializers.SerializerMethodField() + my_train_count = serializers.SerializerMethodField() + + class Meta: + model = CaseBase + fields = [ + 'id', 'title', 'case_type', 'case_type_display', 'difficulty', + 'difficulty_score', 'department', 'department_name', + 'chief_complaint', 'description', 'patient_age', 'patient_gender', + 'tags', 'competency_tags', 'estimated_minutes', 'osce_enabled', + 'created_at', 'my_best_score', 'my_train_count', + ] + + def _stat(self, obj): + return (self.context.get('case_stats') or {}).get(obj.id) + + def get_my_best_score(self, obj): + s = self._stat(obj) + return s['best_score'] if s else None + + def get_my_train_count(self, obj): + s = self._stat(obj) + return s['train_count'] if s else None + + +# ── 公共工具 ────────────────────────────────────────────────────────────── + +def _published(): + """已发布病例基集（已过滤 is_deleted）。""" + return CaseBase.objects.filter(publish_status=2, status=1) + + +def _apply_common_filters(qs, request, *, allow_department=True): + """通用查询参数：search / case_type / difficulty / department。""" + p = request.query_params + search = (p.get('search') or '').strip() + if search: + qs = qs.filter( + Q(title__icontains=search) + | Q(chief_complaint__icontains=search) + | Q(tags__icontains=search) + ) + if p.get('case_type'): + qs = qs.filter(case_type=p['case_type']) + if p.get('difficulty'): + qs = qs.filter(difficulty=p['difficulty']) + if allow_department and p.get('department'): + qs = qs.filter(department_id=p['department']) + return qs + + +def _completed_records(user): + """当前用户已完成训练记录（只读表，不 JOIN，仅取 case_id/分数）。""" + return TrainingRecord.objects.filter(user_id=user.id, status='completed') + + +def _norm_score(score, score_type): + if score is None: + return None + return float(score) * 20 if score_type == 'five_point' else float(score) + + +def _user_case_stats(user): + """{case_id: {best_score, train_count}}（归一百分制）。""" + stats = {} + rows = _completed_records(user).values('case_id', 'total_score', 'score_type') + for r in rows: + cid = r['case_id'] + if cid is None: + continue + s = _norm_score(r['total_score'], r['score_type']) + cur = stats.setdefault(cid, {'best_score': None, 'train_count': 0}) + cur['train_count'] += 1 + if s is not None and (cur['best_score'] is None or s > cur['best_score']): + cur['best_score'] = round(s, 1) + return stats + + +def _paginate(qs, request, *, case_stats=None): + paginator = MobilePagination() + page = paginator.paginate_queryset(qs, request) + ctx = {'request': request, 'case_stats': case_stats or {}} + data = MobileCaseSerializer(page, many=True, context=ctx).data + return paginator.get_paginated_response(data) + + +# ── 5.1 推荐病例（个性化）──────────────────────────────────────────────────── + +def _recommend_score(case, *, dept_id, trained_ids, weak_dims): + """个性化推荐分：未训练 > 同科室 > 命中薄弱能力标签。""" + score = 0 + if case.id not in trained_ids: + score += 2 + if dept_id and case.department_id == dept_id: + score += 1 + if weak_dims and case.competency_tags: + tags = ' '.join(str(t) for t in case.competency_tags) + if any(w and w in tags for w in weak_dims): + score += 1 + return score + + +@extend_schema(summary='推荐病例（个性化）', tags=['病例列表']) +@api_view(['GET']) +@permission_classes([IsAuthenticated]) +def recommended(request): + user = request.user + qs = _apply_common_filters(_published(), request) + cases = list(qs.select_related('department')) + + trained_ids = set( + _completed_records(user).values_list('case_id', flat=True) + ) + weak_dims = [str(w) for w in (user.weak_dimensions or [])] + dept_id = user.department_id + + cases.sort( + key=lambda c: ( + _recommend_score(c, dept_id=dept_id, trained_ids=trained_ids, weak_dims=weak_dims), + c.created_at or c.id, + ), + reverse=True, + ) + return _paginate(cases, request) + + +# ── 5.2 科室专项 ──────────────────────────────────────────────────────────── + +@extend_schema(summary='科室专项病例', tags=['病例列表']) +@api_view(['GET']) +@permission_classes([IsAuthenticated]) +def specialty(request): + qs = _apply_common_filters(_published(), request, allow_department=False) + dept = request.query_params.get('department') or request.user.department_id + if dept: + qs = qs.filter(department_id=dept) + qs = qs.select_related('department').order_by('-created_at', '-id') + return _paginate(qs, request) + + +# ── 5.3 薄弱环节 ──────────────────────────────────────────────────────────── + +@extend_schema(summary='薄弱环节病例', tags=['病例列表']) +@api_view(['GET']) +@permission_classes([IsAuthenticated]) +def weak(request): + user = request.user + stats = _user_case_stats(user) + weak_ids = [cid for cid, s in stats.items() + if s['best_score'] is not None and s['best_score'] < WEAK_SCORE_THRESHOLD] + + base = _apply_common_filters(_published(), request).select_related('department') + if weak_ids: + cases = list(base.filter(id__in=weak_ids)) + # worst-first：最好成绩越低越靠前 + cases.sort(key=lambda c: stats[c.id]['best_score']) + return _paginate(cases, request, case_stats=stats) + + # 冷启动：无低分记录 → 回退到命中用户薄弱能力标签的已发布病例（competency_tags 为 JSON，按 Python 匹配） + weak_dims = [str(w) for w in (user.weak_dimensions or []) if w] + cases = [] + if weak_dims: + for c in base.order_by('-created_at', '-id'): + tags = ' '.join(str(t) for t in (c.competency_tags or [])) + if any(w in tags for w in weak_dims): + cases.append(c) + return _paginate(cases, request, case_stats=stats) + + +# ── 5.4 教学互动 ──────────────────────────────────────────────────────────── + +@extend_schema(summary='教学互动病例', tags=['病例列表']) +@api_view(['GET']) +@permission_classes([IsAuthenticated]) +def teaching(request): + qs = _apply_common_filters( + _published().filter(case_type='teaching'), request + ).select_related('department').order_by('-created_at', '-id') + return _paginate(qs, request) + + +# ── 5.5 教师任务（暂同教学互动）────────────────────────────────────────────── + +@extend_schema( + summary='教师任务病例', tags=['病例列表'], + description='CMS 暂无「指派病例」功能，先与教学互动一致（已发布 case_type=teaching）。', +) +@api_view(['GET']) +@permission_classes([IsAuthenticated]) +def teacher_task(request): + qs = _apply_common_filters( + _published().filter(case_type='teaching'), request + ).select_related('department').order_by('-created_at', '-id') + return _paginate(qs, request) diff --git a/apps/training/migrations/0004_trainingsession.py b/apps/training/migrations/0004_trainingsession.py new file mode 100644 index 0000000..311600f --- /dev/null +++ b/apps/training/migrations/0004_trainingsession.py @@ -0,0 +1,34 @@ +# Generated by Django 5.2.14 on 2026-06-13 05:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('training', '0003_alter_trainingrecord_options_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='TrainingSession', + fields=[ + ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')), + ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')), + ('id', models.BigAutoField(primary_key=True, serialize=False)), + ('case_id', models.BigIntegerField(blank=True, null=True, verbose_name='病例ID')), + ('case_type', models.CharField(blank=True, max_length=30, verbose_name='病例类型')), + ('training_mode', models.CharField(blank=True, max_length=50, verbose_name='训练模式')), + ('status', models.CharField(blank=True, max_length=30, verbose_name='状态')), + ('external_user_id', models.CharField(blank=True, max_length=128, verbose_name='宿主系统用户ID')), + ('started_at', models.DateTimeField(blank=True, null=True, verbose_name='开始时间')), + ('completed_at', models.DateTimeField(blank=True, null=True, verbose_name='完成时间')), + ], + options={ + 'verbose_name': '训练会话', + 'verbose_name_plural': '训练会话', + 'db_table': 'training_session', + 'managed': False, + }, + ), + ] diff --git a/apps/user/auth/__init__.py b/apps/user/auth/__init__.py index e7751ff..f868960 100644 --- a/apps/user/auth/__init__.py +++ b/apps/user/auth/__init__.py @@ -2,7 +2,10 @@ from rest_framework_simplejwt.tokens import RefreshToken from config.exceptions import AppError -ALLOWED_ROLE_TYPES = ('student', 'doctor', 'teacher') +# 系统五类角色：super_admin / hospital_admin / content_admin / doctor（带教医生）/ student +ROLE_TYPES = ('super_admin', 'hospital_admin', 'content_admin', 'doctor', 'student') +# 移动端可自注册的角色（带教老师即 doctor，不单列 teacher） +ALLOWED_ROLE_TYPES = ('student', 'doctor') # CMS 端可登录的角色（U3 密码登录）：超级管理员 / 医院管理员 / 内容管理员 / 医生(带教老师) CMS_ROLE_TYPES = ('super_admin', 'hospital_admin', 'content_admin', 'doctor') diff --git a/apps/user/management/commands/init_users.py b/apps/user/management/commands/init_users.py index 2266031..eca8edb 100644 --- a/apps/user/management/commands/init_users.py +++ b/apps/user/management/commands/init_users.py @@ -2,6 +2,8 @@ import os from django.core.management.base import BaseCommand from django.contrib.auth import get_user_model +from apps.user.models import TeacherStudentRelation + User = get_user_model() @@ -24,9 +26,12 @@ class Command(BaseCommand): # 创建超级管理员 self._create_superadmin() - # 创建测试角色用户 + # 创建测试角色用户（覆盖五类角色：医院管理员/内容管理员/带教医生/学生） self._create_test_users() + # 师生关系（只在 doctor 与 student 之间） + self._create_relation() + self.stdout.write(self.style.SUCCESS('\n[完成] 用户初始化完成')) def _create_superadmin(self): @@ -55,6 +60,13 @@ class Command(BaseCommand): def _create_test_users(self): """创建测试用户""" test_users = [ + { + 'username': 'hospital_admin', + 'password': 'hospital123', + 'real_name': '医院管理员', + 'role_type': 'hospital_admin', + 'phone': '13800138003', + }, { 'username': 'doctor1', 'password': 'doctor123', @@ -103,3 +115,16 @@ class Command(BaseCommand): f'[已存在] 用户: {user_data["username"]} ({user_data["real_name"]})' ) ) + + def _create_relation(self): + """初始化一条师生关系（带教医生 doctor1 → 学生 student1）。""" + teacher = User.objects.filter(username='doctor1', role_type='doctor').first() + student = User.objects.filter(username='student1', role_type='student').first() + if not teacher or not student: + return + _, created = TeacherStudentRelation.objects.get_or_create( + teacher=teacher, student=student, + defaults={'relation_type': '指导', 'status': 1}, + ) + msg = '[创建] 师生关系: doctor1 → student1' if created else '[已存在] 师生关系: doctor1 → student1' + self.stdout.write((self.style.SUCCESS if created else self.style.WARNING)(msg)) diff --git a/apps/user/permissions.py b/apps/user/permissions.py index 9364f4a..9e287db 100644 --- a/apps/user/permissions.py +++ b/apps/user/permissions.py @@ -16,7 +16,7 @@ class IsUserListPermitted(BasePermission): user = request.user if _is_admin(user): return True - if user.role_type == 'teacher': + if user.role_type == 'doctor': # 带教医生：可看名下学生 return True raise AppError('USER_NO_LIST_PERMISSION', '您没有查看用户列表的权限', status_code=403) @@ -32,8 +32,8 @@ class IsUserDetailPermitted(BasePermission): # 本人：可查看自己 if user.id == obj.id: return True - # 教师：可查看自己名下活跃学生 - if user.role_type == 'teacher': + # 带教医生：可查看自己名下活跃学生 + if user.role_type == 'doctor': if TeacherStudentRelation.objects.filter( teacher=user, student=obj, status=1 ).exists(): diff --git a/apps/user/serializers.py b/apps/user/serializers.py index 6127152..f3dc3a0 100644 --- a/apps/user/serializers.py +++ b/apps/user/serializers.py @@ -120,6 +120,16 @@ class TeacherStudentRelationSerializer(serializers.ModelSerializer): model = TeacherStudentRelation fields = '__all__' + def validate(self, attrs): + # 师生关系只能在 带教医生(doctor) 与 学生(student) 之间建立 + teacher = attrs.get('teacher') or getattr(self.instance, 'teacher', None) + student = attrs.get('student') or getattr(self.instance, 'student', None) + if teacher is not None and teacher.role_type != 'doctor': + raise serializers.ValidationError({'teacher': '带教方必须是带教医生(doctor)'}) + if student is not None and student.role_type != 'student': + raise serializers.ValidationError({'student': '学生方必须是学生(student)'}) + return attrs + class InstitutionSerializer(serializers.ModelSerializer): class Meta: diff --git a/apps/user/views.py b/apps/user/views.py index e81ac27..14d340e 100644 --- a/apps/user/views.py +++ b/apps/user/views.py @@ -67,8 +67,8 @@ class UserViewSet(viewsets.ModelViewSet): if self.action == 'list': if user.role_type in ('super_admin', 'content_admin') or user.is_staff: return qs # 管理员：全员 - elif user.role_type == 'teacher': - # 教师：仅自己名下活跃学生 + elif user.role_type == 'doctor': + # 带教医生：仅自己名下活跃学生 student_ids = TeacherStudentRelation.objects.filter( teacher=user, status=1 ).values_list('student_id', flat=True) diff --git a/test/swagger_tryout.py b/test/swagger_tryout.py index d2956f5..f88fe31 100644 --- a/test/swagger_tryout.py +++ b/test/swagger_tryout.py @@ -432,7 +432,7 @@ django_eval( f'admin = User.objects.create_user(username="{ADMIN_PHONE}", password="{ROLE_PWD}", ' f' phone="{ADMIN_PHONE}", real_name="Swagger管理员", role_type="super_admin", status=1); ' f'teacher = User.objects.create_user(username="{TEACHER_PHONE}", password="{ROLE_PWD}", ' - f' phone="{TEACHER_PHONE}", real_name="Swagger教师", role_type="teacher", status=1); ' + f' phone="{TEACHER_PHONE}", real_name="Swagger带教医生", role_type="doctor", status=1); ' f'student = User.objects.create_user(username="{STUDENT_PHONE}", password="{ROLE_PWD}", ' f' phone="{STUDENT_PHONE}", real_name="Swagger学生", role_type="student", status=1); ' f'TeacherStudentRelation.objects.create(teacher=teacher, student=student, ' diff --git a/test/test_mobile_case_list.py b/test/test_mobile_case_list.py index d59a032..ffcf7cc 100644 --- a/test/test_mobile_case_list.py +++ b/test/test_mobile_case_list.py @@ -1,179 +1,179 @@ -"""移动端病例列表 5 接口测试。 - -- 5.1 推荐病例 GET /api/case/mobile/recommended/ -- 5.2 科室专项 GET /api/case/mobile/specialty/ -- 5.3 薄弱环节 GET /api/case/mobile/weak/ -- 5.4 教学互动 GET /api/case/mobile/teaching/ -- 5.5 教师任务 GET /api/case/mobile/teacher-task/ - -均只取「已发布」病例（publish_status=2 & status=1 & is_deleted=0）。 -training_record 是 managed=False 只读表，迁移占位结构缺新列，故在 setUpClass 里按真实列 -重建该表（仅测试库），用 TransactionTestCase 允许 DDL（与 test_cms_training 一致）。 -""" -import json - -from django.core.cache import cache -from django.db import connection -from django.test import TransactionTestCase -from django.utils import timezone -from rest_framework.test import APIClient - -from apps.user.models import Department -from apps.case.models import CaseBase -from .conftest import create_test_user, get_auth_client, ensure_institution - -CREATE_TR = """ -CREATE TABLE training_record ( - id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, - user_id BIGINT NULL, case_id BIGINT NULL, teacher_id BIGINT NULL, - training_mode VARCHAR(50) NULL, case_type VARCHAR(30) NULL, - start_time DATETIME NULL, end_time DATETIME NULL, duration_seconds INT NULL, - total_score DECIMAL(5,2) NULL, ai_score DECIMAL(5,2) NULL, teacher_score DECIMAL(5,2) NULL, - evaluation_level VARCHAR(20) NULL, status VARCHAR(30) NULL, - feedback TEXT NULL, thinking_chain TEXT NULL, diagnosis_path TEXT NULL, - wrong_points JSON NULL, missed_questions JSON NULL, recommendation_result JSON NULL, - ai_feedback_structured JSON NULL, osce_station_score JSON NULL, - interruption_count INT NULL, emotion_analysis JSON NULL, - prompt_version VARCHAR(50) NULL, rag_context_version VARCHAR(50) NULL, - external_user_id VARCHAR(128) NULL, session_id BIGINT NULL, evaluation_record_id BIGINT NULL, - score_type VARCHAR(20) NULL, pdf_file_path VARCHAR(512) NULL, - created_at DATETIME NULL, updated_at DATETIME NULL -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 -""" - -REC_URL = '/api/case/mobile/recommended/' -SPEC_URL = '/api/case/mobile/specialty/' -WEAK_URL = '/api/case/mobile/weak/' -TEACH_URL = '/api/case/mobile/teaching/' -TASK_URL = '/api/case/mobile/teacher-task/' - - -class MobileCaseListTest(TransactionTestCase): - - @classmethod - def setUpClass(cls): - super().setUpClass() - with connection.cursor() as c: - c.execute('SET FOREIGN_KEY_CHECKS=0') - c.execute('DROP TABLE IF EXISTS training_record') - c.execute(CREATE_TR) - c.execute('SET FOREIGN_KEY_CHECKS=1') - - def _insert(self, user_id, case_id, total_score, status='completed', score_type='percentage'): - now = timezone.now() - with connection.cursor() as c: - c.execute( - "INSERT INTO training_record (user_id, case_id, training_mode, status, " - "total_score, end_time, start_time, score_type, ai_feedback_structured, " - "created_at, updated_at) VALUES (%s,%s,'practice',%s,%s,%s,%s,%s,%s,%s,%s)", - [user_id, case_id, status, total_score, now, now, score_type, - json.dumps({}), now, now], - ) - return c.lastrowid - - def setUp(self): - cache.clear() - with connection.cursor() as c: - c.execute('DELETE FROM training_record') - self.inst = ensure_institution(name='测试医院', code='MCL-H1') - self.dept1 = Department.objects.create(name='心内科', category='临床') - self.dept2 = Department.objects.create(name='呼吸科', category='临床') - - # 已发布病例 - self.pub_trad1 = CaseBase.objects.create( - title='急性心梗', case_type='traditional', department=self.dept1, - institution=self.inst, chief_complaint='胸痛', tags='心内科,胸痛', - publish_status=2, status=1) - self.pub_teach = CaseBase.objects.create( - title='医患沟通教学', case_type='teaching', department=self.dept1, - institution=self.inst, competency_tags=['沟通人文', '医患沟通'], - publish_status=2, status=1) - self.pub_trad2 = CaseBase.objects.create( - title='肺炎诊治', case_type='traditional', department=self.dept2, - institution=self.inst, tags='呼吸科', publish_status=2, status=1) - # 不应出现：草稿 / 禁用 / 已下架(软删) - CaseBase.objects.create(title='草稿病例', case_type='traditional', - department=self.dept1, institution=self.inst, publish_status=0, status=1) - CaseBase.objects.create(title='禁用病例', case_type='traditional', - department=self.dept1, institution=self.inst, publish_status=2, status=0) - deleted = CaseBase.objects.create(title='已下架病例', case_type='teaching', - department=self.dept1, institution=self.inst, publish_status=2, status=1) - deleted.delete() # 软删除 - - self.stu = create_test_user(phone='13980000001', real_name='学生甲', - role_type='student', institution=self.inst) - self.stu.department = self.dept1 - self.stu.weak_dimensions = ['沟通人文'] - self.stu.save(update_fields=['department', 'weak_dimensions']) - - # stu 已训练：pub_trad1 低分(60→薄弱)，pub_trad2 高分(90) - self._insert(self.stu.id, self.pub_trad1.id, 60) - self._insert(self.stu.id, self.pub_trad2.id, 90) - - self.client = get_auth_client(self.stu) - - def _ids(self, resp): - return [r['id'] for r in resp.json()['results']] - - # ── 鉴权 ──────────────────────────────────────────────────────────────── - def test_unauthenticated_401(self): - for url in (REC_URL, SPEC_URL, WEAK_URL, TEACH_URL, TASK_URL): - self.assertEqual(APIClient().get(url).status_code, 401, url) - - # ── 5.1 推荐 ─────────────────────────────────────────────────────────── - def test_recommended_only_published(self): - resp = self.client.get(REC_URL) - self.assertEqual(resp.status_code, 200, resp.content) - ids = set(self._ids(resp)) - self.assertEqual(resp.json()['count'], 3) - self.assertEqual(ids, {self.pub_trad1.id, self.pub_teach.id, self.pub_trad2.id}) - - def test_recommended_untrained_first(self): - # pub_teach 未训练且命中薄弱标签+同科室 → 应排在已训练病例之前 - resp = self.client.get(REC_URL) - self.assertEqual(self._ids(resp)[0], self.pub_teach.id, resp.content) - - # ── 5.2 科室专项 ────────────────────────────────────────────────────────── - def test_specialty_default_user_dept(self): - resp = self.client.get(SPEC_URL) - self.assertEqual(set(self._ids(resp)), {self.pub_trad1.id, self.pub_teach.id}) - - def test_specialty_explicit_department(self): - resp = self.client.get(SPEC_URL, {'department': self.dept2.id}) - self.assertEqual(self._ids(resp), [self.pub_trad2.id]) - - # ── 5.3 薄弱环节 ────────────────────────────────────────────────────────── - def test_weak_low_score_cases(self): - resp = self.client.get(WEAK_URL) - self.assertEqual(resp.status_code, 200, resp.content) - results = resp.json()['results'] - self.assertEqual([r['id'] for r in results], [self.pub_trad1.id]) - self.assertEqual(results[0]['my_best_score'], 60.0) - self.assertEqual(results[0]['my_train_count'], 1) - - def test_weak_cold_start_fallback(self): - # 新用户无训练记录，但 weak_dimensions 命中 pub_teach 能力标签 → 回退命中 - fresh = create_test_user(phone='13980000009', role_type='student', institution=self.inst) - fresh.weak_dimensions = ['沟通人文'] - fresh.save(update_fields=['weak_dimensions']) - resp = get_auth_client(fresh).get(WEAK_URL) - self.assertEqual(self._ids(resp), [self.pub_teach.id], resp.content) - - # ── 5.4 教学互动 ────────────────────────────────────────────────────────── - def test_teaching_only_teaching_type(self): - resp = self.client.get(TEACH_URL) - self.assertEqual(self._ids(resp), [self.pub_teach.id]) - - # ── 5.5 教师任务（暂同教学互动）──────────────────────────────────────────── - def test_teacher_task_same_as_teaching(self): - resp = self.client.get(TASK_URL) - self.assertEqual(self._ids(resp), [self.pub_teach.id]) - - # ── 通用过滤 ────────────────────────────────────────────────────────────── - def test_search_filter(self): - resp = self.client.get(REC_URL, {'search': '心梗'}) - self.assertEqual(self._ids(resp), [self.pub_trad1.id]) - - def test_case_type_filter(self): - resp = self.client.get(REC_URL, {'case_type': 'teaching'}) - self.assertEqual(self._ids(resp), [self.pub_teach.id]) +"""移动端病例列表 5 接口测试。 + +- 5.1 推荐病例 GET /api/case/mobile/recommended/ +- 5.2 科室专项 GET /api/case/mobile/specialty/ +- 5.3 薄弱环节 GET /api/case/mobile/weak/ +- 5.4 教学互动 GET /api/case/mobile/teaching/ +- 5.5 教师任务 GET /api/case/mobile/teacher-task/ + +均只取「已发布」病例（publish_status=2 & status=1 & is_deleted=0）。 +training_record 是 managed=False 只读表，迁移占位结构缺新列，故在 setUpClass 里按真实列 +重建该表（仅测试库），用 TransactionTestCase 允许 DDL（与 test_cms_training 一致）。 +""" +import json + +from django.core.cache import cache +from django.db import connection +from django.test import TransactionTestCase +from django.utils import timezone +from rest_framework.test import APIClient + +from apps.user.models import Department +from apps.case.models import CaseBase +from .conftest import create_test_user, get_auth_client, ensure_institution + +CREATE_TR = """ +CREATE TABLE training_record ( + id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, + user_id BIGINT NULL, case_id BIGINT NULL, teacher_id BIGINT NULL, + training_mode VARCHAR(50) NULL, case_type VARCHAR(30) NULL, + start_time DATETIME NULL, end_time DATETIME NULL, duration_seconds INT NULL, + total_score DECIMAL(5,2) NULL, ai_score DECIMAL(5,2) NULL, teacher_score DECIMAL(5,2) NULL, + evaluation_level VARCHAR(20) NULL, status VARCHAR(30) NULL, + feedback TEXT NULL, thinking_chain TEXT NULL, diagnosis_path TEXT NULL, + wrong_points JSON NULL, missed_questions JSON NULL, recommendation_result JSON NULL, + ai_feedback_structured JSON NULL, osce_station_score JSON NULL, + interruption_count INT NULL, emotion_analysis JSON NULL, + prompt_version VARCHAR(50) NULL, rag_context_version VARCHAR(50) NULL, + external_user_id VARCHAR(128) NULL, session_id BIGINT NULL, evaluation_record_id BIGINT NULL, + score_type VARCHAR(20) NULL, pdf_file_path VARCHAR(512) NULL, + created_at DATETIME NULL, updated_at DATETIME NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 +""" + +REC_URL = '/api/case/mobile/recommended/' +SPEC_URL = '/api/case/mobile/specialty/' +WEAK_URL = '/api/case/mobile/weak/' +TEACH_URL = '/api/case/mobile/teaching/' +TASK_URL = '/api/case/mobile/teacher-task/' + + +class MobileCaseListTest(TransactionTestCase): + + @classmethod + def setUpClass(cls): + super().setUpClass() + with connection.cursor() as c: + c.execute('SET FOREIGN_KEY_CHECKS=0') + c.execute('DROP TABLE IF EXISTS training_record') + c.execute(CREATE_TR) + c.execute('SET FOREIGN_KEY_CHECKS=1') + + def _insert(self, user_id, case_id, total_score, status='completed', score_type='percentage'): + now = timezone.now() + with connection.cursor() as c: + c.execute( + "INSERT INTO training_record (user_id, case_id, training_mode, status, " + "total_score, end_time, start_time, score_type, ai_feedback_structured, " + "created_at, updated_at) VALUES (%s,%s,'practice',%s,%s,%s,%s,%s,%s,%s,%s)", + [user_id, case_id, status, total_score, now, now, score_type, + json.dumps({}), now, now], + ) + return c.lastrowid + + def setUp(self): + cache.clear() + with connection.cursor() as c: + c.execute('DELETE FROM training_record') + self.inst = ensure_institution(name='测试医院', code='MCL-H1') + self.dept1 = Department.objects.create(name='心内科', category='临床') + self.dept2 = Department.objects.create(name='呼吸科', category='临床') + + # 已发布病例 + self.pub_trad1 = CaseBase.objects.create( + title='急性心梗', case_type='traditional', department=self.dept1, + institution=self.inst, chief_complaint='胸痛', tags='心内科,胸痛', + publish_status=2, status=1) + self.pub_teach = CaseBase.objects.create( + title='医患沟通教学', case_type='teaching', department=self.dept1, + institution=self.inst, competency_tags=['沟通人文', '医患沟通'], + publish_status=2, status=1) + self.pub_trad2 = CaseBase.objects.create( + title='肺炎诊治', case_type='traditional', department=self.dept2, + institution=self.inst, tags='呼吸科', publish_status=2, status=1) + # 不应出现：草稿 / 禁用 / 已下架(软删) + CaseBase.objects.create(title='草稿病例', case_type='traditional', + department=self.dept1, institution=self.inst, publish_status=0, status=1) + CaseBase.objects.create(title='禁用病例', case_type='traditional', + department=self.dept1, institution=self.inst, publish_status=2, status=0) + deleted = CaseBase.objects.create(title='已下架病例', case_type='teaching', + department=self.dept1, institution=self.inst, publish_status=2, status=1) + deleted.delete() # 软删除 + + self.stu = create_test_user(phone='13980000001', real_name='学生甲', + role_type='student', institution=self.inst) + self.stu.department = self.dept1 + self.stu.weak_dimensions = ['沟通人文'] + self.stu.save(update_fields=['department', 'weak_dimensions']) + + # stu 已训练：pub_trad1 低分(60→薄弱)，pub_trad2 高分(90) + self._insert(self.stu.id, self.pub_trad1.id, 60) + self._insert(self.stu.id, self.pub_trad2.id, 90) + + self.client = get_auth_client(self.stu) + + def _ids(self, resp): + return [r['id'] for r in resp.json()['results']] + + # ── 鉴权 ──────────────────────────────────────────────────────────────── + def test_unauthenticated_401(self): + for url in (REC_URL, SPEC_URL, WEAK_URL, TEACH_URL, TASK_URL): + self.assertEqual(APIClient().get(url).status_code, 401, url) + + # ── 5.1 推荐 ─────────────────────────────────────────────────────────── + def test_recommended_only_published(self): + resp = self.client.get(REC_URL) + self.assertEqual(resp.status_code, 200, resp.content) + ids = set(self._ids(resp)) + self.assertEqual(resp.json()['count'], 3) + self.assertEqual(ids, {self.pub_trad1.id, self.pub_teach.id, self.pub_trad2.id}) + + def test_recommended_untrained_first(self): + # pub_teach 未训练且命中薄弱标签+同科室 → 应排在已训练病例之前 + resp = self.client.get(REC_URL) + self.assertEqual(self._ids(resp)[0], self.pub_teach.id, resp.content) + + # ── 5.2 科室专项 ────────────────────────────────────────────────────────── + def test_specialty_default_user_dept(self): + resp = self.client.get(SPEC_URL) + self.assertEqual(set(self._ids(resp)), {self.pub_trad1.id, self.pub_teach.id}) + + def test_specialty_explicit_department(self): + resp = self.client.get(SPEC_URL, {'department': self.dept2.id}) + self.assertEqual(self._ids(resp), [self.pub_trad2.id]) + + # ── 5.3 薄弱环节 ────────────────────────────────────────────────────────── + def test_weak_low_score_cases(self): + resp = self.client.get(WEAK_URL) + self.assertEqual(resp.status_code, 200, resp.content) + results = resp.json()['results'] + self.assertEqual([r['id'] for r in results], [self.pub_trad1.id]) + self.assertEqual(results[0]['my_best_score'], 60.0) + self.assertEqual(results[0]['my_train_count'], 1) + + def test_weak_cold_start_fallback(self): + # 新用户无训练记录，但 weak_dimensions 命中 pub_teach 能力标签 → 回退命中 + fresh = create_test_user(phone='13980000009', role_type='student', institution=self.inst) + fresh.weak_dimensions = ['沟通人文'] + fresh.save(update_fields=['weak_dimensions']) + resp = get_auth_client(fresh).get(WEAK_URL) + self.assertEqual(self._ids(resp), [self.pub_teach.id], resp.content) + + # ── 5.4 教学互动 ────────────────────────────────────────────────────────── + def test_teaching_only_teaching_type(self): + resp = self.client.get(TEACH_URL) + self.assertEqual(self._ids(resp), [self.pub_teach.id]) + + # ── 5.5 教师任务（暂同教学互动）──────────────────────────────────────────── + def test_teacher_task_same_as_teaching(self): + resp = self.client.get(TASK_URL) + self.assertEqual(self._ids(resp), [self.pub_teach.id]) + + # ── 通用过滤 ────────────────────────────────────────────────────────────── + def test_search_filter(self): + resp = self.client.get(REC_URL, {'search': '心梗'}) + self.assertEqual(self._ids(resp), [self.pub_trad1.id]) + + def test_case_type_filter(self): + resp = self.client.get(REC_URL, {'case_type': 'teaching'}) + self.assertEqual(self._ids(resp), [self.pub_teach.id]) diff --git a/test/test_user_happy.py b/test/test_user_happy.py index 8637544..68592dd 100644 --- a/test/test_user_happy.py +++ b/test/test_user_happy.py @@ -245,7 +245,7 @@ class UserListDetailHappyPathTest(CacheTestCase): """HP-6: teacher GET /users/ → 200，仅包含名下活跃学生""" teacher = create_test_user( phone='13900100010', password='Teacher1', - real_name='王老师', role_type='teacher', + real_name='王老师', role_type='doctor', ) stu_own = create_test_user( phone='13900100011', password='Stu12345', @@ -277,7 +277,7 @@ class UserListDetailHappyPathTest(CacheTestCase): """HP-7: 已结束(status=0)的师生关系学生不出现在列表""" teacher = create_test_user( phone='13900100020', password='Teacher1', - real_name='李老师', role_type='teacher', + real_name='李老师', role_type='doctor', ) stu_active = create_test_user( phone='13900100021', password='Stu12345', @@ -339,7 +339,7 @@ class UserListDetailHappyPathTest(CacheTestCase): """HP-10: teacher GET /users/{student.id}/ → 200，可查看名下学生""" teacher = create_test_user( phone='13900100050', password='Teacher1', - real_name='赵老师', role_type='teacher', + real_name='赵老师', role_type='doctor', ) student = create_test_user( phone='13900100051', password='Stu12345', @@ -370,7 +370,7 @@ class UserListDetailHappyPathTest(CacheTestCase): ) teacher = create_test_user( phone='13900100063', password='Teacher1', - real_name='张老师', role_type='teacher', + real_name='张老师', role_type='doctor', ) client = get_auth_client(admin) diff --git a/test/test_user_negative.py b/test/test_user_negative.py index 52f6799..5a11f42 100644 --- a/test/test_user_negative.py +++ b/test/test_user_negative.py @@ -272,16 +272,16 @@ class UserListDetailNegativeTest(CacheTestCase): self.assertEqual(resp.status_code, 403, resp.content) self.assertEqual(resp.json()['code'], 'USER_NO_LIST_PERMISSION') - def test_doctor_list_403(self): - """N12: doctor GET /users/ → 403 USER_NO_LIST_PERMISSION""" + def test_doctor_list_returns_own_students_only(self): + """N12（新设计）: doctor=带教医生 GET /users/ → 200，仅名下学生（无学生时为空列表）""" doctor = create_test_user( phone='13800002002', password='Doc12345', real_name='医生', role_type='doctor', ) client = get_auth_client(doctor) resp = client.get(USER_LIST_URL) - self.assertEqual(resp.status_code, 403, resp.content) - self.assertEqual(resp.json()['code'], 'USER_NO_LIST_PERMISSION') + self.assertEqual(resp.status_code, 200, resp.content) + self.assertEqual(resp.json()['count'], 0) # 无名下学生 → 空 def test_unauth_list_401(self): """N13: 未登录 GET /users/ → 401""" @@ -315,7 +315,7 @@ class UserListDetailNegativeTest(CacheTestCase): """N16: teacher 查看非名下学生详情 → 403 USER_NO_VIEW_PERMISSION""" teacher = create_test_user( phone='13800002030', password='Teacher1', - real_name='刘老师', role_type='teacher', + real_name='刘老师', role_type='doctor', ) unrelated = create_test_user( phone='13800002031', password='Stu12345', @@ -331,7 +331,7 @@ class UserListDetailNegativeTest(CacheTestCase): """N17: teacher 查看已结束关系学生详情 → 403 USER_NO_VIEW_PERMISSION""" teacher = create_test_user( phone='13800002040', password='Teacher1', - real_name='陈老师', role_type='teacher', + real_name='陈老师', role_type='doctor', ) student = create_test_user( phone='13800002041', password='Stu12345',