115-基于Flask的医疗保健数据预测分析系统

基于 Flask 的医疗保健数据预测分析系统实战分享

面向落地的实践复盘，基于真实代码。本文将完整介绍项目背景、目录结构、技术栈、真实功能清单、关键代码走读、可视化展示占位、快速开始与 API 速览，并附上后续规划与经验总结，便于二次开发与教学展示。

目录

• 项目背景与目标
• 技术栈（后端/数据库/数据处理/可视化/运维）
• 项目演示
• 项目目录结构
• 基于真实代码的功能清单
• 关键代码走读（含源码片段）
• 可视化展示（占位）
• 快速开始（从零到跑通）
• API 速览（含示例）
• 实践经验与踩坑
• 后续计划
• 联系方式

项目背景与目标

这是一个基于 Flask 的医疗保健数据管理与分析系统，聚焦“患者数据管理、统计分析、可视化与报告导出”。系统提供患者档案的增删改查、批量导入导出、统计面板、可视化图表与报告导出等能力，适合课程设计、教学演示与中小型数据分析后台的快速搭建。

技术栈

• 后端与框架：Flask、Flask-SQLAlchemy、Flask-Login、Flask-WTF、Flask-Migrate、WTForms、Werkzeug
• 数据库与ORM：MySQL（PyMySQL、mysql-connector-python 驱动）、SQLAlchemy；测试环境内置 SQLite
• 数据处理：pandas、numpy、openpyxl、xlsxwriter、chardet
• 可视化（前端渲染）：Chart.js（实际模板引入）、ECharts（首页介绍中出现）
• 机器学习基础依赖：scikit-learn、joblib（代码中已准备，前端未直接提供“预测”入口）
• 其他：python-dotenv、python-dateutil、uuid

依赖来自项目根目录 requirements.txt，可一键安装。

项目演示

😀 项目源码获取，码界筑梦坊各平台同名，博客底部含联系方式卡片，欢迎咨询！

项目目录结构

精简展示关键目录，保持与仓库一致：

app/
  __init__.py
  models/
    patient.py
    user.py
  routes/
    admin.py
    analytics.py
    api.py
    auth.py
    import_route.py
    main.py
    patient.py
  services/
    data_service.py
    import_service.py
  templates/
    admin/
      backup.html
      import_data.html
      settings.html
      users.html
    analytics/
      charts.html
      dashboard.html
      report_export.html
      reports.html
    auth/
      change_password.html
      login.html
      profile.html
      register.html
    base.html
    dashboard.html
    index.html
    patient/
      add.html
      detail.html
      edit.html
      list.html
config.py
data/
  healthcare_dataset.csv
design_115_health.sql
migrations/
scripts/
  create_database.py
  import_data.py
  init_db.py
requirements.txt
run.py
start.py
static/
uploads/

基于真实代码的功能清单

以下功能均来自于实际前后端代码，非设想：

• 患者管理
  • 新增、编辑、删除患者
  • 患者列表分页、筛选、搜索
  • 患者详细信息查看
  • 批量导入（CSV/Excel）、批量导出（CSV）
• 数据分析与统计
  • 患者总数、性别分布、年龄组、疾病类型、入院类型、保险提供商等统计
  • 医生工作量、医院统计、月度趋势、风险等级分布
• 数据可视化
  • 图表数据 API（多类型统计），前端页面通过 Chart.js 渲染
• 统计报告
  • 报告页面与 HTML 导出（/analytics/export-report）
• 用户与权限
  • 注册、登录、登出；个人信息与密码修改；基于角色的权限控制
• 管理与工具
  • 数据库创建、管理员初始化脚本；数据导入脚本；基础备份与设置页面

说明：后端存在基于规则的“费用预测”和“风险评估”API（/api/predict-cost、/api/predict-risk），但前端未提供独立的预测功能页或入口，默认不展示为已上线前端功能。

关键代码走读

1) 应用工厂与蓝图注册（app/__init__.py）

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    db.init_app(app)
    login_manager.init_app(app)

    # 蓝图注册
    from app.routes.main import main as main_blueprint
    app.register_blueprint(main_blueprint)
    from app.routes.auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint, url_prefix='/auth')
    from app.routes.patient import patient as patient_blueprint
    app.register_blueprint(patient_blueprint, url_prefix='/patient')
    from app.routes.analytics import analytics as analytics_blueprint
    app.register_blueprint(analytics_blueprint, url_prefix='/analytics')
    from app.routes.api import api as api_blueprint
    app.register_blueprint(api_blueprint, url_prefix='/api')
    from app.routes.import_route import import_bp as import_blueprint
    app.register_blueprint(import_blueprint, url_prefix='/import')
    from app.routes.admin import admin as admin_blueprint
    app.register_blueprint(admin_blueprint, url_prefix='/admin')
    return app

2) 患者列表/筛选与导出（app/routes/patient.py）

@patient.route('/')
@login_required
def list():
    page = request.args.get('page', 1, type=int)
    filters = {
        'name': request.args.get('name', '').strip(),
        'age_min': request.args.get('age_min', type=int),
        'age_max': request.args.get('age_max', type=int),
        'gender': request.args.get('gender', '').strip(),
        'medical_condition': request.args.get('medical_condition', '').strip(),
        'admission_type': request.args.get('admission_type', '').strip(),
        'insurance_provider': request.args.get('insurance_provider', '').strip(),
        'test_results': request.args.get('test_results', '').strip(),
        'sort_by': request.args.get('sort_by', 'created_at'),
        'sort_order': request.args.get('sort_order', 'desc')
    }
    filters = {k: v for k, v in filters.items() if v}
    pagination = DataService().search_patients(filters, page, per_page=15)
    return render_template('patient/list.html', patients=pagination.items,
                          pagination=pagination, filters=filters, ...)

@patient.route('/export')
@login_required
def export():
    if not current_user.can_export_data():
        flash('您没有权限执行此操作', 'error')
        return redirect(url_for('patient.list'))
    patients = DataService().get_all_patients_for_export(filters)
    # 组装 CSV 并返回下载响应

3) 统计与可视化数据服务（app/services/data_service.py）

class DataService:
    def get_medical_condition_stats(self):
        stats = (
            db.session.query(
                Patient.medical_condition,
                func.count(Patient.id).label('count'),
                func.avg(Patient.age).label('avg_age'),
                func.avg(Patient.billing_amount).label('avg_cost')
            ).group_by(Patient.medical_condition).all()
        )
        mapping = Patient.get_medical_condition_mapping()
        return [{
            'condition': mapping.get(s.medical_condition, s.medical_condition),
            'count': s.count,
            'avg_age': round(float(s.avg_age), 1),
            'avg_cost': round(float(s.avg_cost), 2)
        } for s in stats]

    def get_monthly_admission_trend(self, months=24):
        stats = (
            db.session.query(
                func.date_format(Patient.date_of_admission, '%Y-%m').label('month'),
                func.count(Patient.id).label('count'),
                func.avg(Patient.billing_amount).label('avg_cost')
            ).filter(Patient.date_of_admission.isnot(None))
             .group_by(func.date_format(Patient.date_of_admission, '%Y-%m'))
             .order_by(func.date_format(Patient.date_of_admission, '%Y-%m')).all()
        )
        return [{'month': s.month, 'count': s.count, 'avg_cost': round(float(s.avg_cost), 2)} for s in stats]

4) CSV/Excel 批量导入（app/services/import_service.py）

class ImportService:
    def import_csv_to_database(self, csv_file_path: str, user_id: int, batch_size: int = 1000) -> Dict:
        df = pd.read_csv(csv_file_path, encoding='utf-8')
        df_processed = self._preprocess_dataframe(df)
        success_count, errors = self._batch_import_patients(df_processed, user_id, batch_size)
        return { 'success': success_count > 0, 'success_count': success_count, 'errors': errors }

    def _preprocess_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
        # 列名中英映射、枚举标准化、日期与数值字段校验
        # 缺失值处理、非法数据拦截与日志记录
        ...

5) 图表数据 API（app/routes/analytics.py 与 app/routes/api.py）

@analytics.route('/api/chart_data/<chart_type>')
def chart_data(chart_type):
    if chart_type == 'gender_distribution':
        stats = DataService().get_gender_stats()
        return jsonify({'labels': [s['gender'] for s in stats], 'data': [s['count'] for s in stats]})
    elif chart_type == 'monthly_trend':
        stats = DataService().get_monthly_admission_trend()
        return jsonify({'labels': [s['month'] for s in stats], 'admission_counts': [s['count'] for s in stats]})
    ...

6) 配置（config.py）

class Config:
    MYSQL_HOST = 'localhost'
    MYSQL_PORT = 3306
    MYSQL_USER = 'root'
    MYSQL_PASSWORD = '123456'
    MYSQL_DB = 'design_115_health'
    SQLALCHEMY_DATABASE_URI = (
        f'mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}?charset=utf8mb4'
    )
    PATIENTS_PER_PAGE = 15
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024

7) 模型设计要点

• 患者模型（app/models/patient.py）核心字段、风险评估与显示属性：

class Patient(db.Model):
    __tablename__ = 'patients'
    id = db.Column(db.Integer, primary_key=True)
    # 基本信息
    name = db.Column(db.String(100), nullable=False, index=True)
    age = db.Column(db.Integer, nullable=False, index=True)
    gender = db.Column(db.String(10), nullable=False, index=True)
    blood_type = db.Column(db.String(5), nullable=False, index=True)
    # 医疗信息
    medical_condition = db.Column(db.String(100), nullable=False, index=True)
    date_of_admission = db.Column(db.Date, nullable=False, index=True)
    doctor = db.Column(db.String(100), nullable=False, index=True)
    hospital = db.Column(db.String(200), nullable=False, index=True)
    insurance_provider = db.Column(db.String(100), nullable=False, index=True)
    billing_amount = db.Column(db.Numeric(10, 2), nullable=False, index=True)
    room_number = db.Column(db.Integer, nullable=False)
    admission_type = db.Column(db.String(20), nullable=False, index=True)
    discharge_date = db.Column(db.Date, nullable=False, index=True)
    medication = db.Column(db.String(100), nullable=False, index=True)
    test_results = db.Column(db.String(20), nullable=False, index=True)
    # 系统字段
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    @property
    def stay_duration(self):
        if self.date_of_admission and self.discharge_date:
            return (self.discharge_date - self.date_of_admission).days
        return 0

    @property
    def risk_level(self):
        risk_score = 0
        if self.age > 65: risk_score += 2
        elif self.age > 50: risk_score += 1
        if self.medical_condition in ['癌症', '高血压', 'Cancer', 'Hypertension']:
            risk_score += 2
        if self.admission_type in ['急诊', 'Emergency']:
            risk_score += 2
        elif self.admission_type in ['急症', 'Urgent']:
            risk_score += 1
        if self.test_results in ['异常', 'Abnormal']:
            risk_score += 2
        elif self.test_results in ['不确定', 'Inconclusive']:
            risk_score += 1
        return '高风险' if risk_score >= 6 else ('中风险' if risk_score >= 3 else '低风险')

    @property
    def medical_condition_display(self):
        mapping = self.get_medical_condition_mapping()
        return mapping.get(self.medical_condition, self.medical_condition)

• 用户模型（app/models/user.py）权限设计（简化摘录）：

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    is_active = db.Column(db.Boolean, default=True)
    is_admin = db.Column(db.Boolean, default=False)

    def has_permission(self, permission):
        if not self.is_active:
            return False
        if self.is_admin:
            return True
        user_permissions = ['view_patients','view_analytics','view_charts','view_reports','view_dashboard']
        return permission in user_permissions

    # 管理员快捷判断
    def can_manage_patients(self):
        return self.is_admin
    def can_import_data(self):
        return self.is_admin
    def can_export_data(self):
        return self.is_admin

可视化展示（占位）

以下为文章与项目展示的图片占位（后期可替换为实际截图）。

• 仪表盘总览（Dashboard）
  • 占位：docs/images/dashboard-overview.png
• 多维度图表（疾病分布、入院类型、年龄段、月度趋势等）
  • 占位：docs/images/charts-examples.png
• 患者列表与筛选
  • 占位：docs/images/patient-list.png
• 导入导出与报告
  • 占位：docs/images/import-export-report.png

模板中通过 CDN 引入 Chart.js（如 analytics/charts.html、analytics/dashboard.html），前端读取后端 API 的 JSON 数据进行渲染。

快速开始

1. 安装依赖

pip install -r requirements.txt

2. 初始化数据库（创建库、表与管理员）

python scripts/create_database.py

3. 导入示例数据（可选）

python scripts/import_data.py

4. 启动应用

python run.py
# 访问：http://localhost:5000

API 速览（节选）

• 列表与检索
  • GET /api/patients?name=张三&medical_condition=Diabetes&page=1&per_page=20
• 单条数据
  • GET /api/patients/<id>，PUT /api/patients/<id>，DELETE /api/patients/<id>
• 图表数据
  • GET /analytics/api/chart_data/gender_distribution
  • GET /analytics/api/chart_data/monthly_trend
• 报告导出
  • GET /analytics/export-report（返回 HTML 文件下载）
• 规则版预测（后端存在 API，前端默认无入口）
  • POST /api/predict-cost
  • POST /api/predict-risk

示例：预测费用请求体（简化规则）

{
  "age": 66,
  "medical_condition": "Diabetes",
  "admission_type": "Emergency",
  "stay_days": 7
}

实践经验与踩坑

• 数据清洗先行：导入前的列名映射、日期与数值校验能拦住绝大多数问题。
• 统一字典映射：后端存原值，前端/导出显示映射后的中文，代码通过 mapping 保持一致。
• 图表走 JSON：后端只产数据不渲染图，前端用 Chart.js/ECharts 渲染，职责更清晰。
• 权限粒度适中：导出、删除等操作均做权限约束，避免误操作。
• 生产配置独立：ProductionConfig 中的 Cookie 安全和日志建议与开发环境分离。

后续计划

• 前端新增“预测”功能页：与现有 /api/predict-* API 打通，支持单条与批量预测。
• 模型化：接入持久化的 ML 模型（scikit-learn/joblib），替换当前规则估计接口。
• 可视化增强：统一封装 Chart 组件，支持主题切换与导出图片/数据。
• 多租户与审计：更细粒度的权限、操作日志与数据隔离。

联系方式

码界筑梦坊各大平台同名