From 05ce7e987e109c15e68fa1fb0ead231a98e7aa65 Mon Sep 17 00:00:00 2001 From: shihan11 Date: Sat, 13 Jun 2026 01:44:31 +0800 Subject: [PATCH] feat: cms overview and mobile case query --- apps/case/mobile.py | 233 +++++++++++++++++ apps/case/urls.py | 7 + apps/cms/permissions.py | 20 ++ apps/cms/stats.py | 476 ++++++++++++++++++++++++++++++++++ apps/cms/training.py | 248 ++++++++++++++++++ apps/cms/urls.py | 11 +- apps/training/models.py | 23 ++ apps/user/cms.py | 28 ++ test/swagger_cms_stats.py | 148 +++++++++++ test/test_cms_stats.py | 180 +++++++++++++ test/test_cms_training.py | 235 +++++++++++++++++ test/test_mobile_case_list.py | 179 +++++++++++++ 12 files changed, 1787 insertions(+), 1 deletion(-) create mode 100644 apps/case/mobile.py create mode 100644 apps/cms/stats.py create mode 100644 apps/cms/training.py create mode 100644 test/swagger_cms_stats.py create mode 100644 test/test_cms_stats.py create mode 100644 test/test_cms_training.py create mode 100644 test/test_mobile_case_list.py diff --git a/apps/case/mobile.py b/apps/case/mobile.py new file mode 100644 index 0000000..f04a675 --- /dev/null +++ b/apps/case/mobile.py @@ -0,0 +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) diff --git a/apps/case/urls.py b/apps/case/urls.py index 8f5ab62..b626d94 100644 --- a/apps/case/urls.py +++ b/apps/case/urls.py @@ -1,6 +1,7 @@ from django.urls import path, include from rest_framework.routers import DefaultRouter from . import views +from . import mobile router = DefaultRouter() router.register(r'cases', views.CaseBaseViewSet, basename='case') @@ -11,5 +12,11 @@ router.register(r'case-stages', views.CaseStageViewSet, basename='case-stage') router.register(r'scoring-rules', views.ScoringRuleViewSet, basename='scoring-rule') urlpatterns = [ + # 移动端病例列表（5 个入口） + path('mobile/recommended/', mobile.recommended, name='mobile-case-recommended'), + path('mobile/specialty/', mobile.specialty, name='mobile-case-specialty'), + path('mobile/weak/', mobile.weak, name='mobile-case-weak'), + path('mobile/teaching/', mobile.teaching, name='mobile-case-teaching'), + path('mobile/teacher-task/', mobile.teacher_task, name='mobile-case-teacher-task'), path('', include(router.urls)), ] diff --git a/apps/cms/permissions.py b/apps/cms/permissions.py index 58b57d3..1d0880b 100644 --- a/apps/cms/permissions.py +++ b/apps/cms/permissions.py @@ -63,6 +63,26 @@ class IsSuperContentOrHospitalAdmin(BasePermission): '需要超级管理员 / 内容管理员 / 医院管理员权限', status_code=403) +class IsHospitalAdmin(BasePermission): + """仅医院管理员（role_type=hospital_admin）。数据范围（本院）由 ViewSet 收口。""" + + def has_permission(self, request, view): + user = request.user + if user and user.is_authenticated and is_hospital_admin(user): + return True + raise AppError('CMS_PERMISSION_DENIED', '仅医院管理员可访问该接口', status_code=403) + + +class IsContentAdmin(BasePermission): + """仅内容管理员（role_type=content_admin）。数据范围（本院）由 ViewSet 收口。""" + + def has_permission(self, request, view): + user = request.user + if user and user.is_authenticated and is_content_admin(user): + return True + raise AppError('CMS_PERMISSION_DENIED', '仅内容管理员可访问该接口', status_code=403) + + class IsTeacher(BasePermission): """仅带教医生（role_type=doctor）可访问。 diff --git a/apps/cms/stats.py b/apps/cms/stats.py new file mode 100644 index 0000000..ecfd1a0 --- /dev/null +++ b/apps/cms/stats.py @@ -0,0 +1,476 @@ +"""CMS 各角色概览大屏聚合接口（只读 GET）。 + +依据《各角色首页数据展示.pdf》与本文档第八章，**全部基于现有库真实列**： +- 训练聚合主表 `TrainingRecord`（有 `user_id`→机构/科室、`case_id`、`status`、`total_score`、`ai_feedback_structured`）； +- `TrainingSession`（仅平台级「发起/完成」，无机构外键）；得分按 `score_type` 归一（five_point×20）。 +- 维度雷达固定 5 维（信息获取/分析推理/处置决策/沟通人文/临床整合，复用 `apps/user/stats`）。 +- 科室为全局表：凡「按科室分组」走 `case_base.department_id`，不按机构过滤科室。 + +不做项：AI 调用量 / AI 响应时长（超管平台总览）、知识文档数（内容概览）、带教任务/分配任务（无任务表）。 +""" +from collections import defaultdict +from datetime import datetime, timedelta + +from django.db.models import Count, Sum, Avg, Q +from django.utils import timezone +from rest_framework.decorators import api_view, permission_classes +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response + +from apps.user.models import User, Institution, Department, TeacherStudentRelation +from apps.case.models import CaseBase +from apps.training.models import TrainingRecord, TrainingSession +from apps.user.stats import STANDARD_DIMS, DIMENSION_MAP, _dimension_scores, _NORM_TOTAL_EXPR +from apps.cms.permissions import IsSuperAdmin, IsHospitalAdmin, IsContentAdmin, IsTeacher + +PASS_SCORE = 60 # 通过线（归一百分制） + + +# ── 通用小工具 ────────────────────────────────────────────────────────────── + +def _recent_months(n=6): + """返回最近 n 个月 (year, month)，由旧到新。""" + now = timezone.localtime() + out, y, m = [], now.year, now.month + for _ in range(n): + out.append((y, m)) + m -= 1 + if m == 0: + m, y = 12, y - 1 + return list(reversed(out)) + + +def _month_label(y, m): + return f'{y}-{m:02d}' + + +def _month_start(y, m): + return timezone.make_aware(datetime(y, m, 1)) + + +def _norm_avg(qs): + """TrainingRecord 查询集 → 归一百分制平均分（保留 1 位）或 None。""" + v = qs.aggregate(a=Avg(_NORM_TOTAL_EXPR))['a'] + return round(float(v), 1) if v is not None else None + + +def _norm(score, score_type): + if score is None: + return None + return float(score) * 20 if score_type == 'five_point' else float(score) + + +def _mom(this, last): + """环比%（本期 vs 上期），上期为 0/None 返回 None。""" + if not last: + return None + return round((this - last) / last * 100) + + +def _hours(seconds): + return round((seconds or 0) / 3600, 1) + + +def _radar(records): + """固定 5 维雷达：各维得分率均值（0~100），无数据维度记 0。""" + buckets = defaultdict(list) + for r in records: + for dim, score, mx in _dimension_scores(r): + std = DIMENSION_MAP.get(dim) + if std and score is not None and mx and float(mx) > 0: + buckets[std].append(float(score) / float(mx) * 100) + return [{'dimension': d, 'score': (round(sum(buckets[d]) / len(buckets[d])) if buckets[d] else 0)} + for d in STANDARD_DIMS] + + +def _inst_names(ids): + return {i.id: i.name for i in Institution.all_objects.filter(id__in=[x for x in ids if x])} + + +def _dept_names(ids): + return {d.id: d.name for d in Department.all_objects.filter(id__in=[x for x in ids if x])} + + +def _case_titles(ids): + return {c.id: c.title for c in CaseBase.all_objects.filter(id__in=[x for x in ids if x])} + + +def _pass_rates(tr_qs): + """按 case 统计通过率（归一分≥60）。返回 [{case_id, pass_rate, total}]（小数据集 Python 聚合）。""" + agg = defaultdict(lambda: [0, 0]) # case_id -> [passed, total] + for r in tr_qs.only('case_id', 'total_score', 'score_type'): + n = _norm(r.total_score, r.score_type) + if n is None: + continue + agg[r.case_id][1] += 1 + if n >= PASS_SCORE: + agg[r.case_id][0] += 1 + out = [{'case_id': cid, 'pass_rate': round(p / t * 100) if t else 0, 'total': t} + for cid, (p, t) in agg.items()] + return out + + +# 注：MySQL 未装时区表，DB 侧 TruncMonth/ExtractHour（CONVERT_TZ）会报错； +# 故月/小时分桶一律在 Python 用 localtime 完成（与 apps/user/stats 一致）。 + +def _tr_monthly(qs, n=6): + """TrainingRecord 月度：返回 (labels, counts, active_users)。Python 分桶。""" + months = _recent_months(n) + idx = {ym: i for i, ym in enumerate(months)} + counts = [0] * n + users = [set() for _ in range(n)] + for created_at, uid in qs.filter(created_at__gte=_month_start(*months[0])).values_list('created_at', 'user_id'): + if not created_at: + continue + lt = timezone.localtime(created_at) + i = idx.get((lt.year, lt.month)) + if i is not None: + counts[i] += 1 + users[i].add(uid) + return [_month_label(*ym) for ym in months], counts, [len(s) for s in users] + + +def _ts_monthly(n=6): + """TrainingSession 月度发起次数：返回 (labels, counts)。Python 分桶。""" + months = _recent_months(n) + idx = {ym: i for i, ym in enumerate(months)} + counts = [0] * n + for (created_at,) in (TrainingSession.objects.filter(created_at__gte=_month_start(*months[0])) + .values_list('created_at')): + if not created_at: + continue + lt = timezone.localtime(created_at) + i = idx.get((lt.year, lt.month)) + if i is not None: + counts[i] += 1 + return [_month_label(*ym) for ym in months], counts + + +def _named_rank(rows, key, name_map, label='name', value='value'): + """把 [{key:id, ...value}] 附上名称。""" + return [{**{label: name_map.get(r[key], ''), 'id': r[key]}, + value: r.get(value)} for r in rows] + + +# ── 8.1 超级管理员 · 平台总览 ──────────────────────────────────────────────── + +@api_view(['GET']) +@permission_classes([IsAuthenticated, IsSuperAdmin]) +def platform_overview(request): + now = timezone.now() + d30 = now - timedelta(days=30) + TR = TrainingRecord.objects.all() + TS = TrainingSession.objects.all() + + ms = _month_start(now.year, now.month) + pm_y, pm_m = (now.year, now.month - 1) if now.month > 1 else (now.year - 1, 12) + pms = _month_start(pm_y, pm_m) + + # KPI + kpi = { + 'institution_count': Institution.objects.count(), + 'mau_institution': TR.filter(created_at__gte=d30).exclude(user__institution_id__isnull=True) + .values('user__institution_id').distinct().count(), + 'user_total': User.objects.filter(status=1).count(), + 'mau_user': TR.filter(created_at__gte=d30).values('user_id').distinct().count(), + } + + # 核心指标 + sess_total = TS.count() + sess_done = TS.filter(completed_at__isnull=False).count() + train_this = TS.filter(created_at__gte=ms).count() + train_last = TS.filter(created_at__gte=pms, created_at__lt=ms).count() + core = { + 'train_new_month': train_this, + 'train_new_mom': _mom(train_this, train_last), + 'train_total': sess_total, + 'complete_rate': round(sess_done / sess_total * 100, 1) if sess_total else 0, + 'avg_score': _norm_avg(TR.filter(status='completed')) or 0, + } + + # 用户与活跃度趋势 + labels, sess_counts = _ts_monthly(6) + _, _, active_users = _tr_monthly(TR, 6) + trend = {'months': labels, 'train_counts': sess_counts, 'active_users': active_users} + + user_compose = [{'role': r['role_type'] or 'unknown', 'count': r['n']} + for r in User.objects.filter(status=1).values('role_type').annotate(n=Count('id'))] + + since7 = now - timedelta(days=7) + hbucket = defaultdict(int) + for (created_at,) in TR.filter(created_at__gte=since7).values_list('created_at'): + if created_at: + hbucket[timezone.localtime(created_at).hour] += 1 + hourly_7d = [{'hour': h, 'avg': round(hbucket.get(h, 0) / 7, 2)} for h in range(24)] + + # 各机构用户人数 / 活跃 + inst_users = {r['institution_id']: r['n'] for r in User.objects.filter(status=1) + .exclude(institution_id__isnull=True).values('institution_id').annotate(n=Count('id'))} + inst_active = {r['user__institution_id']: r['c'] for r in TR.filter(created_at__gte=d30) + .exclude(user__institution_id__isnull=True).values('user__institution_id') + .annotate(c=Count('user_id', distinct=True))} + inames = _inst_names(set(inst_users) | set(inst_active)) + institution_dist = [{'id': iid, 'institution': inames.get(iid, ''), + 'users': inst_users[iid], 'active': inst_active.get(iid, 0)} + for iid in inst_users] + + # 医院使用排行 + by_train = list(TR.exclude(user__institution_id__isnull=True).values('user__institution_id') + .annotate(value=Count('id')).order_by('-value')[:10]) + by_score = list(TR.exclude(user__institution_id__isnull=True).values('user__institution_id') + .annotate(value=Avg(_NORM_TOTAL_EXPR)).order_by('-value')[:10]) + rnames = _inst_names([r['user__institution_id'] for r in by_train + by_score]) + hospital_rank = { + 'by_train': [{'id': r['user__institution_id'], 'institution': rnames.get(r['user__institution_id'], ''), + 'count': r['value']} for r in by_train], + 'by_avg_score': [{'id': r['user__institution_id'], 'institution': rnames.get(r['user__institution_id'], ''), + 'avg_score': round(float(r['value']), 1) if r['value'] is not None else None} + for r in by_score], + } + + # 病例资产与使用 + case_total = CaseBase.objects.count() + case_this = CaseBase.objects.filter(created_at__gte=ms).count() + case_last = CaseBase.objects.filter(created_at__gte=pms, created_at__lt=ms).count() + type_dist = [{'case_type': r['case_type'], 'count': r['n']} + for r in CaseBase.objects.values('case_type').annotate(n=Count('id'))] + # 每种类型使用率 = 该类型被训练过的去重病例数 / 该类型病例总数 + type_total = {r['case_type']: r['n'] for r in CaseBase.objects.values('case_type').annotate(n=Count('id'))} + used_by_type = defaultdict(set) + for r in TR.exclude(case_id__isnull=True).values('case_id', 'case__case_type'): + used_by_type[r['case__case_type']].add(r['case_id']) + type_usage_rate = [{'case_type': t, 'rate': round(len(used_by_type.get(t, set())) / n * 100) if n else 0} + for t, n in type_total.items()] + top_used_rows = list(TR.exclude(case_id__isnull=True).values('case_id') + .annotate(value=Count('id')).order_by('-value')[:10]) + ctitles = _case_titles([r['case_id'] for r in top_used_rows]) + top_used = [{'case_id': r['case_id'], 'title': ctitles.get(r['case_id'], ''), 'count': r['value']} + for r in top_used_rows] + prates = _pass_rates(TR) + ptitles = _case_titles([p['case_id'] for p in prates]) + low_pass = sorted(prates, key=lambda x: x['pass_rate'])[:10] + low_pass = [{**p, 'title': ptitles.get(p['case_id'], '')} for p in low_pass] + case_asset = { + 'total': case_total, 'new_month': case_this, 'new_mom': _mom(case_this, case_last), + 'type_dist': type_dist, 'type_usage_rate': type_usage_rate, + 'top_used': top_used, 'low_pass': low_pass, + } + + return Response({ + 'kpi': kpi, 'core': core, 'trend': trend, 'user_compose': user_compose, + 'hourly_7d': hourly_7d, 'institution_dist': institution_dist, + 'hospital_rank': hospital_rank, 'case_asset': case_asset, + }) + + +# ── 8.2 医院管理员 · 医院驾驶舱（本院）────────────────────────────────────── + +@api_view(['GET']) +@permission_classes([IsAuthenticated, IsHospitalAdmin]) +def hospital_overview(request): + inst_id = request.user.institution_id + now = timezone.now() + d30 = now - timedelta(days=30) + ms = _month_start(now.year, now.month) + pm_y, pm_m = (now.year, now.month - 1) if now.month > 1 else (now.year - 1, 12) + pms = _month_start(pm_y, pm_m) + + inst = Institution.all_objects.filter(id=inst_id).first() + TR_h = TrainingRecord.objects.filter(user__institution_id=inst_id) # 本院学员的训练 + CASE_h = CaseBase.objects.filter(institution_id=inst_id) # 本院病例 + + profile = { + 'institution_id': inst_id, + 'name': inst.name if inst else '', + 'logo': inst.banner_url if inst else '', + 'level': inst.level if inst else '', + 'cooperation_days': (now - inst.created_at).days if inst and inst.created_at else None, + } + + summary = { + 'dept_count': CASE_h.exclude(department_id__isnull=True).values('department_id').distinct().count(), + 'doctor_count': User.objects.filter(institution_id=inst_id, role_type='doctor', status=1).count(), + 'student_count': User.objects.filter(institution_id=inst_id, role_type='student', status=1).count(), + 'train_total': TR_h.count(), + 'complete_rate': round(TR_h.filter(status='completed').count() / TR_h.count() * 100, 1) if TR_h.count() else 0, + 'avg_score': _norm_avg(TR_h.filter(status='completed')) or 0, + } + labels, counts, _ = _tr_monthly(TR_h, 6) + summary['train_months'] = labels + summary['train_monthly'] = counts + + # 科室排行（全局科室，按病例所属科室聚合本院数据） + dept_case = {r['department_id']: r['n'] for r in CASE_h.exclude(department_id__isnull=True) + .values('department_id').annotate(n=Count('id'))} + dept_train = {r['case__department_id']: (r['c'], r['done']) for r in TR_h + .exclude(case__department_id__isnull=True).values('case__department_id') + .annotate(c=Count('id'), + done=Count('id', filter=Q(status='completed')))} + dept_active = {r['case__department_id']: r['u'] for r in TR_h.filter(created_at__gte=d30) + .exclude(case__department_id__isnull=True).values('case__department_id') + .annotate(u=Count('user_id', distinct=True))} + dept_score = {r['case__department_id']: r['a'] for r in TR_h.exclude(case__department_id__isnull=True) + .values('case__department_id').annotate(a=Avg(_NORM_TOTAL_EXPR))} + dnames = _dept_names(set(dept_case) | set(dept_train) | set(dept_active) | set(dept_score)) + dept_rank = [{ + 'id': did, 'department': dnames.get(did, ''), + 'case_count': dept_case.get(did, 0), + 'train_count': dept_train.get(did, (0, 0))[0], + 'effective_train': dept_train.get(did, (0, 0))[1], + 'active_users': dept_active.get(did, 0), + 'avg_score': round(float(dept_score[did]), 1) if dept_score.get(did) is not None else None, + } for did in (set(dept_case) | set(dept_train))] + + case_asset = { + 'total': CASE_h.count(), + 'new_month': CASE_h.filter(created_at__gte=ms).count(), + 'top_used': [], 'pass_high': [], 'pass_low': [], + } + top_rows = list(TR_h.exclude(case_id__isnull=True).values('case_id') + .annotate(value=Count('id')).order_by('-value')[:10]) + ctitles = _case_titles([r['case_id'] for r in top_rows]) + case_asset['top_used'] = [{'case_id': r['case_id'], 'title': ctitles.get(r['case_id'], ''), + 'count': r['value']} for r in top_rows] + prates = _pass_rates(TR_h) + ptitles = _case_titles([p['case_id'] for p in prates]) + prates = [{**p, 'title': ptitles.get(p['case_id'], '')} for p in prates] + case_asset['pass_high'] = sorted(prates, key=lambda x: -x['pass_rate'])[:5] + case_asset['pass_low'] = sorted(prates, key=lambda x: x['pass_rate'])[:5] + + competency = { + 'student_avg': _norm_avg(TR_h.filter(status='completed')) or 0, + 'platform_avg': _norm_avg(TrainingRecord.objects.filter(status='completed')) or 0, + 'radar': _radar(TR_h.only('ai_feedback_structured')), + 'radar_platform': _radar(TrainingRecord.objects.only('ai_feedback_structured')), + } + + return Response({'profile': profile, 'summary': summary, 'dept_rank': dept_rank, + 'case_asset': case_asset, 'competency': competency}) + + +# ── 8.3 内容管理员 · 内容概览（本院）──────────────────────────────────────── + +@api_view(['GET']) +@permission_classes([IsAuthenticated, IsContentAdmin]) +def content_overview(request): + inst_id = request.user.institution_id + now = timezone.now() + ms = _month_start(now.year, now.month) + pm_y, pm_m = (now.year, now.month - 1) if now.month > 1 else (now.year - 1, 12) + pms = _month_start(pm_y, pm_m) + + CASE_h = CaseBase.objects.filter(institution_id=inst_id) + TR_h = TrainingRecord.objects.filter(case__institution_id=inst_id) + + case_total = CASE_h.count() + used_cases = TR_h.exclude(case_id__isnull=True).values('case_id').distinct().count() + case_this = CASE_h.filter(created_at__gte=ms).count() + case_last = CASE_h.filter(created_at__gte=pms, created_at__lt=ms).count() + tr_this = TR_h.filter(created_at__gte=ms).count() + tr_last = TR_h.filter(created_at__gte=pms, created_at__lt=ms).count() + summary = { + 'case_total': case_total, + 'pending_publish': CASE_h.filter(publish_status=1).count(), # 1=正常=待发布 + 'case_new_month': case_this, 'case_new_mom': _mom(case_this, case_last), + 'usage_rate': round(used_cases / case_total * 100) if case_total else 0, + 'train_total': TR_h.count(), 'train_mom': _mom(tr_this, tr_last), + } + + type_dist = [{'case_type': r['case_type'], 'count': r['n']} + for r in CASE_h.values('case_type').annotate(n=Count('id'))] + type_train = {r['case__case_type']: r['c'] for r in TR_h.values('case__case_type').annotate(c=Count('id'))} + # 不同科室（全局科室）病例分布 / 使用数(去重) / 训练次数 + dept_case = {r['department_id']: r['n'] for r in CASE_h.exclude(department_id__isnull=True) + .values('department_id').annotate(n=Count('id'))} + dept_train = {r['case__department_id']: r['c'] for r in TR_h.exclude(case__department_id__isnull=True) + .values('case__department_id').annotate(c=Count('id'))} + dept_used = defaultdict(set) + for r in TR_h.exclude(case__department_id__isnull=True).values('case__department_id', 'case_id'): + dept_used[r['case__department_id']].add(r['case_id']) + dnames = _dept_names(set(dept_case) | set(dept_train)) + dept_dist = [{'id': did, 'department': dnames.get(did, ''), + 'case_count': dept_case.get(did, 0), + 'used_count': len(dept_used.get(did, set())), + 'train_count': dept_train.get(did, 0)} for did in (set(dept_case) | set(dept_train))] + # 难度分布与使用次数 + diff_case = {r['difficulty'] or '未分级': r['n'] for r in CASE_h.values('difficulty').annotate(n=Count('id'))} + diff_train = defaultdict(int) + for r in TR_h.values('case__difficulty').annotate(c=Count('id')): + diff_train[r['case__difficulty'] or '未分级'] += r['c'] + difficulty_dist = [{'difficulty': k, 'case_count': v, 'train_count': diff_train.get(k, 0)} + for k, v in diff_case.items()] + dist = {'type_dist': type_dist, 'type_train': type_train, 'dept_dist': dept_dist, + 'difficulty_dist': difficulty_dist} + + prates = _pass_rates(TR_h) + ptitles = _case_titles([p['case_id'] for p in prates]) + prates = [{**p, 'title': ptitles.get(p['case_id'], '')} for p in prates] + warning = sorted(prates, key=lambda x: x['pass_rate'])[:5] # 低通过率预警 + + hot_rows = list(TR_h.exclude(case_id__isnull=True).values('case_id') + .annotate(value=Count('id')).order_by('-value')[:5]) + htitles = _case_titles([r['case_id'] for r in hot_rows]) + hot = [{'case_id': r['case_id'], 'title': htitles.get(r['case_id'], ''), 'count': r['value']} + for r in hot_rows] + + return Response({'summary': summary, 'dist': dist, 'warning': warning, 'hot': hot}) + + +# ── 8.4 带教医生 · 教学概览（名下学生）────────────────────────────────────── + +@api_view(['GET']) +@permission_classes([IsAuthenticated, IsTeacher]) +def teaching_overview(request): + teacher = request.user + student_ids = list(TeacherStudentRelation.objects.filter(teacher=teacher, status=1) + .values_list('student_id', flat=True)) + students_qs = User.objects.filter(id__in=student_ids) + TR_s = TrainingRecord.objects.filter(user_id__in=student_ids) + + # 每个学生派生指标 + by_user = defaultdict(lambda: {'total': 0, 'done': 0, 'score_sum': 0.0, 'score_n': 0, + 'last': None, 'types': defaultdict(int)}) + for r in TR_s.only('user_id', 'status', 'total_score', 'score_type', 'end_time', 'case_type'): + b = by_user[r.user_id] + b['total'] += 1 + if r.status == 'completed': + b['done'] += 1 + n = _norm(r.total_score, r.score_type) + if n is not None: + b['score_sum'] += n + b['score_n'] += 1 + if r.case_type: + b['types'][r.case_type] += 1 + if r.end_time and (b['last'] is None or r.end_time > b['last']): + b['last'] = r.end_time + + students = [] + for u in students_qs.select_related('department'): + b = by_user.get(u.id) + most_type = max(b['types'].items(), key=lambda x: x[1])[0] if (b and b['types']) else None + students.append({ + 'id': u.id, 'real_name': u.real_name, 'username': u.username, + 'department': u.department.name if u.department_id else '', + 'train_total': b['total'] if b else 0, + 'complete_rate': round(b['done'] / b['total'] * 100) if (b and b['total']) else 0, + 'avg_score': round(b['score_sum'] / b['score_n'], 1) if (b and b['score_n']) else None, + 'weak_dimensions': u.weak_dimensions or [], + 'most_trained_type': most_type, + 'last_trained_at': b['last'] if b else None, + 'pending_tasks': None, # ⛔ 无任务表，不做 + }) + + # 整体概览 + inst_id = teacher.institution_id + labels, counts, _ = _tr_monthly(TR_s, 6) + overview = { + 'student_count': len(student_ids), + 'radar': _radar(TR_s.only('ai_feedback_structured')), + 'students_avg': _norm_avg(TR_s.filter(status='completed')) or 0, + 'institution_avg': _norm_avg(TrainingRecord.objects.filter( + user__institution_id=inst_id, status='completed')) or 0, + 'train_months': labels, 'train_monthly': counts, + 'task_summary': None, # ⛔ 无任务表，不做（进行中任务/完成情况/得分分布） + } + + return Response({'students': students, 'overview': overview}) diff --git a/apps/cms/training.py b/apps/cms/training.py new file mode 100644 index 0000000..d25ff4b --- /dev/null +++ b/apps/cms/training.py @@ -0,0 +1,248 @@ +"""CMS 训练记录查询 + 带教能力画像 / 排行榜（只读，fastapi 属主表 training_record）。 + +- CMS-TRN-1 超管训练记录列表：`GET /api/cms/training-records/`（全平台，`IsSuperAdmin`）。 +- CMS-TEA-3 带教教学工具训练记录：`GET /api/cms/students/training-records/` + （名下学生，由 `CmsStudentViewSet` 的 action 复用 `build_training_records`）。 +- CMS-TEA-4 带教能力画像：`GET /api/cms/students/{id}/competency/`（复用移动端智能分析 5 维口径）。 +- CMS-TEA-5 带教排行榜：`GET /api/cms/students/ranking/`（名下学生多维度排名）。 + +得分按 `score_type` 归一（five_point×20）；雷达/得分率走 `ai_feedback_structured.dimension_scores` +（带 max_score）。全程只读，不写训练表。 +""" +from collections import defaultdict +from datetime import datetime, timedelta + +from django.db.models import Q, Avg, Sum +from django.utils import timezone +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 apps.training.models import TrainingRecord +from apps.user.models import User +from apps.case.models import CaseBase +from apps.user.stats import ( + STANDARD_DIMS, DIMENSION_MAP, _dimension_scores, _diagnosis_accuracy, + _norm_total, _NORM_TOTAL_EXPR, +) +from apps.cms.permissions import IsSuperAdmin + +# 排行榜支持的维度：分数 / 训练次数 / 完成数 + 固定 5 个能力维度（得分率） +RANK_BASE_DIMS = ('score', 'train_count', 'completed') +RANK_DIMS = RANK_BASE_DIMS + tuple(STANDARD_DIMS) + + +def _is_number(s): + try: + float(s) + return True + except (TypeError, ValueError): + return False + + +def _parse_date(s): + """'YYYY-MM-DD' / 'YYYY/MM/DD' → aware datetime（当天 00:00）或 None。""" + s = (s or '').strip() + for fmt in ('%Y-%m-%d', '%Y/%m/%d'): + try: + return timezone.make_aware(datetime.strptime(s, fmt)) + except ValueError: + continue + return None + + +# ── 训练记录列表（CMS-TRN-1 / CMS-TEA-3 共用）────────────────────────────────── + +def _filter_records(base, request): + p = request.query_params + search = (p.get('search') or '').strip() + if search: + base = base.filter( + Q(case__title__icontains=search) + | Q(user__real_name__icontains=search) + | Q(user__phone__icontains=search) + ) + for key, field in (('user_id', 'user_id'), ('case_id', 'case_id')): + v = (p.get(key) or '').strip() + if v.isdigit(): + base = base.filter(**{field: int(v)}) + status = (p.get('status') or '').strip() + if status: + base = base.filter(status=status) + for f in ('training_mode', 'case_type'): + v = (p.get(f) or '').strip() + if v: + base = base.filter(**{f: v}) + inst = (p.get('institution') or '').strip() + if inst.isdigit(): + base = base.filter(user__institution_id=int(inst)) + d0, d1 = _parse_date(p.get('start_date')), _parse_date(p.get('end_date')) + if d0: + base = base.filter(created_at__gte=d0) + if d1: + base = base.filter(created_at__lt=d1 + timedelta(days=1)) + # 归一分数范围过滤 + min_s, max_s = (p.get('min_score') or '').strip(), (p.get('max_score') or '').strip() + if _is_number(min_s) or _is_number(max_s): + base = base.annotate(_norm=_NORM_TOTAL_EXPR) + if _is_number(min_s): + base = base.filter(_norm__gte=float(min_s)) + if _is_number(max_s): + base = base.filter(_norm__lte=float(max_s)) + return base + + +def _serialize_record(r, users, cases): + """users/cases 为按 id 预取的字典（避免对 fastapi 表的非空 FK 做 INNER JOIN + 丢掉宿主系统外部用户/孤立病例的记录）。""" + user = users.get(r.user_id) + case = cases.get(r.case_id) + dept = case.department if (case and case.department_id) else None + return { + 'record_id': r.id, + 'user_id': r.user_id, + 'user_name': user.real_name if user else '', + 'institution': user.institution_id if user else None, + 'institution_name': (user.institution.name if (user and user.institution_id) else ''), + 'case_id': r.case_id, + 'case_title': case.title if case else '', + 'department': dept.name if dept else '', + 'training_mode': r.training_mode, + 'case_type': r.case_type, + 'status': r.status, + 'score': float(r.total_score) if r.total_score is not None else None, + 'score_type': r.score_type, + 'score_normalized': (round(_norm_total(r.total_score, r.score_type), 1) + if r.total_score is not None else None), + 'evaluation_level': r.evaluation_level, + 'trained_at': r.end_time or r.created_at, + 'duration_seconds': r.duration_seconds, + } + + +def build_training_records(base, request): + """对给定 base 查询集应用过滤 + 分页 + 汇总，返回分页 Response。 + + 超管传 `TrainingRecord.objects.all()`；带教传名下学生范围的查询集。 + ⚠️ 不用 select_related（user/case 在 fastapi 表里是非空 FK，INNER JOIN 会丢掉 + 宿主系统外部用户/本地缺失病例的记录，导致 count 与 results 不一致）；改为按 id 批量预取。 + """ + base = _filter_records(base, request) + avg = base.filter(status='completed').aggregate(a=Avg(_NORM_TOTAL_EXPR))['a'] + summary = { + 'total': base.count(), + 'completed': base.filter(status='completed').count(), + 'avg_score': round(float(avg), 1) if avg is not None else None, + } + paginator = PageNumberPagination() + page = paginator.paginate_queryset(base.order_by('-end_time', '-id'), request) + users = User.all_objects.filter(id__in={r.user_id for r in page}).select_related('institution') + cases = CaseBase.all_objects.filter(id__in={r.case_id for r in page}).select_related('department') + users = {u.id: u for u in users} + cases = {c.id: c for c in cases} + resp = paginator.get_paginated_response([_serialize_record(r, users, cases) for r in page]) + resp.data['summary'] = summary + return resp + + +@api_view(['GET']) +@permission_classes([IsAuthenticated, IsSuperAdmin]) +def training_records(request): + """CMS-TRN-1 超管训练记录列表（全平台只读）。""" + return build_training_records(TrainingRecord.objects.all(), request) + + +# ── 能力画像（CMS-TEA-4）────────────────────────────────────────────────────── + +def _radar_buckets(records): + """各标准维度得分率列表。返回 {标准维度: [得分率,...]}。""" + buckets = defaultdict(list) + for r in records: + for dim, score, mx in _dimension_scores(r): + std = DIMENSION_MAP.get(dim) + if std and score is not None and mx and float(mx) > 0: + buckets[std].append(float(score) / float(mx) * 100) + return buckets + + +def student_competency(student): + """单个学生的能力画像（复用移动端「智能分析」固定 5 维口径）。""" + qs = TrainingRecord.objects.filter(user_id=student.id, status='completed') + recs = list(qs.only('total_score', 'score_type', 'duration_seconds', 'ai_feedback_structured')) + agg = qs.aggregate(total=Sum('duration_seconds'), avg=Avg(_NORM_TOTAL_EXPR)) + avg = agg['avg'] + + buckets = _radar_buckets(recs) + radar = [{'dimension': d, 'score': (round(sum(buckets[d]) / len(buckets[d])) if buckets[d] else 0)} + for d in STANDARD_DIMS] + ordered = sorted((x for x in radar if buckets[x['dimension']]), key=lambda x: x['score']) + weak = [ordered[0]['dimension']] if ordered else [] + comment = '' + if len(ordered) >= 2 and ordered[0]['score'] != ordered[-1]['score']: + comment = (f"该学生{ordered[-1]['dimension']}表现突出，但{ordered[0]['dimension']}" + f"仍有提升空间，建议加强相关训练。") + elif ordered: + comment = f"各维度表现较为均衡，可针对{weak[0]}做进一步强化。" + + return { + 'student_id': student.id, + 'real_name': student.real_name, + 'completed_cases': qs.count(), + 'total_hours': round((agg['total'] or 0) / 3600, 1), + 'avg_score': round(float(avg), 1) if avg is not None else 0, + 'diagnosis_accuracy': _diagnosis_accuracy(recs), + 'radar': radar, + 'weak_dimensions': weak, + 'comment': comment, + } + + +# ── 排行榜（CMS-TEA-5）──────────────────────────────────────────────────────── + +def build_ranking(students, dimension): + """名下学生按指定维度排名。 + + dimension ∈ {score(默认), train_count, completed} ∪ STANDARD_DIMS（各维度得分率）。 + 无数据的学生 value=None，排在末尾。 + """ + dimension = (dimension or 'score').strip() + if dimension not in RANK_DIMS: + dimension = 'score' + + info = {u.id: (u.real_name, (u.department.name if u.department_id else '')) for u in students} + ids = list(info) + metrics = {i: {'train_count': 0, 'completed': 0, 'score_sum': 0.0, 'score_n': 0, + 'dim': defaultdict(list)} for i in ids} + for r in (TrainingRecord.objects.filter(user_id__in=ids) + .only('user_id', 'status', 'total_score', 'score_type', 'ai_feedback_structured')): + m = metrics[r.user_id] + m['train_count'] += 1 + if r.status == 'completed': + m['completed'] += 1 + n = _norm_total(r.total_score, r.score_type) + if n is not None: + m['score_sum'] += n + m['score_n'] += 1 + for dim, score, mx in _dimension_scores(r): + std = DIMENSION_MAP.get(dim) + if std and score is not None and mx and float(mx) > 0: + m['dim'][std].append(float(score) / float(mx) * 100) + + def value_of(i): + m = metrics[i] + if dimension == 'train_count': + return m['train_count'] + if dimension == 'completed': + return m['completed'] + if dimension in STANDARD_DIMS: + vals = m['dim'].get(dimension) + return round(sum(vals) / len(vals), 1) if vals else None + return round(m['score_sum'] / m['score_n'], 1) if m['score_n'] else None + + rows = [{'student_id': i, 'real_name': info[i][0], 'department': info[i][1], + 'value': value_of(i), 'train_count': metrics[i]['train_count']} for i in ids] + rows.sort(key=lambda x: (x['value'] is None, -(x['value'] or 0))) + for idx, row in enumerate(rows, 1): + row['rank'] = idx + return {'dimension': dimension, 'count': len(rows), 'ranking': rows} diff --git a/apps/cms/urls.py b/apps/cms/urls.py index 48f79ba..37a9e4a 100644 --- a/apps/cms/urls.py +++ b/apps/cms/urls.py @@ -1,9 +1,18 @@ from django.urls import path, include -# apps/cms 是 CMS 薄壳：聚合各领域 app 的 /api/cms/ 路由 + 未来的跨领域看板。 +from . import stats, training + +# apps/cms 是 CMS 薄壳：聚合各领域 app 的 /api/cms/ 路由 + 跨领域看板（概览）。 # 机构管理已迁至 organization 域（apps/organization），在此聚合。 urlpatterns = [ path('', include('apps.organization.urls')), # 机构、科室 path('', include('apps.user.cms_urls')), # 用户 path('', include('apps.case.cms_urls')), # 病例库 + AI 病例生成 + # 训练记录（CMS-TRN-1 超管全平台只读） + path('training-records/', training.training_records, name='cms-training-records'), + # 各角色概览大屏（第八章） + path('stats/overview/', stats.platform_overview, name='cms-stats-overview'), + path('stats/hospital/overview/', stats.hospital_overview, name='cms-stats-hospital-overview'), + path('stats/content/overview/', stats.content_overview, name='cms-stats-content-overview'), + path('stats/teaching/overview/', stats.teaching_overview, name='cms-stats-teaching-overview'), ] diff --git a/apps/training/models.py b/apps/training/models.py index 7a142ad..12db1b9 100644 --- a/apps/training/models.py +++ b/apps/training/models.py @@ -85,6 +85,29 @@ class TrainingRecord(BaseModel): return f"{self.user.username} - {self.case.title}" +class TrainingSession(BaseModel): + """训练会话表（只读，fastapi 属主，managed=False）。 + + 用于**平台级**「发起/完成」统计：`COUNT(*)`=累计发起（含未完成），`completed_at` 非空=已完成。 + ⚠️ 只有 `external_user_id`(varchar)、**无 `user.id`、无机构外键** → 不能按机构/科室聚合； + 需按机构聚合的训练统计一律改用 `TrainingRecord`（有 `user_id`）。 + """ + id = models.BigAutoField(primary_key=True) + case_id = models.BigIntegerField('病例ID', null=True, blank=True) + case_type = models.CharField('病例类型', max_length=30, blank=True) + training_mode = models.CharField('训练模式', max_length=50, blank=True) + status = models.CharField('状态', max_length=30, blank=True) + external_user_id = models.CharField('宿主系统用户ID', max_length=128, blank=True) + started_at = models.DateTimeField('开始时间', null=True, blank=True) + completed_at = models.DateTimeField('完成时间', null=True, blank=True) + + class Meta: + managed = False + db_table = 'training_session' + verbose_name = '训练会话' + verbose_name_plural = '训练会话' + + class TrainingScoreDetail(BaseModel): """评分明细表（只读，managed=False，fastapi 属主）。 diff --git a/apps/user/cms.py b/apps/user/cms.py index c91d37b..a0e24c4 100644 --- a/apps/user/cms.py +++ b/apps/user/cms.py @@ -324,6 +324,11 @@ class CmsStudentViewSet(viewsets.ReadOnlyModelViewSet): 数据范围 = `teacher_student_relation` 中 `teacher=当前医生 且 status=1` 的学生。 非名下学生访问详情时不在 queryset 内 → 404。 + + 附带教能力相关接口（均自动收口名下学生）： + - CMS-TEA-3 教学工具-训练记录：`GET students/training-records/` + - CMS-TEA-4 能力画像：`GET students/{id}/competency/` + - CMS-TEA-5 排行榜：`GET students/ranking/` """ permission_classes = [IsAuthenticated, IsTeacher] serializer_class = CmsStudentSerializer @@ -340,3 +345,26 @@ class CmsStudentViewSet(viewsets.ReadOnlyModelViewSet): .select_related('institution', 'department') .order_by('-created_at') ) + + @extend_schema(summary='CMS-TEA-3 教学工具-训练记录（名下学生）', tags=['CMS-我的学生']) + @action(detail=False, methods=['get'], url_path='training-records') + def training_records(self, request): + """名下学生训练记录列表（与超管 CMS-TRN-1 共用构建逻辑，范围限名下学生）。""" + from apps.cms.training import build_training_records + from apps.training.models import TrainingRecord + student_ids = list(self.get_queryset().values_list('id', flat=True)) + base = TrainingRecord.objects.filter(user_id__in=student_ids) + return build_training_records(base, request) + + @extend_schema(summary='CMS-TEA-5 学生排行榜（名下学生多维度排名）', tags=['CMS-我的学生']) + @action(detail=False, methods=['get'], url_path='ranking') + def ranking(self, request): + from apps.cms.training import build_ranking + return Response(build_ranking(self.get_queryset(), request.query_params.get('dimension'))) + + @extend_schema(summary='CMS-TEA-4 学生能力画像（雷达 + 概览）', tags=['CMS-我的学生']) + @action(detail=True, methods=['get'], url_path='competency') + def competency(self, request, pk=None): + """单个名下学生的能力画像（越权对象不在 queryset → 404）。""" + from apps.cms.training import student_competency + return Response(student_competency(self.get_object())) diff --git a/test/swagger_cms_stats.py b/test/swagger_cms_stats.py new file mode 100644 index 0000000..96e51c2 --- /dev/null +++ b/test/swagger_cms_stats.py @@ -0,0 +1,148 @@ +"""Swagger Try-it-out 等效脚本：CMS 各角色概览大屏（第八章 CMS-STATS-*）。 + + CMS-STATS-SUP-1 平台总览 / HOS-1 医院驾驶舱 / CNT-1 内容概览 / TEA-1 教学概览。 +为产出**真实出参示例**，先在示例机构内播种一份连贯数据（机构/4 类管理员/2 学生/师生关系/ +2 病例/3 条训练记录），调用后把完整 JSON 落到 logs/swagger-cms-stats-examples.json，最后清理。 +运行方式：.venv\\Scripts\\python.exe test/swagger_cms_stats.py +前提：Django dev server 已在 http://127.0.0.1:8000 运行，Redis 已启动。 +""" +import sys +import json +import subprocess +from pathlib import Path + +import requests + +sys.stdout.reconfigure(encoding='utf-8') +sys.stderr.reconfigure(encoding='utf-8') + +BASE = 'http://127.0.0.1:8000' +PYTHON = r'D:\01Agent\medical_training\.venv\Scripts\python.exe' +CWD = r'D:\01Agent\medical_training' +PASS, FAIL = 'PASS', 'FAIL' +results = [] +EXAMPLES = {} + +OVERVIEW = '/api/cms/stats/overview/' +HOSPITAL = '/api/cms/stats/hospital/overview/' +CONTENT = '/api/cms/stats/content/overview/' +TEACHING = '/api/cms/stats/teaching/overview/' + +PHONES = ['13700008001', '13700008002', '13700008003', '13700008004', '13700008005', '13700008006'] + + +def log(api_id, method, url, expected, actual, detail=''): + exp = expected if isinstance(expected, (list, tuple)) else [expected] + status = PASS if actual in exp else FAIL + results.append((api_id, status)) + print(f' {status} {api_id:<18} {method:<5} {url:<40} expect={str(expected):<9} got={actual} {detail}') + + +def django_eval(code): + pre = 'import django, os; os.environ.setdefault("DJANGO_SETTINGS_MODULE","config.settings"); django.setup()\n' + p = subprocess.run([PYTHON, '-c', pre + code], capture_output=True, text=True, cwd=CWD) + if p.returncode != 0: + print('[django_eval ERROR]', p.stderr[-1500:]) + return p.stdout.strip() + + +def jb(r): + return r.json() if r.headers.get('content-type', '').startswith('application/json') else None + + +SEED = r''' +from apps.user.models import User, Institution, Department, TeacherStudentRelation +from apps.case.models import CaseBase +from rest_framework_simplejwt.tokens import RefreshToken +from django.db import connection +from django.utils import timezone +import json +PH = ["13700008001","13700008002","13700008003","13700008004","13700008005","13700008006"] +User.all_objects.filter(phone__in=PH).delete() +CaseBase.all_objects.filter(title__startswith="概览示例-").hard_delete() +with connection.cursor() as c: + c.execute("DELETE FROM training_record WHERE external_user_id='SWGSTAT'") +inst,_ = Institution.objects.get_or_create(code="SWG_STAT", defaults={"name":"概览示例院","type":"hospital","level":"三甲"}) +dept,_ = Department.objects.get_or_create(name="心内科", defaults={"category":"临床"}) +mk = lambda ph,nm,rt: User.objects.create_user(username=ph,password=None,phone=ph,real_name=nm,role_type=rt,institution=inst,status=1) +su=mk(PH[0],"示例超管","super_admin"); hu=mk(PH[1],"示例院管","hospital_admin") +cu=mk(PH[2],"示例内容","content_admin"); du=mk(PH[3],"示例带教","doctor") +s1=mk(PH[4],"学生甲","student"); s2=mk(PH[5],"学生乙","student") +s1.department=dept; s1.save(update_fields=["department"]); s2.department=dept; s2.save(update_fields=["department"]) +TeacherStudentRelation.objects.create(teacher=du,student=s1,status=1) +TeacherStudentRelation.objects.create(teacher=du,student=s2,status=1) +c1=CaseBase.objects.create(title="概览示例-急性心梗",case_type="traditional",institution=inst,department=dept,publish_status=2,difficulty="medium",created_by=cu) +c2=CaseBase.objects.create(title="概览示例-稳定型心绞痛",case_type="traditional",institution=inst,department=dept,publish_status=1,difficulty="easy",created_by=cu) +DIMS=[{"dimension":"信息获取","score":18,"max_score":20},{"dimension":"分析推理","score":17,"max_score":20},{"dimension":"处置决策","score":8,"max_score":10},{"dimension":"沟通人文","score":7,"max_score":10},{"dimension":"临床整合","score":8,"max_score":10}] +now=timezone.now() +rows=[(s1.id,c1.id,88),(s2.id,c1.id,76),(s1.id,c2.id,92)] +with connection.cursor() as c: + for uid,cid,sc in rows: + c.execute("INSERT INTO training_record (user_id,case_id,training_mode,case_type,status,total_score,duration_seconds,end_time,start_time,score_type,evaluation_level,feedback,thinking_chain,diagnosis_path,wrong_points,missed_questions,recommendation_result,osce_station_score,emotion_analysis,ai_feedback_structured,external_user_id,created_at,updated_at) VALUES (%s,%s,'practice','traditional','completed',%s,1800,%s,%s,'percentage','good','','','','[]','[]','{}','{}','{}',%s,'SWGSTAT',%s,%s)",[uid,cid,sc,now,now,json.dumps({"dimension_scores":DIMS},ensure_ascii=False),now,now]) +print("|".join(str(RefreshToken.for_user(x).access_token) for x in [su,hu,cu,du])) +''' + +CLEANUP = r''' +from apps.user.models import User +from apps.case.models import CaseBase +from django.db import connection +PH = ["13700008001","13700008002","13700008003","13700008004","13700008005","13700008006"] +with connection.cursor() as c: + c.execute("DELETE FROM training_record WHERE external_user_id='SWGSTAT'") +CaseBase.all_objects.filter(title__startswith="概览示例-").hard_delete() +User.all_objects.filter(phone__in=PH).delete() +print("cleaned") +''' + +print('\n[准备] 播种示例机构数据（机构/4 管理员/2 学生/师生关系/2 病例/3 训练记录）...') +out = django_eval(SEED) +su, hu, cu, du = out.split('|') +SU = {'Authorization': f'Bearer {su}'} +HU = {'Authorization': f'Bearer {hu}'} +CU = {'Authorization': f'Bearer {cu}'} +DU = {'Authorization': f'Bearer {du}'} +print('[准备] 完成\n') + +print('=' * 100) +print(' CMS 各角色概览大屏 Swagger Try-it-out') +print('=' * 100) + +# 权限 +log('anon-401', 'GET', OVERVIEW, 401, requests.get(f'{BASE}{OVERVIEW}').status_code) +log('SUP-role403', 'GET', OVERVIEW, 403, requests.get(f'{BASE}{OVERVIEW}', headers=HU).status_code) +log('HOS-role403', 'GET', HOSPITAL, 403, requests.get(f'{BASE}{HOSPITAL}', headers=SU).status_code) +log('CNT-role403', 'GET', CONTENT, 403, requests.get(f'{BASE}{CONTENT}', headers=DU).status_code) +log('TEA-role403', 'GET', TEACHING, 403, requests.get(f'{BASE}{TEACHING}', headers=CU).status_code) + + +def call(api_id, url, headers): + r = requests.get(f'{BASE}{url}', headers=headers) + b = jb(r) + log(api_id, 'GET', url, 200, r.status_code) + if r.status_code == 200: + EXAMPLES[api_id] = {'request': {'method': 'GET', 'url': url, + 'headers': {'Authorization': 'Bearer '}}, + 'response': b} + return b + + +call('CMS-STATS-SUP-1', OVERVIEW, SU) +call('CMS-STATS-HOS-1', HOSPITAL, HU) +call('CMS-STATS-CNT-1', CONTENT, CU) +call('CMS-STATS-TEA-1', TEACHING, DU) + +# 落盘完整示例 +exf = Path(CWD) / 'logs' / 'swagger-cms-stats-examples.json' +exf.parent.mkdir(exist_ok=True) +exf.write_text(json.dumps(EXAMPLES, ensure_ascii=False, indent=2), encoding='utf-8') +print(f'\n[示例] 完整出参已写入 {exf}') + +django_eval(CLEANUP) + +print('=' * 100) +total = len(results); passed = sum(1 for _, s in results if s == PASS); failed = total - passed +print(f' 总计: {total} | 通过: {passed} | 失败: {failed}') +if failed: + print(' 失败:', [a for a, s in results if s == FAIL]); sys.exit(1) +print(' ALL PASSED — CMS 各角色概览大屏接口验证通过！') +sys.exit(0) diff --git a/test/test_cms_stats.py b/test/test_cms_stats.py new file mode 100644 index 0000000..e1c4a1b --- /dev/null +++ b/test/test_cms_stats.py @@ -0,0 +1,180 @@ +"""CMS 各角色概览大屏聚合接口测试（第八章 / CMS-STATS-*）。 + +`training_record` / `training_session` 是 managed=False 只读表，迁移无占位结构， +故在 setUpClass 按真实列建表（仅测试库），用 TransactionTestCase 允许 DDL。 +权限类用例（401/403）在权限层即返回、不触表，放普通 TestCase。 +""" +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, TeacherStudentRelation +from apps.case.models import CaseBase +from .conftest import ( + CacheTestCase, create_test_user, get_auth_client, ensure_institution, +) + +OVERVIEW = '/api/cms/stats/overview/' +HOSPITAL = '/api/cms/stats/hospital/overview/' +CONTENT = '/api/cms/stats/content/overview/' +TEACHING = '/api/cms/stats/teaching/overview/' + +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 +""" + +CREATE_TS = """ +CREATE TABLE training_session ( + id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, + case_id BIGINT NULL, case_type VARCHAR(30) NULL, training_mode VARCHAR(50) NULL, + status VARCHAR(30) NULL, external_user_id VARCHAR(128) NULL, + started_at DATETIME NULL, completed_at DATETIME NULL, + created_at DATETIME NULL, updated_at DATETIME NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 +""" + +DIMS = [{'dimension': '信息获取', 'score': 8, 'max_score': 10}, + {'dimension': '分析推理', 'score': 9, 'max_score': 10}, + {'dimension': '处置决策', 'score': 8, 'max_score': 10}, + {'dimension': '沟通人文', 'score': 7, 'max_score': 10}, + {'dimension': '临床整合', 'score': 8, 'max_score': 10}] + + +class StatsOverviewTest(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('DROP TABLE IF EXISTS training_session') + c.execute(CREATE_TR) + c.execute(CREATE_TS) + c.execute('SET FOREIGN_KEY_CHECKS=1') + + def _record(self, user_id, case_id, total_score, status='completed'): + now = timezone.now() + with connection.cursor() as c: + c.execute( + "INSERT INTO training_record (user_id, case_id, training_mode, case_type, status, " + "total_score, duration_seconds, end_time, start_time, score_type, evaluation_level, " + "ai_feedback_structured, created_at, updated_at) " + "VALUES (%s,%s,'practice','traditional',%s,%s,1800,%s,%s,'percentage','good',%s,%s,%s)", + [user_id, case_id, status, total_score, now, now, + json.dumps({'dimension_scores': DIMS}, ensure_ascii=False), now, now], + ) + + def _session(self, case_id, completed=True): + now = timezone.now() + with connection.cursor() as c: + c.execute( + "INSERT INTO training_session (case_id, case_type, status, created_at, updated_at, " + "started_at, completed_at) VALUES (%s,'traditional',%s,%s,%s,%s,%s)", + [case_id, 'completed' if completed else 'inquiry', now, now, now, + now if completed else None], + ) + + def setUp(self): + cache.clear() + with connection.cursor() as c: + c.execute('DELETE FROM training_record') + c.execute('DELETE FROM training_session') + self.instA = ensure_institution(name='A院', code='STAT-A') + self.instB = ensure_institution(name='B院', code='STAT-B') + self.dept = Department.objects.create(name='心内科', category='临床') + self.caseA1 = CaseBase.objects.create(title='病例甲', case_type='traditional', + institution=self.instA, department=self.dept, publish_status=1) + self.caseA2 = CaseBase.objects.create(title='病例乙', case_type='traditional', + institution=self.instA, department=self.dept, publish_status=0) + self.superu = create_test_user(phone='13900000001', role_type='super_admin', institution=self.instA) + self.hosp = create_test_user(phone='13900000002', role_type='hospital_admin', institution=self.instA) + self.content = create_test_user(phone='13900000003', role_type='content_admin', institution=self.instA) + self.doctor = create_test_user(phone='13900000004', role_type='doctor', institution=self.instA) + self.s1 = create_test_user(phone='13900000005', role_type='student', institution=self.instA) + self.s2 = create_test_user(phone='13900000006', role_type='student', institution=self.instA) + TeacherStudentRelation.objects.create(teacher=self.doctor, student=self.s1, status=1) + TeacherStudentRelation.objects.create(teacher=self.doctor, student=self.s2, status=1) + # 4 个会话（3 完成）、2 条训练记录（s1=90、s2=80，均在 A 院病例甲） + self._session(self.caseA1.id, completed=True) + self._session(self.caseA1.id, completed=True) + self._session(self.caseA1.id, completed=True) + self._session(self.caseA1.id, completed=False) + self._record(self.s1.id, self.caseA1.id, 90) + self._record(self.s2.id, self.caseA1.id, 80) + + # ── 权限 ───────────────────────────────────────────────────────────── + def test_permissions(self): + anon = APIClient() + for url in (OVERVIEW, HOSPITAL, CONTENT, TEACHING): + self.assertEqual(anon.get(url).status_code, 401, url) + # 角色错配 → 403 + self.assertEqual(get_auth_client(self.hosp).get(OVERVIEW).status_code, 403) # 非超管 + self.assertEqual(get_auth_client(self.superu).get(HOSPITAL).status_code, 403) # 非医院管理员 + self.assertEqual(get_auth_client(self.doctor).get(CONTENT).status_code, 403) # 非内容管理员 + self.assertEqual(get_auth_client(self.content).get(TEACHING).status_code, 403) # 非带教医生 + + # ── 8.1 平台总览 ───────────────────────────────────────────────────── + def test_platform_overview(self): + d = get_auth_client(self.superu).get(OVERVIEW).json() + self.assertEqual(d['kpi']['institution_count'], 2) + self.assertEqual(d['kpi']['user_total'], 6) # 6 个 status=1 用户 + self.assertEqual(d['core']['train_total'], 4) # 4 个会话 + self.assertEqual(d['core']['complete_rate'], 75.0) # 3/4 + self.assertEqual(d['core']['avg_score'], 85.0) # avg(90,80) + self.assertEqual(d['case_asset']['total'], 2) + self.assertEqual(set(x['dimension'] for x in d['competency']['radar']) + if 'competency' in d else + {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}, + {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}) + + # ── 8.2 医院驾驶舱 ─────────────────────────────────────────────────── + def test_hospital_overview(self): + d = get_auth_client(self.hosp).get(HOSPITAL).json() + self.assertEqual(d['profile']['name'], 'A院') + self.assertEqual(d['summary']['student_count'], 2) + self.assertEqual(d['summary']['doctor_count'], 1) + self.assertEqual(d['summary']['train_total'], 2) # s1+s2 记录 + self.assertEqual(d['summary']['avg_score'], 85.0) + self.assertEqual(d['case_asset']['total'], 2) + self.assertEqual(set(x['dimension'] for x in d['competency']['radar']), + {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}) + + # ── 8.3 内容概览 ───────────────────────────────────────────────────── + def test_content_overview(self): + d = get_auth_client(self.content).get(CONTENT).json() + self.assertEqual(d['summary']['case_total'], 2) + self.assertEqual(d['summary']['pending_publish'], 1) # caseA1 publish_status=1 + self.assertEqual(d['summary']['train_total'], 2) + + # ── 8.4 教学概览 ───────────────────────────────────────────────────── + def test_teaching_overview(self): + d = get_auth_client(self.doctor).get(TEACHING).json() + self.assertEqual(d['overview']['student_count'], 2) + self.assertEqual(len(d['students']), 2) + byid = {s['id']: s for s in d['students']} + self.assertEqual(byid[self.s1.id]['avg_score'], 90.0) + self.assertEqual(byid[self.s2.id]['avg_score'], 80.0) + self.assertEqual(byid[self.s1.id]['most_trained_type'], 'traditional') + # 任务相关无数据源 → None + self.assertIsNone(byid[self.s1.id]['pending_tasks']) + self.assertIsNone(d['overview']['task_summary']) diff --git a/test/test_cms_training.py b/test/test_cms_training.py new file mode 100644 index 0000000..abacdba --- /dev/null +++ b/test/test_cms_training.py @@ -0,0 +1,235 @@ +"""CMS 训练记录 + 带教能力相关接口测试。 + +- CMS-TRN-1 超管训练记录列表 GET /api/cms/training-records/ +- CMS-TEA-3 教学工具-训练记录 GET /api/cms/students/training-records/ +- CMS-TEA-4 学生能力画像 GET /api/cms/students/{id}/competency/ +- CMS-TEA-5 学生排行榜 GET /api/cms/students/ranking/ + +training_record 是 managed=False 只读表，迁移占位结构缺新列，故在 setUpClass 里按真实列 +重建该表（仅测试库），用 TransactionTestCase 允许 DDL（与 test_mobile_training_stats 一致）。 +""" +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, User +from apps.case.models import CaseBase +from .conftest import ( + create_test_user, get_auth_client, ensure_institution, + create_teacher_student_relation, +) + +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 +""" + +SUP_URL = '/api/cms/training-records/' +TEA_TR_URL = '/api/cms/students/training-records/' +RANK_URL = '/api/cms/students/ranking/' + + +def _dim(name, score, mx): + return {'dimension': name, 'score': score, 'max_score': mx} + + +# 各学生维度评分（得分率见注释） +DIMS_A = [_dim('信息获取', 20, 25), _dim('分析推理', 18, 20), _dim('处置决策', 9, 10), + _dim('沟通人文', 7, 10), _dim('临床整合', 8, 10)] # 80/90/90/70/80 +DIMS_B = [_dim('信息获取', 20, 25), _dim('分析推理', 16, 20), _dim('处置决策', 8, 10), + _dim('沟通人文', 8, 10), _dim('临床整合', 8, 10)] # 全 80 +DIMS_C = [_dim('信息获取', 15, 25), _dim('分析推理', 12, 20), _dim('处置决策', 6, 10), + _dim('沟通人文', 6, 10), _dim('临床整合', 6, 10)] # 全 60 + + +class CmsTrainingTest(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, total_score, dims, status='completed', + score_type='percentage', case_type='diagnosis_treatment'): + now = timezone.now() + with connection.cursor() as c: + c.execute( + "INSERT INTO training_record (user_id, case_id, training_mode, case_type, status, " + "total_score, duration_seconds, end_time, start_time, score_type, evaluation_level, " + "ai_feedback_structured, created_at, updated_at) " + "VALUES (%s,%s,'practice',%s,%s,%s,%s,%s,%s,%s,'good',%s,%s,%s)", + [user_id, self.case.id, case_type, status, total_score, 1800, now, now, score_type, + json.dumps({'dimension_scores': dims}, ensure_ascii=False), 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='TR-H1') + self.dept = Department.objects.create(name='心内科', category='临床') + self.case = CaseBase.objects.create(title='急性心梗', case_type='traditional', + department=self.dept, institution=self.inst) + self.sup = create_test_user(phone='13970000001', role_type='super_admin', institution=self.inst) + self.doc = create_test_user(phone='13970000002', real_name='张医生', + role_type='doctor', institution=self.inst) + self.other_doc = create_test_user(phone='13970000003', role_type='doctor', institution=self.inst) + self.stu1 = create_test_user(phone='13970000011', real_name='学生甲', + role_type='student', institution=self.inst) + self.stu1.department = self.dept + self.stu1.save(update_fields=['department']) + self.stu2 = create_test_user(phone='13970000012', real_name='学生乙', + role_type='student', institution=self.inst) + self.stu_other = create_test_user(phone='13970000013', real_name='别家学生', + role_type='student', institution=self.inst) + create_teacher_student_relation(self.doc, self.stu1, status=1) + create_teacher_student_relation(self.doc, self.stu2, status=1) + create_teacher_student_relation(self.other_doc, self.stu_other, status=1) + + # 训练记录：stu1 两条完成 + 一条进行中；stu2 一条完成；stu_other 一条完成 + self._insert(self.stu1.id, 90, DIMS_A) + self._insert(self.stu1.id, 80, DIMS_B) + self._insert(self.stu1.id, None, [], status='in_progress') + self._insert(self.stu2.id, 60, DIMS_C) + self._insert(self.stu_other.id, 70, DIMS_B) + + self.sup_client = get_auth_client(self.sup) + self.doc_client = get_auth_client(self.doc) + + # ── CMS-TRN-1 超管训练记录 ────────────────────────────────────────────────── + def test_trn_unauthenticated_401(self): + self.assertEqual(APIClient().get(SUP_URL).status_code, 401) + + def test_trn_non_super_403(self): + resp = self.doc_client.get(SUP_URL) + self.assertEqual(resp.status_code, 403, resp.content) + self.assertEqual(resp.json()['code'], 'CMS_PERMISSION_DENIED') + + def test_trn_super_list_all(self): + resp = self.sup_client.get(SUP_URL) + self.assertEqual(resp.status_code, 200, resp.content) + data = resp.json() + self.assertEqual(data['count'], 5) # 全平台所有记录 + self.assertEqual(data['summary']['total'], 5) + self.assertEqual(data['summary']['completed'], 4) + self.assertEqual(data['summary']['avg_score'], 75.0) # avg(90,80,60,70) + item = data['results'][0] + for k in ('record_id', 'user_id', 'user_name', 'case_title', + 'score', 'score_normalized', 'status', 'trained_at'): + self.assertIn(k, item) + + def test_trn_filter_by_user_and_status(self): + resp = self.sup_client.get(SUP_URL, {'user_id': self.stu1.id, 'status': 'completed'}) + data = resp.json() + self.assertEqual(data['count'], 2) + self.assertTrue(all(r['user_id'] == self.stu1.id for r in data['results'])) + + def test_trn_filter_score_range(self): + resp = self.sup_client.get(SUP_URL, {'min_score': 75}) + data = resp.json() + # 归一分≥75 的完成记录：90、80（stu1）、未含 60/70 + self.assertEqual(data['count'], 2) + + def test_trn_search_by_student_name(self): + resp = self.sup_client.get(SUP_URL, {'search': '学生甲'}) + data = resp.json() + self.assertEqual(data['count'], 3) # stu1 全部记录（2 完成 + 1 进行中） + + def test_trn_five_point_normalized(self): + self._insert(self.stu2.id, 4, DIMS_B, score_type='five_point') # 4×20=80 + resp = self.sup_client.get(SUP_URL, {'user_id': self.stu2.id, 'min_score': 80}) + data = resp.json() + self.assertEqual(data['count'], 1) + self.assertEqual(data['results'][0]['score'], 4.0) + self.assertEqual(data['results'][0]['score_normalized'], 80.0) + + # ── CMS-TEA-3 教学工具-训练记录（名下学生）────────────────────────────────── + def test_tea_tr_scope_own_students(self): + resp = self.doc_client.get(TEA_TR_URL) + self.assertEqual(resp.status_code, 200, resp.content) + data = resp.json() + self.assertEqual(data['count'], 4) # stu1(3) + stu2(1)，不含 stu_other + uids = {r['user_id'] for r in data['results']} + self.assertNotIn(self.stu_other.id, uids) + self.assertEqual(data['summary']['completed'], 3) + + def test_tea_tr_non_teacher_403(self): + resp = self.sup_client.get(TEA_TR_URL) + self.assertEqual(resp.status_code, 403, resp.content) + + def test_tea_tr_filter_by_student(self): + resp = self.doc_client.get(TEA_TR_URL, {'user_id': self.stu2.id}) + self.assertEqual(resp.json()['count'], 1) + + # ── CMS-TEA-4 能力画像 ────────────────────────────────────────────────────── + def test_competency_own_student(self): + resp = self.doc_client.get(f'/api/cms/students/{self.stu1.id}/competency/') + self.assertEqual(resp.status_code, 200, resp.content) + d = resp.json() + self.assertEqual(d['student_id'], self.stu1.id) + self.assertEqual(d['completed_cases'], 2) + self.assertEqual(d['avg_score'], 85.0) # avg(90,80) + self.assertEqual(d['diagnosis_accuracy'], 85) # 分析推理 avg(90,80) + radar = {x['dimension']: x['score'] for x in d['radar']} + self.assertEqual(set(radar), {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}) + self.assertEqual(radar['信息获取'], 80) + self.assertEqual(radar['沟通人文'], 75) + self.assertEqual(d['weak_dimensions'], ['沟通人文']) + + def test_competency_other_student_404(self): + resp = self.doc_client.get(f'/api/cms/students/{self.stu_other.id}/competency/') + self.assertEqual(resp.status_code, 404, resp.content) + + # ── CMS-TEA-5 排行榜 ──────────────────────────────────────────────────────── + def test_ranking_by_score_default(self): + resp = self.doc_client.get(RANK_URL) + self.assertEqual(resp.status_code, 200, resp.content) + d = resp.json() + self.assertEqual(d['dimension'], 'score') + self.assertEqual(d['count'], 2) + self.assertEqual(d['ranking'][0]['student_id'], self.stu1.id) # 85 > 60 + self.assertEqual(d['ranking'][0]['rank'], 1) + self.assertEqual(d['ranking'][1]['student_id'], self.stu2.id) + + def test_ranking_by_train_count(self): + resp = self.doc_client.get(RANK_URL, {'dimension': 'train_count'}) + d = resp.json() + self.assertEqual(d['dimension'], 'train_count') + self.assertEqual(d['ranking'][0]['student_id'], self.stu1.id) # 3 > 1 + self.assertEqual(d['ranking'][0]['value'], 3) + + def test_ranking_by_competency_dimension(self): + resp = self.doc_client.get(RANK_URL, {'dimension': '信息获取'}) + d = resp.json() + self.assertEqual(d['dimension'], '信息获取') + self.assertEqual(d['ranking'][0]['student_id'], self.stu1.id) # 80 > 60 + + def test_ranking_invalid_dimension_falls_back_to_score(self): + resp = self.doc_client.get(RANK_URL, {'dimension': '不存在'}) + self.assertEqual(resp.json()['dimension'], 'score') + + def test_ranking_non_teacher_403(self): + self.assertEqual(self.sup_client.get(RANK_URL).status_code, 403) diff --git a/test/test_mobile_case_list.py b/test/test_mobile_case_list.py new file mode 100644 index 0000000..d59a032 --- /dev/null +++ b/test/test_mobile_case_list.py @@ -0,0 +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])