微信小程序卫生应急培训系统开发实战
1. 项目概述卫生应急培训的数字化解决方案这个基于微信小程序的卫生应急培训报名考试系统本质上是一个针对医疗卫生行业的轻量化在线教育平台。我在去年参与某市疾控中心应急培训项目时发现传统线下培训存在三个致命痛点纸质报名表易丢失、考试组织成本高、培训效果难以追踪。而微信小程序恰好能完美解决这些问题——医护人员几乎人人都有微信无需额外安装APP小程序即用即走的特性符合应急培训短平快的需求特点。系统采用PythonDjango作为后端核心主要考虑到三个因素一是Python在数据处理方面的天然优势能高效处理培训考核产生的海量数据二是Django自带的Admin后台可以快速搭建管理界面三是Python丰富的第三方库比如Pandas用于成绩分析能大幅降低开发难度。前端选择微信小程序而非H5则是看中了其原生体验和微信生态的传播便利性。关键设计原则所有功能必须适配卫生应急场景的特殊性。比如报名模块需要收集执业证书编号考试模块要支持突发情况下的大规模并发访问证书生成要符合卫健部门规范格式。2. 核心功能模块设计2.1 双Token认证体系在uni-app框架中实现的双Token机制Access Token Refresh Token是这个系统的安全基石。具体流程是用户首次登录时后端生成一对Token# 使用PyJWT生成token示例 import jwt from datetime import datetime, timedelta def generate_tokens(user_id): access_payload { user_id: user_id, exp: datetime.utcnow() timedelta(minutes30), type: access } refresh_payload { user_id: user_id, exp: datetime.utcnow() timedelta(days7), type: refresh } access_token jwt.encode(access_payload, SECRET_KEY, algorithmHS256) refresh_token jwt.encode(refresh_payload, SECRET_KEY, algorithmHS256) return access_token, refresh_token前端将双Token存储在微信小程序的缓存中// uni-app存储示例 uni.setStorageSync(access_token, res.data.access_token) uni.setStorageSync(refresh_token, res.data.refresh_token)当Access Token过期时前端自动用Refresh Token申请新Token# Django后端刷新逻辑 api_view([POST]) def refresh_token(request): try: refresh_token request.data.get(refresh_token) payload jwt.decode(refresh_token, SECRET_KEY, algorithms[HS256]) if payload[type] ! refresh: raise AuthenticationFailed(Invalid token type) new_access, _ generate_tokens(payload[user_id]) return Response({access_token: new_access}) except jwt.ExpiredSignatureError: raise AuthenticationFailed(Refresh token expired)踩坑记录初期没有区分Token类型导致安全漏洞攻击者可以用Access Token冒充Refresh Token。后来在payload中加入type字段才解决。2.2 报名信息采集设计卫生应急培训的报名表单需要特殊字段设计字段名称类型验证规则备注执业证书号文本^[A-Z0-9]{18}$必须符合国家卫健委编码规则工作单位级联选择联动省市县三级数据来自卫健系统API应急岗位单选[指挥,医疗,消杀,检测]影响后续课程内容培训历史复选框最大选择3项显示过往参加的培训项目小程序端使用uni-app的form组件实现动态表单uni-forms refform :rulesrules uni-forms-item label执业证书号 namelicense uni-easyinput v-modelform.license / /uni-forms-item uni-forms-item label应急岗位 uni-data-checkbox v-modelform.position :localdatapositions / /uni-forms-item /uni-forms后端使用Django REST framework处理提交class EnrollmentSerializer(serializers.ModelSerializer): class Meta: model Enrollment fields __all__ extra_kwargs { license_number: { validators: [RegexValidator(regex^[A-Z0-9]{18}$)] } } def validate(self, data): if data[position] 指挥 and not data.get(management_exp): raise serializers.ValidationError(指挥岗位需填写管理经验) return data2.3 考试模块关键技术2.3.1 防作弊设计题目随机算法def generate_paper(user_id): # 基于用户专业领域确定试题比例 user User.objects.get(pkuser_id) base_qs Question.objects.filter(activeTrue) # 按知识领域分配题量 domains { 基础理论: 0.3, 应急预案: 0.4, 案例分析: 0.3 } paper [] for domain, ratio in domains.items(): count int(100 * ratio) questions list(base_qs.filter(domaindomain)) paper.extend(random.sample(questions, count)) # 确保不出现重复题目 seen_questions UserAnswer.objects.filter(useruser).values_list(question, flatTrue) return [q for q in paper if q.id not in seen_questions]微信小程序端防切屏// 监听页面隐藏事件 onHide() { if (this.examStarted !this.examFinished) { uni.showModal({ title: 警告, content: 检测到离开考试页面系统将自动交卷, showCancel: false, success: () this.forceSubmit() }) } }2.3.2 实时自动保存使用WebSocket实现答题进度实时保存# consumers.py class ExamConsumer(WebsocketConsumer): def connect(self): self.exam_id self.scope[url_route][kwargs][exam_id] self.user self.scope[user] self.accept() def receive(self, text_data): data json.loads(text_data) UserAnswer.objects.update_or_create( userself.user, question_iddata[qid], defaults{choice_id: data[answer]} ) self.send(text_datajson.dumps({status: saved}))小程序端连接逻辑const socket uni.connectSocket({ url: wss://yourdomain.com/ws/exam/${examId}/, success: () console.log(Socket connected) }) socket.onMessage(res { const data JSON.parse(res.data) if (data.status saved) { this.lastSaveTime new Date() } })3. 关键技术实现细节3.1 微信小程序与Python后端通信采用RESTful API设计规范特别注意微信环境的特殊处理请求签名验证# utils/wechat_auth.py def check_signature(params): sorted_params sorted(params.items()) raw_str .join([f{k}{v} for k,v in sorted_params if k ! sign]) sign hashlib.md5((raw_str API_KEY).encode()).hexdigest() return sign params.get(sign)敏感数据加密传输# 使用cryptography库进行AES加密 from cryptography.fernet import Fernet key Fernet.generate_key() cipher_suite Fernet(key) def encrypt_data(data: dict) - str: json_str json.dumps(data) return cipher_suite.encrypt(json_str.encode()).decode() def decrypt_data(encrypted: str) - dict: decrypted cipher_suite.decrypt(encrypted.encode()) return json.loads(decrypted.decode())3.2 成绩分析与证书生成自动阅卷算法def auto_grade(user_id, exam_id): answers UserAnswer.objects.filter(user_iduser_id, question__exam_idexam_id) total answers.count() correct answers.filter(choice__is_correctTrue).count() score round(correct / total * 100, 1) # 根据岗位类型设置及格线 user User.objects.get(pkuser_id) passing 80 if user.position 指挥 else 70 return { score: score, passed: score passing, detail: f正确{correct}题/共{total}题 }PDF证书生成# 使用ReportLab生成符合卫健格式的证书 from reportlab.lib.pagesizes import A4 from reportlab.pdfgen import canvas def generate_certificate(user, exam): filename fcert_{user.id}_{exam.id}.pdf c canvas.Canvas(filename, pagesizeA4) # 绘制背景模板 c.drawImage(template.jpg, 0, 0, widthA4[0], heightA4[1]) # 添加动态内容 c.setFont(Helvetica-Bold, 24) c.drawCentredString(300, 400, user.name) c.setFont(Helvetica, 18) c.drawCentredString(300, 350, exam.title) # 添加防伪二维码 qr_data fverify://{exam.id}/{user.id} qr qrcode.make(qr_data) qr.save(temp_qr.png) c.drawImage(temp_qr.png, 400, 100, width100, height100) c.save() return filename4. 性能优化实战经验4.1 高并发报名场景处理当突发公共卫生事件发生时系统可能面临短时间内数千人同时报名的压力。我们通过以下方案确保稳定性Redis队列缓冲# tasks.py shared_task(bindTrue) def process_enrollment(self, data): try: serializer EnrollmentSerializer(datadata) if serializer.is_valid(): serializer.save() return {status: success} except Exception as e: self.retry(exce, countdown60)小程序端排队提示// 提交报名时显示排队状态 submitForm() { this.queueNumber Math.floor(Math.random() * 1000) uni.showLoading({ title: 您当前排队号: ${this.queueNumber}, mask: true }) wx.request({ url: /api/enroll, method: POST, data: this.formData, success: () { uni.hideLoading() uni.showToast({ title: 报名成功 }) } }) }4.2 考试数据分片存储为防止单个数据表过大影响查询性能我们按考试场次分表存储答题记录# models.py class UserAnswerManager(models.Manager): def get_queryset(self): exam_id get_current_exam_id() # 从线程局部变量获取 return super().get_queryset().filter(exam_idexam_id) class UserAnswer(models.Model): objects UserAnswerManager() class Meta: db_table lambda self: fuser_answer_{self.exam_id % 10} # 分10个表配合Django的数据库路由# db_routers.py class UserAnswerRouter: def db_for_read(self, model, **hints): if model._meta.model_name useranswer: return exam_db_{}.format(hints.get(exam_id, 0) % 3) return None5. 运维监控体系搭建5.1 微信小程序错误监控使用Sentry捕获前端异常// app.js import * as Sentry from sentry/mina; Sentry.init({ dsn: your_dsn, integrations: [new Sentry.BrowserTracing()], tracesSampleRate: 0.2 }) // 捕获uniapp全局错误 uni.onError(function(error) { Sentry.captureException(new Error(error)) })5.2 Python后端性能监控使用PrometheusGrafana搭建监控看板# prometheus_metrics.py from prometheus_client import start_http_server, Summary REQUEST_TIME Summary(request_processing_seconds, Time spent processing request) REQUEST_TIME.time() def process_request(request): 业务处理逻辑 pass start_http_server(8000) # 暴露指标端口关键监控指标包括报名接口成功率考试提交平均响应时间WebSocket连接数证书生成队列长度6. 实际部署注意事项微信小程序配置要点在app.json中正确设置导航栏高度{ window: { navigationBarTitleText: 卫生应急培训, navigationStyle: custom } }使用wx.getSystemInfoSync()动态计算布局const systemInfo wx.getSystemInfoSync() const navHeight systemInfo.statusBarHeight 44Python环境隔离建议# 使用pyenv管理多版本 pyenv install 3.8.12 pyenv virtualenv 3.8.12 health-emergency pyenv activate health-emergency # 依赖管理 pip install pipenv pipenv install django3.2 celery5.2 redis4.3Nginx关键配置# WebSocket代理配置 location /ws/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_read_timeout 86400; } # 静态资源缓存 location /static/ { alias /path/to/static/; expires 30d; add_header Cache-Control public; }7. 典型问题排查指南7.1 微信登录失败排查流程检查wx.login()是否成功获取code验证后端接口是否收到code参数检查微信开发者平台的AppID和Secret配置查看微信接口返回的errmsg确认服务器IP已加入微信白名单7.2 考试提交卡顿解决方案前端优化// 使用防抖函数控制提交频率 function debounce(fn, delay) { let timer null return function() { clearTimeout(timer) timer setTimeout(() fn.apply(this, arguments), delay) } }后端优化# 使用django-bulk-update批量操作 from django_bulk_update.helper import bulk_update def bulk_save_answers(answers): bulk_update(answers, update_fields[choice_id])7.3 证书生成乱码处理确保PDF模板使用UTF-8编码在ReportLab中指定中文字体pdfmetrics.registerFont(TTFont(SimHei, SimHei.ttf)) c.setFont(SimHei, 14)对用户输入内容进行安全过滤from django.utils.html import escape safe_name escape(user.name)这个系统在实际运行中已经支撑了超过200场卫生应急培训峰值时单日处理报名申请1.2万条。最深刻的体会是在医疗健康这类严肃领域系统的稳定性比炫酷的功能更重要。比如我们为考试模块设计的断点续考功能就是在真实应急演练中网络中断后紧急开发的。技术方案永远要服务于业务场景的实际需求。