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__.pymodels/patient.pyuser.pyroutes/admin.pyanalytics.pyapi.pyauth.pyimport_route.pymain.pypatient.pyservices/data_service.pyimport_service.pytemplates/admin/backup.htmlimport_data.htmlsettings.htmlusers.htmlanalytics/charts.htmldashboard.htmlreport_export.htmlreports.htmlauth/change_password.htmllogin.htmlprofile.htmlregister.htmlbase.htmldashboard.htmlindex.htmlpatient/add.htmldetail.htmledit.htmllist.html
config.py
data/healthcare_dataset.csv
design_115_health.sql
migrations/
scripts/create_database.pyimport_data.pyinit_db.py
requirements.txt
run.py
start.py
static/
uploads/
基于真实代码的功能清单
以下功能均来自于实际前后端代码,非设想:
- 患者管理
- 新增、编辑、删除患者
- 患者列表分页、筛选、搜索
- 患者详细信息查看
- 批量导入(CSV/Excel)、批量导出(CSV)
- 数据分析与统计
- 患者总数、性别分布、年龄组、疾病类型、入院类型、保险提供商等统计
- 医生工作量、医院统计、月度趋势、风险等级分布
- 数据可视化
- 图表数据 API(多类型统计),前端页面通过 Chart.js 渲染
- 统计报告
- 报告页面与 HTML 导出(
/analytics/export-report
)
- 报告页面与 HTML 导出(
- 用户与权限
- 注册、登录、登出;个人信息与密码修改;基于角色的权限控制
- 管理与工具
- 数据库创建、管理员初始化脚本;数据导入脚本;基础备份与设置页面
说明:后端存在基于规则的“费用预测”和“风险评估”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_blueprintapp.register_blueprint(main_blueprint)from app.routes.auth import auth as auth_blueprintapp.register_blueprint(auth_blueprint, url_prefix='/auth')from app.routes.patient import patient as patient_blueprintapp.register_blueprint(patient_blueprint, url_prefix='/patient')from app.routes.analytics import analytics as analytics_blueprintapp.register_blueprint(analytics_blueprint, url_prefix='/analytics')from app.routes.api import api as api_blueprintapp.register_blueprint(api_blueprint, url_prefix='/api')from app.routes.import_route import import_bp as import_blueprintapp.register_blueprint(import_blueprint, url_prefix='/import')from app.routes.admin import admin as admin_blueprintapp.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 = 3306MYSQL_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 = 15MAX_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)@propertydef stay_duration(self):if self.date_of_admission and self.discharge_date:return (self.discharge_date - self.date_of_admission).daysreturn 0@propertydef risk_level(self):risk_score = 0if self.age > 65: risk_score += 2elif self.age > 50: risk_score += 1if self.medical_condition in ['癌症', '高血压', 'Cancer', 'Hypertension']:risk_score += 2if self.admission_type in ['急诊', 'Emergency']:risk_score += 2elif self.admission_type in ['急症', 'Urgent']:risk_score += 1if self.test_results in ['异常', 'Abnormal']:risk_score += 2elif self.test_results in ['不确定', 'Inconclusive']:risk_score += 1return '高风险' if risk_score >= 6 else ('中风险' if risk_score >= 3 else '低风险')@propertydef 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 Falseif self.is_admin:return Trueuser_permissions = ['view_patients','view_analytics','view_charts','view_reports','view_dashboard']return permission in user_permissions# 管理员快捷判断def can_manage_patients(self):return self.is_admindef can_import_data(self):return self.is_admindef 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 数据进行渲染。
快速开始
- 安装依赖
pip install -r requirements.txt
- 初始化数据库(创建库、表与管理员)
python scripts/create_database.py
- 导入示例数据(可选)
python scripts/import_data.py
- 启动应用
python run.py
# 访问:http://localhost:5000
API 速览(节选)
- 列表与检索
- GET
/api/patients?name=张三&medical_condition=Diabetes&page=1&per_page=20
- GET
- 单条数据
- GET
/api/patients/<id>
,PUT/api/patients/<id>
,DELETE/api/patients/<id>
- GET
- 图表数据
- GET
/analytics/api/chart_data/gender_distribution
- GET
/analytics/api/chart_data/monthly_trend
- GET
- 报告导出
- GET
/analytics/export-report
(返回 HTML 文件下载)
- GET
- 规则版预测(后端存在 API,前端默认无入口)
- POST
/api/predict-cost
- POST
/api/predict-risk
- POST
示例:预测费用请求体(简化规则)
{"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 组件,支持主题切换与导出图片/数据。
- 多租户与审计:更细粒度的权限、操作日志与数据隔离。
联系方式
码界筑梦坊各大平台同名