"""CMS 用户管理 + 师生关系管理 + 带教医生「我的学生」。 用户管理（超管全平台 / 医院管理员本院）：CMS-USER-1~8、CMS-HUSER-1~3。 师生关系管理（医院管理员）：CMS-REL-1~4。 带教医生「我的学生」（doctor 名下学生，只读）：CMS-TEA-1~2。 """ import re from django.db import IntegrityError from rest_framework import viewsets, filters, status from rest_framework.decorators import action from rest_framework.parsers import MultiPartParser, FormParser from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework import serializers from django_filters.rest_framework import DjangoFilterBackend from drf_spectacular.utils import extend_schema, extend_schema_view from config.exceptions import AppError from apps.cms.permissions import IsSuperOrHospitalAdmin, IsTeacher, is_super from apps.common.excel import xlsx_response, rows_from_xlsx from .models import User, Institution, TeacherStudentRelation from .utils.jwt_redis import invalidate_user_tokens ALL_ROLES = ('super_admin', 'hospital_admin', 'content_admin', 'doctor', 'student') # 医院管理员可管理 / 创建的角色（本院：医生、学生、内容管理员） # —— 医院管理员可给本院工作人员授予内容管理员权限，故内容管理员也纳入其人员管理范围。 HOSPITAL_ADMIN_ROLES = ('doctor', 'student', 'content_admin') ROLE_LABEL_BY_CODE = { 'super_admin': '超级管理员', 'hospital_admin': '医院管理员', 'content_admin': '内容管理员', 'doctor': '医生', 'student': '学生', } ROLE_CODE_BY_LABEL = { '超级管理员': 'super_admin', '医院管理员': 'hospital_admin', '内容管理员': 'content_admin', '医生': 'doctor', '带教医生': 'doctor', '带教老师': 'doctor', '学生': 'student', '医学生': 'student', } USER_IMPORT_HEADERS = ['手机号', '姓名', '角色', '机构编码'] USER_EXPORT_HEADERS = ['ID', '手机号', '姓名', '角色', '机构', '状态'] def _resolve_role(value): value = (value or '').strip() if value in ALL_ROLES: return value return ROLE_CODE_BY_LABEL.get(value) # ── 序列化器 ────────────────────────────────────────────────────────────────── class CmsUserSerializer(serializers.ModelSerializer): """读取（列表 / 详情）。""" institution_name = serializers.CharField(source='institution.name', read_only=True, default=None) role_label = serializers.SerializerMethodField() class Meta: model = User fields = [ 'id', 'username', 'real_name', 'phone', 'role_type', 'role_label', 'institution', 'institution_name', 'gender', 'title_name', 'major', 'training_stage', 'status', 'created_at', ] def get_role_label(self, obj): return ROLE_LABEL_BY_CODE.get(obj.role_type, obj.role_type) class CmsUserWriteSerializer(serializers.ModelSerializer): """新增 / 编辑（role 必填）。""" phone = serializers.CharField(max_length=20) class Meta: model = User # 科室不在 CMS 维护（科室仅用于病例分类，用户在移动端自选学习科室） fields = [ 'real_name', 'phone', 'role_type', 'institution', 'gender', 'title_name', 'major', 'training_stage', 'status', ] def validate_phone(self, value): value = (value or '').strip() if not re.match(r'^1[3-9]\d{9}$', value): raise AppError('CMS_VALIDATION_ERROR', '手机号格式不合法', status_code=400) # 含已停用账号：手机号唯一约束对软删行仍生效，提示需复用应先恢复/换号 qs = User.all_objects.filter(phone=value) if self.instance is not None: qs = qs.exclude(pk=self.instance.pk) if qs.exists(): raise AppError('CMS_USER_PHONE_EXISTS', '手机号已存在（含已停用账号），如需复用请先恢复或更换手机号', status_code=400) return value def validate(self, attrs): """新增：角色/机构/姓名必填。 编辑：局部更新——角色/机构不必每次带（不传保持原值），但**若传则不能为空/非法**。 医院管理员：角色仅限 doctor/student，机构强制本院。 """ actor = self.context['request'].user actor_is_super = is_super(actor) is_create = self.instance is None # 角色：新增必填；编辑时若出现在请求里则校验非空且合法 if is_create or 'role_type' in attrs: role = attrs.get('role_type') if not role: raise AppError('CMS_VALIDATION_ERROR', '角色（role_type）不能为空', status_code=400) if role not in ALL_ROLES: raise AppError('CMS_VALIDATION_ERROR', '角色非法', status_code=400) if not actor_is_super and role not in HOSPITAL_ADMIN_ROLES: raise AppError('CMS_ROLE_NOT_ALLOWED', '医院管理员只能管理医生 / 学生 / 内容管理员', status_code=403) # 机构 if actor_is_super: # 超管：机构必填（编辑时若传则不能为空） if is_create or 'institution' in attrs: if not attrs.get('institution'): raise AppError('CMS_VALIDATION_ERROR', '机构（institution）不能为空', status_code=400) else: # 医院管理员：机构强制为本院（忽略请求里传的机构） if not actor.institution_id: raise AppError('CMS_NO_INSTITUTION', '当前医院管理员无所属机构，无法管理用户', status_code=403) attrs['institution'] = actor.institution # 姓名：仅新增必填 if is_create and not (attrs.get('real_name') or '').strip(): raise AppError('CMS_VALIDATION_ERROR', '姓名必填', status_code=400) return attrs def create(self, validated_data): phone = validated_data['phone'] try: return User.objects.create_user( username=phone, password=f'Pass{phone}', **validated_data ) except IntegrityError: raise AppError('CMS_USER_PHONE_EXISTS', '手机号已存在（含已停用账号），如需复用请先恢复或更换手机号', status_code=400) def update(self, instance, validated_data): for key, val in validated_data.items(): setattr(instance, key, val) instance.save() return instance # ── ViewSet ────────────────────────────────────────────────────────────────── @extend_schema_view( list=extend_schema(summary='CMS-USER-1 用户列表', tags=['CMS-用户']), create=extend_schema(summary='CMS-USER-2 新增用户', tags=['CMS-用户']), retrieve=extend_schema(summary='用户详情', tags=['CMS-用户']), partial_update=extend_schema(summary='CMS-USER-3 编辑用户', tags=['CMS-用户']), destroy=extend_schema(summary='CMS-USER-4 停用用户（逻辑删除）', tags=['CMS-用户']), ) class CmsUserViewSet(viewsets.ModelViewSet): """CMS 用户管理。停用为逻辑删除（User 软删除）。 - 超级管理员：全平台、任意角色。 - 医院管理员（CMS-HUSER-*）：仅本院、仅 doctor/student（医生管理 / 医学生管理）。 前端按 `?role_type=doctor` / `?role_type=student` 区分两个页面。 """ permission_classes = [IsAuthenticated, IsSuperOrHospitalAdmin] filter_backends = [DjangoFilterBackend, filters.SearchFilter] filterset_fields = ['role_type', 'institution', 'status', 'gender'] search_fields = ['username', 'real_name', 'phone'] # 仅 GET / POST：查=GET，增删改=POST（编辑→{id}/update/，停用→{id}/disable/） http_method_names = ['get', 'post', 'head', 'options'] def get_queryset(self): qs = User.objects.select_related('institution', 'department').all().order_by('-created_at') user = self.request.user if is_super(user): return qs # 医院管理员：仅本院 + 仅医生/学生（也保证只能查/改/停用本院 doctor/student） return qs.filter(institution_id=user.institution_id, role_type__in=HOSPITAL_ADMIN_ROLES) def get_serializer_class(self): if self.action in ('create', 'update_user'): return CmsUserWriteSerializer return CmsUserSerializer def create(self, request, *args, **kwargs): write = self.get_serializer(data=request.data) write.is_valid(raise_exception=True) user = write.save() return Response(CmsUserSerializer(user).data, status=status.HTTP_201_CREATED) @extend_schema(summary='CMS-USER-3 编辑用户', tags=['CMS-用户']) @action(detail=True, methods=['post'], url_path='update') def update_user(self, request, pk=None): """编辑用户（POST 局部更新，等价旧 PATCH /{id}/）。""" instance = self.get_object() write = self.get_serializer(instance, data=request.data, partial=True) write.is_valid(raise_exception=True) user = write.save() return Response(CmsUserSerializer(user).data) @extend_schema(summary='CMS-USER-4 停用用户（逻辑删除）', tags=['CMS-用户']) @action(detail=True, methods=['post'], url_path='disable') def disable(self, request, pk=None): """停用用户（软删除，等价旧 DELETE /{id}/）。""" self.get_object().delete() return Response({'message': '已停用'}) @extend_schema(summary='CMS-USER-5 重置密码', tags=['CMS-用户']) @action(detail=True, methods=['post'], url_path='reset-password') def reset_password(self, request, pk=None): """重置为默认密码 `Pass{手机号}`（或请求体 password），并踢下线。""" user = self.get_object() new_password = (request.data.get('password') or '').strip() or f'Pass{user.phone}' user.set_password(new_password) user.save(update_fields=['password']) invalidate_user_tokens(user.id) return Response({'message': '密码重置成功', 'password': new_password}) @extend_schema(summary='CMS-USER-7 下载导入模板', tags=['CMS-用户']) @action(detail=False, methods=['get'], url_path='import-template') def import_template(self, request): return xlsx_response('用户导入模板.xlsx', USER_IMPORT_HEADERS, []) @extend_schema(summary='CMS-USER-8 导出用户', tags=['CMS-用户']) @action(detail=False, methods=['get'], url_path='export') def export(self, request): qs = self.filter_queryset(self.get_queryset()) rows = [ [u.id, u.phone, u.real_name, ROLE_LABEL_BY_CODE.get(u.role_type, u.role_type), u.institution.name if u.institution_id else '', '正常' if u.status == 1 else '禁用'] for u in qs ] return xlsx_response('用户列表.xlsx', USER_EXPORT_HEADERS, rows) @extend_schema(summary='CMS-USER-6 导入用户', tags=['CMS-用户']) @action(detail=False, methods=['post'], url_path='import', parser_classes=[MultiPartParser, FormParser]) def import_users(self, request): """Excel 批量导入用户。 - 超管：列 手机号 | 姓名 | 角色 | 机构编码（均必填）。 - 医院管理员：机构强制本院（忽略机构编码列），角色仅限 医生 / 学生。 """ actor = request.user actor_is_super = is_super(actor) if not actor_is_super and not actor.institution_id: raise AppError('CMS_NO_INSTITUTION', '当前医院管理员无所属机构，无法导入', status_code=403) file = request.FILES.get('file') if not file: raise AppError('CMS_IMPORT_FILE_REQUIRED', '请上传 .xlsx 文件（字段名 file）', status_code=400) try: rows = rows_from_xlsx(file) except Exception: raise AppError('CMS_IMPORT_BAD_FILE', '文件解析失败，请使用导入模板', status_code=400) inst_by_code = {i.code: i for i in Institution.objects.all()} success, errors = 0, [] for idx, row in enumerate(rows, start=2): # 第 2 行起为数据 phone = (row.get('手机号') or '').strip() real_name = (row.get('姓名') or '').strip() role = _resolve_role(row.get('角色')) if not re.match(r'^1[3-9]\d{9}$', phone): errors.append({'row': idx, 'reason': f'手机号格式不合法：{phone}'}); continue if not real_name: errors.append({'row': idx, 'reason': '姓名为空'}); continue if not role: errors.append({'row': idx, 'reason': f"角色非法：{row.get('角色')}"}); continue if not actor_is_super and role not in HOSPITAL_ADMIN_ROLES: errors.append({'row': idx, 'reason': '医院管理员只能导入医生 / 学生 / 内容管理员'}); continue # 机构：超管按机构编码、医院管理员强制本院 if actor_is_super: inst_code = (row.get('机构编码') or '').strip() if not inst_code: errors.append({'row': idx, 'reason': '机构编码为空'}); continue inst = inst_by_code.get(inst_code) if inst is None: errors.append({'row': idx, 'reason': f'机构编码不存在：{inst_code}'}); continue else: inst = actor.institution if User.all_objects.filter(phone=phone).exists(): errors.append({'row': idx, 'reason': '手机号已存在'}); continue try: User.objects.create_user( username=phone, password=f'Pass{phone}', phone=phone, real_name=real_name, role_type=role, institution=inst, status=1, ) success += 1 except IntegrityError: errors.append({'row': idx, 'reason': '手机号已存在'}) return Response({ 'total': len(rows), 'success': success, 'failed': len(errors), 'errors': errors, }) # ── 带教医生「我的学生」（CMS-TEA-1~2）───────────────────────────────────────── class CmsStudentSerializer(CmsUserSerializer): """学生基础信息（读取）。 复用 `CmsUserSerializer` 的用户对象字段，额外补科室与训练统计，便于带教医生 在「我的学生」列表 / 详情查看学习概况。 """ department_name = serializers.CharField(source='department.name', read_only=True, default=None) class Meta(CmsUserSerializer.Meta): fields = CmsUserSerializer.Meta.fields + [ 'department', 'department_name', 'practice_years', 'total_training_count', 'total_case_count', 'current_level', ] @extend_schema_view( list=extend_schema(summary='CMS-TEA-1 我的学生列表', tags=['CMS-我的学生']), retrieve=extend_schema(summary='CMS-TEA-2 学生基础信息', tags=['CMS-我的学生']), ) class CmsStudentViewSet(viewsets.ReadOnlyModelViewSet): """带教医生「我的学生」：名下进行中（status=1）的学生，只读。 数据范围 = `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 filter_backends = [DjangoFilterBackend, filters.SearchFilter] filterset_fields = ['gender', 'status'] search_fields = ['username', 'real_name', 'phone'] def get_queryset(self): student_ids = TeacherStudentRelation.objects.filter( teacher=self.request.user, status=1 ).values_list('student_id', flat=True) return ( User.objects.filter(id__in=student_ids) .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()))