Python Django/Flask医疗预约系统开发实践
1. 项目概述
医疗预约与诊断系统是医疗机构数字化转型的核心基础设施。这个基于Python Django/Flask框架开发的系统,本质上解决了传统医疗场景中的三大痛点:患者排队时间长、医生资源分配不均、病历信息管理混乱。我在三甲医院信息化建设项目中,曾亲眼见证过一套优秀的预约系统如何将门诊效率提升40%以上。
选择Django+Flask的组合方案,是经过实际项目验证的成熟架构。Django自带完善的Admin后台和ORM,能快速搭建基础数据模型;Flask的轻量级特性则非常适合开发灵活的API接口。这种组合既保证了开发效率,又能满足医疗系统对稳定性和扩展性的严苛要求。
2. 系统架构设计
2.1 技术栈选型分析
核心框架采用Django 4.1 + Flask 2.2的组合方案,主要基于以下考量:
- Django的认证系统(django.contrib.auth)可直接复用,满足医疗系统严格的权限控制需求
- Flask的Blueprint功能可以模块化开发预约、诊断、报表等业务组件
- 使用Django ORM管理基础数据模型(患者、医生、科室等)
- 通过Flask-RESTful构建REST API供移动端调用
数据库选用PostgreSQL 14,因其具有:
- 完善的JSON字段支持(存储动态病历模板)
- 良好的地理空间扩展(用于分院区管理)
- 优于MySQL的事务处理能力(挂号支付场景必需)
2.2 核心模块划分
系统采用微服务架构设计,主要包含以下服务:
| 服务模块 | 技术实现 | 关键功能 |
|---|---|---|
| 用户中心 | Django | 患者/医生注册、权限管理 |
| 预约服务 | Flask + Celery | 号源管理、智能分诊 |
| 诊断系统 | Django + Vue.js | 电子病历、处方开具 |
| 支付网关 | Flask + Alipay SDK | 挂号费/诊费在线支付 |
| 数据分析 | Django ORM + Pandas | 就诊量统计、医生绩效分析 |
3. 关键功能实现
3.1 智能预约调度算法
核心算法采用改进的时间片轮转策略,在services/scheduling.py中实现:
def generate_slots(doctor_id, date): """ 生成可预约时间段 算法逻辑: 1. 读取医生基础坐班时间 2. 排除已预约时段 3. 动态调整时段长度(初诊15min,复诊10min) 4. 保留20%号源用于现场挂号 """ base_slots = DoctorSchedule.objects.get(doctor=doctor_id) appointments = Appointment.objects.filter( doctor=doctor_id, date=date ).values_list('time_slot', flat=True) available_slots = [] for slot in base_slots: if slot not in appointments: # 动态调整时段长度 duration = 15 if is_first_visit(patient) else 10 available_slots.append({ 'start': slot, 'end': slot + timedelta(minutes=duration), 'type': 'online' if len(available_slots) < 0.8*base_slots else 'onsite' }) return available_slots重要提示:医疗系统必须考虑线下场景,永远不要将号源100%开放线上预约,需保留现场挂号通道。
3.2 电子病历模板系统
采用Django的JSONField实现动态表单,关键模型设计:
class MedicalRecord(models.Model): PATIENT_TYPE_CHOICES = [ ('OUT', '门诊'), ('IN', '住院'), ] patient = models.ForeignKey(Patient, on_delete=models.CASCADE) doctor = models.ForeignKey(Doctor, on_delete=models.PROTECT) record_type = models.CharField(max_length=3, choices=PATIENT_TYPE_CHOICES) template = models.ForeignKey(RecordTemplate, on_delete=models.SET_NULL) content = models.JSONField() # 存储结构化病历数据 created_at = models.DateTimeField(auto_now_add=True) def render_html(self): """将JSON数据渲染为HTML病历""" return render_template( f'records/{self.template.name}.html', data=self.content )4. 安全与合规实现
4.1 医疗数据加密方案
采用双层加密策略确保数据安全:
数据库层面:使用PostgreSQL的pgcrypto扩展
CREATE EXTENSION pgcrypto; UPDATE patients SET id_card = pgp_sym_encrypt(id_card, 'encryption_key');应用层面:Django信号机制自动加密
@receiver(pre_save, sender=Patient) def encrypt_sensitive_data(sender, instance, **kwargs): if instance.id_card: instance.id_card = encrypt(instance.id_card)
4.2 审计日志实现
符合医疗信息系统等保三级要求:
class AuditLog(models.Model): ACTION_CHOICES = [ ('VIEW', '查看'), ('EDIT', '修改'), ('DELETE', '删除'), ] user = models.ForeignKey(User, on_delete=models.PROTECT) action = models.CharField(max_length=6, choices=ACTION_CHOICES) model = models.CharField(max_length=50) object_id = models.CharField(max_length=36) ip_address = models.GenericIPAddressField() timestamp = models.DateTimeField(auto_now_add=True) @classmethod def log(cls, request, action, obj): cls.objects.create( user=request.user, action=action, model=obj.__class__.__name__, object_id=str(obj.pk), ip_address=get_client_ip(request) )5. 性能优化实践
5.1 预约高并发处理
使用Redis + Celery实现:
# tasks.py @app.task(bind=True, rate_limit='100/m') def make_appointment(self, patient_id, slot_id): try: with transaction.atomic(): slot = TimeSlot.objects.select_for_update().get(pk=slot_id) if slot.status == 'AVAILABLE': Appointment.objects.create( patient_id=patient_id, time_slot=slot, status='PENDING_PAYMENT' ) slot.status = 'RESERVED' slot.save() return True except Exception as e: self.retry(exc=e, countdown=60)5.2 数据库查询优化
针对高频查询的优化措施:
使用
select_related预加载外键关系Appointment.objects.select_related('patient', 'doctor').filter(date=today)对科室表添加复合索引
class Department(models.Model): class Meta: indexes = [ models.Index(fields=['hospital', 'name']), ]使用Django的
prefetch_related优化多对多查询Doctor.objects.prefetch_related('specialties').filter(department=dept)
6. 部署架构建议
6.1 生产环境配置
推荐使用Docker Swarm或Kubernetes部署:
# docker-compose.prod.yml services: web: image: registry.example.com/medical-app environment: - DATABASE_URL=postgres://user:pass@db:5432/medical - REDIS_URL=redis://redis:6379/0 deploy: replicas: 3 resources: limits: cpus: '2' memory: 2G celery: image: registry.example.com/medical-app command: celery -A core worker -l INFO deploy: replicas: 26.2 监控方案
使用Prometheus + Grafana监控关键指标:
Django应用指标:django-prometheus
INSTALLED_APPS += ['django_prometheus'] MIDDLEWARE.insert(0, 'django_prometheus.middleware.PrometheusBeforeMiddleware')自定义业务指标:
from prometheus_client import Counter APPOINTMENT_CREATED = Counter( 'appointment_created_total', 'Total created appointments', ['department'] ) @api_view(['POST']) def create_appointment(request): APPOINTMENT_CREATED.labels(department=dept.name).inc()
7. 常见问题解决方案
7.1 号源冲突处理
采用乐观锁机制防止超卖:
def reserve_slot(slot_id): slot = TimeSlot.objects.get(pk=slot_id) if slot.status == 'AVAILABLE': rows = TimeSlot.objects.filter( pk=slot.pk, status='AVAILABLE' ).update(status='RESERVED') if rows == 0: raise ConcurrentModificationError()7.2 医保对接方案
典型医保接口封装示例:
class MedicalInsurance: def __init__(self, config): self.wsdl = config['wsdl'] def verify_patient(self, id_card, medical_card): """实名认证""" client = zeep.Client(wsdl=self.wsdl) return client.service.verify( id_card=id_card, medical_card=medical_card ) def submit_bill(self, appointment_id): """医保结算""" appointment = Appointment.objects.get(pk=appointment_id) items = [{ 'item_code': 'REG', 'fee': appointment.fee }] return client.service.submit( patient_id=appointment.patient.medical_card, items=items )这套系统在实际部署时需要特别注意医疗行业的特殊要求:所有数据库操作必须开启事务,关键业务日志至少保留5年,患者敏感信息需要在前端展示时自动脱敏(如身份证号显示为110**********1234)。我在某三甲医院实施时,曾因为忽略医保系统的异步回调机制导致账目不平,后来通过添加补偿事务机制解决了这个问题。
