基于Python的医疗质量管理指标智能提取系统【2025代码版】
系统概述
本系统旨在帮助医疗质量管理部从医院信息系统(HIS)中智能提取《2025年国家医疗质量安全改进目标》中的关键指标数据。系统采用Python编程语言，结合现代数据处理库，实现高效、准确的数据提取与分析功能。
代码语言：python代码运行次数：0运行复制import json
import logging
import logging.handlers
import pandas as pd
import pyodbc
from dataclasses import dataclass
from typing import Dict
# 日志配置
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
file_handler = logging.handlers.RotatingFileHandler('quality_management.log', maxBytes=5*1024*1024, backupCount=3)
file_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
logger.addHandler(file_handler)
@dataclass
class QualityGoal:
key: str
name: str
description: str
sql_query: str
def load_config(path: str) -> dict:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def load_quality_goals(goals_dict: dict) -> Dict[str, QualityGoal]:
return {
k: QualityGoal(k, v['name'], v['description'], v['sql_query'])
for k, v in goals_dict.items()
}
class Database:
def __init__(self, config: dict):
self.config = config
self.conn = None
def connect(self):
try:
conn_str = (f"DRIVER={{{self.config['driver']}}};SERVER={self.config['server']};DATABASE={self.config['database']};Trusted_Connection={self.config.get('trusted_connection', 'yes')};")
if 'username' in self.config and 'password' in self.config:
conn_str += f"UID={self.config['username']};PWD={self.config['password']};"
self.conn = pyodbc.connect(conn_str)
logger.info("数据库连接成功")
except Exception as e:
logger.error(f"数据库连接失败: {e}")
raise
def query(self, sql: str) -> pd.DataFrame:
if self.conn is None:
self.connect()
try:
df = pd.read_sql(sql, self.conn)
logger.info(f"SQL执行成功，返回 {len(df)} 条记录")
return df
except Exception as e:
logger.error(f"SQL执行失败: {e}")
raise
def close(self):
if self.conn:
self.conn.close()
logger.info("数据库连接关闭")
def fetch_goal_data(db: Database, goal_key: str, goal_dict: Dict[str, QualityGoal]) -> pd.DataFrame:
goal = goal_dict.get(goal_key)
if not goal:
logger.error(f"目标 {goal_key} 不存在")
return pd.DataFrame()
try:
return db.query(goal.sql_query)
except Exception as e:
logger.error(f"获取目标 {goal_key} 数据失败: {e}")
return pd.DataFrame()
# 示例加载和使用
if __name__ == "__main__":
DB_CONFIG = load_config("db_config.json")
NATIONAL_QUALITY_GOALS_RAW = load_config("national_quality_goals.json")
national_goals = load_quality_goals(NATIONAL_QUALITY_GOALS_RAW)
db = Database(DB_CONFIG)
df = fetch_goal_data(db, 'acute_stroke', national_goals)
print(df.head())
db.close()系统核心功能模块
1. 数据提取模块
代码语言：python代码运行次数：0运行复制import logging
from dataclasses import dataclass
from typing import Dict
from abc import ABC, abstractmethod
from datetime import datetime
import pandas as pd
import pyodbc
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('quality_management.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class QualityDataExtractor(ABC):
"""
医疗质量数据提取器基类
"""
def __init__(self, db_config: Dict, goal_id: str, goal_name: str, sql_query: str):
self.db_config = db_config
self.goal_id = goal_id
self.goal_name = goal_name
self.sql_query = sql_query
@abstractmethod
def extract_data(self) -> pd.DataFrame:
"""
从HIS系统提取医疗质量数据
"""
pass
@abstractmethod
def calculate_metrics(self, data: pd.DataFrame) -> Dict:
"""
计算医疗质量指标
"""
pass
class SQLServerQualityDataExtractor(QualityDataExtractor):
"""
SQL Server数据库医疗质量数据提取器
"""
def __init__(self, db_config: Dict, goal_id: str, goal_name: str, sql_query: str):
super().__init__(db_config, goal_id, goal_name, sql_query)
self.conn = None
def _connect(self) -> bool:
"""建立数据库连接"""
try:
conn_str = (
f"DRIVER={{{self.db_config['driver']}}};"
f"SERVER={self.db_config['server']};"
f"DATABASE={self.db_config['database']};"
f"UID={self.db_config['username']};"
f"PWD={self.db_config['password']};"
f"Trusted_Connection={self.db_config.get('trusted_connection', 'yes')}"
)
self.conn = pyodbc.connect(conn_str)
logger.info(f"成功连接到SQL Server数据库: {self.db_config['server']}")
return True
except Exception as e:
logger.error(f"连接数据库失败: {e}")
return False
def extract_data(self) -> pd.DataFrame:
"""从HIS系统提取医疗质量数据"""
if not self._connect():
raise ConnectionError("无法连接到数据库")
try:
start_time = datetime.now()
logger.info(f"开始提取数据: {self.goal_name}")
df = pd.read_sql_query(self.sql_query, self.conn)
end_time = datetime.now()
elapsed = (end_time - start_time).total_seconds()
logger.info(f"成功提取数据: {len(df)}条记录，耗时{elapsed:.2f}秒")
return df
except Exception as e:
logger.error(f"提取数据失败: {e}")
raise
finally:
if self.conn is not None:
self.conn.close()
def calculate_metrics(self, data: pd.DataFrame) -> Dict:
if data.empty:
return {"error": "没有可用的数据"}
calculators = {
"acute_stroke": self._calc_acute_stroke,
"cancer_tnm": self._calc_cancer_tnm,
"vte_prevention": self._calc_vte_prevention,
"septic_shock": self._calc_septic_shock,
"iv_therapy": self._calc_iv_therapy,
"adverse_events": self._calc_adverse_events,
"surgery_discussion": self._calc_surgery_discussion,
"key_records": self._calc_key_records,
"reoperation_rate": self._calc_reoperation_rate,
"test_recognition": self._calc_test_recognition
}
func = calculators.get(self.goal_id)
if func is None:
return {"error": "未知的质量目标"}
return func(data)
def _calc_acute_stroke(self, data: pd.DataFrame) -> Dict:
total_patients = len(data)
treated_patients = data['treatment_type'].str.contains('再灌注', na=False).sum()
effectiveness = self._calculate_effectiveness(data)
treatment_rate = round((treated_patients / total_patients) * 100, 2) if total_patients > 0 else 0
return {
"total_patients": total_patients,
"treated_patients": treated_patients,
"treatment_rate": treatment_rate,
"treatment_effectiveness": effectiveness
}
def _calc_cancer_tnm(self, data: pd.DataFrame) -> Dict:
total_patients = len(data)
assessed_patients = data['tnm_assessment_date'].notna().sum()
quality = self._calculate_assessment_quality(data)
assessment_rate = round((assessed_patients / total_patients) * 100, 2) if total_patients > 0 else 0
return {
"total_patients": total_patients,
"assessed_patients": assessed_patients,
"assessment_rate": assessment_rate,
"assessment_quality": quality
}
def _calc_vte_prevention(self, data: pd.DataFrame) -> Dict:
total_patients = len(data)
prevented_patients = ((data['vte_risk_assessment_date'].notna()) &
(data['prevention_start_date'].notna())).sum()
quality = self._calculate_prevention_quality(data)
prevention_rate = round((prevented_patients / total_patients) * 100, 2) if total_patients > 0 else 0
return {
"total_patients": total_patients,
"prevented_patients": prevented_patients,
"prevention_rate": prevention_rate,
"prevention_quality": quality
}
def _calc_septic_shock(self, data: pd.DataFrame) -> Dict:
total_patients = len(data)
completed_bundles = ((data['bundle_1_completed'] == True) &
(data['bundle_2_completed'] == True) &
(data['bundle_3_completed'] == True) &
(data['bundle_4_completed'] == True) &
(data['bundle_5_completed'] == True)).sum()
completion_rate = round((completed_bundles / total_patients) * 100, 2) if total_patients > 0 else 0
details = self._calculate_bundle_completion_details(data)
return {
"total_patients": total_patients,
"completed_bundles": completed_bundles,
"completion_rate": completion_rate,
"bundle_completion_details": details
}
def _calc_iv_therapy(self, data: pd.DataFrame) -> Dict:
total_patients = len(data)
compliant_patients = ((data['indication'].notna()) &
(data['fluid_volume_ml'].notna())).sum()
compliance_rate = round((compliant_patients / total_patients) * 100, 2) if total_patients > 0 else 0
details = self._calculate_therapy_details(data)
return {
"total_patients": total_patients,
"compliant_patients": compliant_patients,
"compliance_rate": compliance_rate,
"therapy_details": details
}
def _calc_adverse_events(self, data: pd.DataFrame) -> Dict:
total_events = len(data)
reported_events = data['event_report_date'].notna().sum()
report_rate = round((reported_events / total_events) * 100, 2) if total_events > 0 else 0
distribution = self._calculate_event_distribution(data)
return {
"total_events": total_events,
"reported_events": reported_events,
"report_rate": report_rate,
"event_distribution": distribution
}
def _calc_surgery_discussion(self, data: pd.DataFrame) -> Dict:
total_surgeries = len(data)
discussed_surgeries = data['multidisciplinary_discussion_date'].notna().sum()
discussion_rate = round((discussed_surgeries / total_surgeries) * 100, 2) if total_surgeries > 0 else 0
quality = self._calculate_discussion_quality(data)
return {
"total_surgeries": total_surgeries,
"discussed_surgeries": discussed_surgeries,
"discussion_rate": discussion_rate,
"discussion_quality": quality
}
def _calc_key_records(self, data: pd.DataFrame) -> Dict:
total_records = len(data)
complete_records = (data['record_completion_status'] == '完整').sum()
completion_rate = round((complete_records / total_records) * 100, 2) if total_records > 0 else 0
record_types = self._calculate_record_types(data)
return {
"total_records": total_records,
"complete_records": complete_records,
"completion_rate": completion_rate,
"record_types": record_types
}
def _calc_reoperation_rate(self, data: pd.DataFrame) -> Dict:
total_reoperations = len(data)
total_surgeries = self._get_total_surgeries()
reoperation_rate = round((total_reoperations / total_surgeries) * 100, 2) if total_surgeries > 0 else 0
reasons = self._calculate_reoperation_reasons(data)
return {
"total_reoperations": total_reoperations,
"reoperation_rate": reoperation_rate,
"reoperation_reasons": reasons
}
def _calc_test_recognition(self, data: pd.DataFrame) -> Dict:
total_tests = len(data)
recognized_tests = (data['result_recognized'] == True).sum()
recognition_rate = round((recognized_tests / total_tests) * 100, 2) if total_tests > 0 else 0
details = self._calculate_recognition_details(data)
return {
"total_tests": total_tests,
"recognized_tests": recognized_tests,
"recognition_rate": recognition_rate,
"recognition_details": details
}
# 私有辅助计算方法
def _calculate_effectiveness(self, data: pd.DataFrame) -> Dict:
effective = ((data['outcome'] == '良好') &
(data['treatment_type'].str.contains('再灌注', na=False))).sum()
total_treated = data['treatment_type'].str.contains('再灌注', na=False).sum()
rate = round((effective / total_treated) * 100, 2) if total_treated > 0 else 0
return {
"effective_count": int(effective),
"total_treated": int(total_treated),
"effectiveness_rate": rate
}
def _calculate_assessment_quality(self, data: pd.DataFrame) -> Dict:
high_quality = ((data['assessment_method'] == '多学科讨论') &
(data['assessment_result'].notna())).sum()
total_assessed = data['assessment_result'].notna().sum()
rate = round((high_quality / total_assessed) * 100, 2) if total_assessed > 0 else 0
return {
"high_quality_count": int(high_quality),
"total_assessed": int(total_assessed),
"quality_rate": rate
}
def _calculate_prevention_quality(self, data: pd.DataFrame) -> Dict:
high_quality = ((data['vte_risk_assessment_date'].notna()) &
(data['prevention_method'] == '低分子肝素')).sum()
total_prevented = data['prevention_method'].notna().sum()
rate = round((high_quality / total_prevented) * 100, 2) if total_prevented > 0 else 0
return {
"high_quality_count": int(high_quality),
"total_prevented": int(total_prevented),
"quality_rate": rate
}
def _calculate_bundle_completion_details(self, data: pd.DataFrame) -> Dict:
return {
"bundle_1": int((data['bundle_1_completed'] == True).sum()),
"bundle_2": int((data['bundle_2_completed'] == True).sum()),
"bundle_3": int((data['bundle_3_completed'] == True).sum()),
"bundle_4": int((data['bundle_4_completed'] == True).sum()),
"bundle_5": int((data['bundle_5_completed'] == True).sum()),
}
def _calculate_therapy_details(self, data: pd.DataFrame) -> Dict:
total_volume = data['fluid_volume_ml'].sum()
avg_volume = round(data['fluid_volume_ml'].mean(), 2) if not data['fluid_volume_ml'].empty else 0
medication_types = data['iv_fluid_type'].value_counts().to_dict()
return {
"total_volume": float(total_volume),
"avg_volume_per_patient": avg_volume,
"medication_types": {str(k): int(v) for k, v in medication_types.items()}
}
def _calculate_event_distribution(self, data: pd.DataFrame) -> Dict:
event_dist = data['event_type'].value_counts().to_dict()
severity_levels = data['event_severity'].value_counts().to_dict()
return {
"event_distribution": {str(k): int(v) for k, v in event_dist.items()},
"severity_levels": {str(k): int(v) for k, v in severity_levels.items()}
}
def _calculate_discussion_quality(self, data: pd.DataFrame) -> Dict:
high_quality = ((data['multidisciplinary_discussion_date'].notna()) &
(data['participants'].str.contains('多学科', na=False))).sum()
total_discussed = data['multidisciplinary_discussion_date'].notna().sum()
rate = round((high_quality / total_discussed) * 100, 2) if total_discussed > 0 else 0
return {
"high_quality_count": int(high_quality),
"total_discussed": int(total_discussed),
"quality_rate": rate
}
def _calculate_record_types(self, data: pd.DataFrame) -> Dict:
record_types = data['key_action'].value_counts().to_dict()
return {str(k): int(v) for k, v in record_types.items()}
def _get_total_surgeries(self) -> int:
# 实际应用中应查询数据库获取总手术数量
return 1000
def _calculate_reoperation_reasons(self, data: pd.DataFrame) -> Dict:
reasons = data['reoperation_reason'].value_counts().to_dict()
return {str(k): int(v) for k, v in reasons.items()}
def _calculate_recognition_details(self, data: pd.DataFrame) -> Dict:
recognized_by_others = ((data['result_recognized'] == True) &
(data['test_source'] != data['recognizing_facility'])).sum()
test_type_dist = data['test_type'].value_counts().to_dict()
return {
"recognized_by_other_hospitals": int(recognized_by_others),
"test_type_distribution": {str(k): int(v) for k, v in test_type_dist.items()}
}2. 数据分析模块
代码语言：python代码运行次数：0运行复制from datetime import datetime
from typing import Dict
class QualityDataAnalyzer:
"""
医疗质量数据分析器
"""
@staticmethod
def analyze_metrics(metrics: Dict, goal_id: str) -> Dict:
"""
分析医疗质量指标
"""
analysis = {"goal_id": goal_id}
# 根据不同目标类型进行相应分析
if goal_id == "acute_stroke":
analysis.update({
"analysis": "急性脑梗死再灌注治疗率" +
f"：总患者数{metrics['total_patients']}，" +
f"再灌注治疗患者数{metrics['treated_patients']}，" +
f"治疗率{metrics['treatment_rate']}%。"
})
if metrics['treatment_rate'] < 60:
analysis["warning"] = "再灌注治疗率低于60%，需要重点关注！"
elif goal_id == "cancer_tnm":
analysis.update({
"analysis": "肿瘤治疗前临床TNM分期评估率" +
f"：总患者数{metrics['total_patients']}，" +
f"评估患者数{metrics['assessed_patients']}，" +
f"评估率{metrics['assessment_rate']}%。"
})
if metrics['assessment_rate'] < 80:
analysis["warning"] = "TNM分期评估率低于80%，需要加强评估工作！"
elif goal_id == "vte_prevention":
analysis.update({
"analysis": "静脉血栓栓塞症规范预防率" +
f"：总患者数{metrics['total_patients']}，" +
f"预防患者数{metrics['prevented_patients']}，" +
f"预防率{metrics['prevention_rate']}%。"
})
if metrics['prevention_rate'] < 70:
analysis["warning"] = "静脉血栓栓塞症规范预防率低于70%，需要加强预防措施！"
elif goal_id == "septic_shock":
analysis.update({
"analysis": "感染性休克集束化治疗完成率" +
f"：总患者数{metrics['total_patients']}，" +
f"完成集束化治疗患者数{metrics['completed_bundles']}，" +
f"完成率{metrics['completion_rate']}%。"
})
if metrics['completion_rate'] < 80:
analysis["warning"] = "感染性休克集束化治疗完成率低于80%，需要加强治疗规范！"
elif goal_id == "iv_therapy":
analysis.update({
"analysis": "住院患者静脉输液规范使用率" +
f"：总患者数{metrics['total_patients']}，" +
f"规范使用患者数{metrics['compliant_patients']}，" +
f"规范使用率{metrics['compliance_rate']}%。"
})
if metrics['compliance_rate'] < 75:
analysis["warning"] = "住院患者静脉输液规范使用率低于75%，需要加强规范使用！"
elif goal_id == "adverse_events":
analysis.update({
"analysis": "医疗质量安全不良事件报告率" +
f"：总事件数{metrics['total_events']}，" +
f"报告事件数{metrics['reported_events']}，" +
f"报告率{metrics['report_rate']}%。"
})
if metrics['report_rate'] < 90:
analysis["warning"] = "医疗质量安全不良事件报告率低于90%，需要鼓励主动报告！"
elif goal_id == "surgery_discussion":
analysis.update({
"analysis": "四级手术术前多学科讨论完成率" +
f"：总四级手术数{metrics['total_surgeries']}，" +
f"完成多学科讨论手术数{metrics['discussed_surgeries']}，" +
f"完成率{metrics['discussion_rate']}%。"
})
if metrics['discussion_rate'] < 85:
analysis["warning"] = "四级手术术前多学科讨论完成率低于85%，需要加强术前讨论！"
elif goal_id == "key_records":
analysis.update({
"analysis": "关键诊疗行为相关记录完整率" +
f"：总记录数{metrics['total_records']}，" +
f"完整记录数{metrics['complete_records']}，" +
f"完整率{metrics['completion_rate']}%。"
})
if metrics['completion_rate'] < 80:
analysis["warning"] = "关键诊疗行为相关记录完整率低于80%，需要加强记录管理！"
elif goal_id == "reoperation_rate":
analysis.update({
"analysis": "非计划重返手术室再手术率" +
f"：总再手术数{metrics['total_reoperations']}，" +
f"再手术率{metrics['reoperation_rate']}%。"
})
if metrics['reoperation_rate'] > 5:
analysis["warning"] = "非计划重返手术室再手术率超过5%，需要分析原因！"
elif goal_id == "test_recognition":
analysis.update({
"analysis": "医疗机构检查检验结果互认率" +
f"：总检查检验数{metrics['total_tests']}，" +
f"互认检查检验数{metrics['recognized_tests']}，" +
f"互认率{metrics['recognition_rate']}%。"
})
if metrics['recognition_rate'] < 80:
analysis["warning"] = "医疗机构检查检验结果互认率低于80%，需要加强互认工作！"
else:
analysis["analysis"] = "未知的质量目标"
return analysis
@staticmethod
def generate_report(metrics: Dict, analysis: Dict, goal_id: str, goal_name: str) -> Dict:
"""
生成医疗质量报告
"""
report = {
"goal_id": goal_id,
"goal_name": goal_name,
"metrics": metrics,
"analysis": analysis,
"report_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"recommendations": []
}
# 根据分析结果生成建议
if "warning" in analysis:
report["recommendations"].append(f"警告：{analysis['warning']}")
if goal_id == "acute_stroke":
if metrics.get('treatment_rate', 100) < 60:
report["recommendations"].append("建议加强急性脑梗死再灌注治疗流程管理，提高治疗及时性和有效性。")
if 'treatment_effectiveness' in metrics and \
metrics['treatment_effectiveness'].get('effectiveness_rate', 100) < 70:
report["recommendations"].append("建议优化再灌注治疗方案，提高治疗效果。")
elif goal_id == "cancer_tnm":
if metrics.get('assessment_rate', 100) < 80:
report["recommendations"].append("建议完善肿瘤治疗前临床TNM分期评估流程，确保评估覆盖所有肿瘤患者。")
if 'assessment_quality' in metrics and \
metrics['assessment_quality'].get('quality_rate', 100) < 70:
report["recommendations"].append("建议加强TNM分期评估质量控制，提高评估质量。")
# 可根据需要扩展其它目标的建议
return report3. 数据可视化模块
代码语言：python代码运行次数：0运行复制import os
import logging
from typing import Dict, List
from datetime import datetime
import matplotlib.pyplot as plt
logger = logging.getLogger(__name__)
class QualityDataVisualizer:
"""
医疗质量数据可视化器
"""
@staticmethod
def visualize_metrics(metrics: Dict, goal_id: str, output_dir: str = "visualizations"):
"""
可视化医疗质量指标（柱状图）
"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 统一绘图函数，减少重复代码
def _save_bar_chart(labels, values, title, ylabel, filename):
fig, ax = plt.subplots(figsize=(10, 6))
ax.bar(labels, values, color='skyblue')
ax.set_title(title)
ax.set_ylabel(ylabel)
for i, v in enumerate(values):
ax.text(i, v + max(values)*0.01, str(v), ha='center', va='bottom')
plt.tight_layout()
plt.savefig(os.path.join(output_dir, filename))
plt.close()
# 不同目标对应的可视化配置
visualization_map = {
"acute_stroke": {
"labels": ["总患者数", "再灌注治疗患者数"],
"values": [metrics.get('total_patients', 0), metrics.get('treated_patients', 0)],
"title": f"急性脑梗死再灌注治疗率：{metrics.get('treatment_rate', 0)}%",
"ylabel": "患者数量",
"filename": "acute_stroke.png"
},
"cancer_tnm": {
"labels": ["总患者数", "评估患者数"],
"values": [metrics.get('total_patients', 0), metrics.get('assessed_patients', 0)],
"title": f"肿瘤治疗前临床TNM分期评估率：{metrics.get('assessment_rate', 0)}%",
"ylabel": "患者数量",
"filename": "cancer_tnm.png"
},
"vte_prevention": {
"labels": ["总患者数", "预防患者数"],
"values": [metrics.get('total_patients', 0), metrics.get('prevented_patients', 0)],
"title": f"静脉血栓栓塞症规范预防率：{metrics.get('prevention_rate', 0)}%",
"ylabel": "患者数量",
"filename": "vte_prevention.png"
},
"septic_shock": {
"labels": ["总患者数", "完成集束化治疗患者数"],
"values": [metrics.get('total_patients', 0), metrics.get('completed_bundles', 0)],
"title": f"感染性休克集束化治疗完成率：{metrics.get('completion_rate', 0)}%",
"ylabel": "患者数量",
"filename": "septic_shock.png"
},
"iv_therapy": {
"labels": ["总患者数", "规范使用患者数"],
"values": [metrics.get('total_patients', 0), metrics.get('compliant_patients', 0)],
"title": f"住院患者静脉输液规范使用率：{metrics.get('compliance_rate', 0)}%",
"ylabel": "患者数量",
"filename": "iv_therapy.png"
},
"adverse_events": {
"labels": ["总事件数", "报告事件数"],
"values": [metrics.get('total_events', 0), metrics.get('reported_events', 0)],
"title": f"医疗质量安全不良事件报告率：{metrics.get('report_rate', 0)}%",
"ylabel": "事件数量",
"filename": "adverse_events.png"
},
"surgery_discussion": {
"labels": ["总四级手术数", "完成多学科讨论手术数"],
"values": [metrics.get('total_surgeries', 0), metrics.get('discussed_surgeries', 0)],
"title": f"四级手术术前多学科讨论完成率：{metrics.get('discussion_rate', 0)}%",
"ylabel": "手术数量",
"filename": "surgery_discussion.png"
},
"key_records": {
"labels": ["总记录数", "完整记录数"],
"values": [metrics.get('total_records', 0), metrics.get('complete_records', 0)],
"title": f"关键诊疗行为相关记录完整率：{metrics.get('completion_rate', 0)}%",
"ylabel": "记录数量",
"filename": "key_records.png"
},
"reoperation_rate": {
"labels": ["总再手术数"],
"values": [metrics.get('total_reoperations', 0)],
"title": f"非计划重返手术室再手术率：{metrics.get('reoperation_rate', 0)}%",
"ylabel": "再手术数量",
"filename": "reoperation_rate.png"
},
"test_recognition": {
"labels": ["总检查检验数", "互认检查检验数"],
"values": [metrics.get('total_tests', 0), metrics.get('recognized_tests', 0)],
"title": f"医疗机构检查检验结果互认率：{metrics.get('recognition_rate', 0)}%",
"ylabel": "检查检验数量",
"filename": "test_recognition.png"
},
}
if goal_id in visualization_map:
cfg = visualization_map[goal_id]
_save_bar_chart(
cfg["labels"], cfg["values"], cfg["title"], cfg["ylabel"], cfg["filename"]
)
else:
logger.warning(f"未知的质量目标: {goal_id}")
@staticmethod
def visualize_trend(metrics_list: List[Dict], goal_id: str, output_dir: str = "visualizations"):
"""
可视化医疗质量指标趋势（折线图）
"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
if len(metrics_list) < 2:
logger.warning("数据点不足，无法绘制趋势图")
return
dates = [
m.get('report_date', datetime.now().strftime("%Y-%m-%d")) for m in metrics_list
]
values = []
# 目标对应指标及y轴标签映射
trend_value_map = {
"acute_stroke": ("treatment_rate", "急性脑梗死再灌注治疗率(%)"),
"cancer_tnm": ("assessment_rate", "肿瘤治疗前临床TNM分期评估率(%)"),
"vte_prevention": ("prevention_rate", "静脉血栓栓塞症规范预防率(%)"),
"septic_shock": ("completion_rate", "感染性休克集束化治疗完成率(%)"),
"iv_therapy": ("compliance_rate", "住院患者静脉输液规范使用率(%)"),
"adverse_events": ("report_rate", "医疗质量安全不良事件报告率(%)"),
"surgery_discussion": ("discussion_rate", "四级手术术前多学科讨论完成率(%)"),
"key_records": ("completion_rate", "关键诊疗行为相关记录完整率(%)"),
"reoperation_rate": ("reoperation_rate", "非计划重返手术室再手术率(%)"),
"test_recognition": ("recognition_rate", "医疗机构检查检验结果互认率(%)"),
}
if goal_id not in trend_value_map:
logger.warning(f"未知的质量目标: {goal_id}")
return
key, ylabel = trend_value_map[goal_id]
values = [m.get(key, 0) for m in metrics_list]
fig, ax = plt.subplots(figsize=(12, 6))
ax.plot(dates, values, marker='o', color='teal')
ax.set_title(f"{goal_id} 趋势")
ax.set_xlabel("日期")
ax.set_ylabel(ylabel)
ax.grid(True)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(os.path.join(output_dir, f"trend_{goal_id}.png"))
plt.close()4. 主程序
代码语言：python代码运行次数：0运行复制import logging
def main():
"""
主程序：从HIS系统提取医疗质量数据并进行分析
"""
try:
goal_id = "acute_stroke"
goal_config = NATIONAL_QUALITY_GOALS.get(goal_id)
if not goal_config:
logger.error(f"未找到质量目标配置: {goal_id}")
return False
# 初始化数据提取器
extractor = SQLServerQualityDataExtractor(
db_config=DB_CONFIG,
goal_id=goal_id,
goal_name=goal_config["name"],
sql_query=goal_config["sql_query"]
)
# 从HIS系统提取数据
data = extractor.extract_data()
logger.info(f"成功提取数据，记录数: {len(data)}")
# 计算医疗质量指标
metrics = extractor.calculate_metrics(data)
logger.info(f"计算完成，指标: {metrics}")
# 分析医疗质量指标
analyzer = QualityDataAnalyzer()
analysis = analyzer.analyze_metrics(metrics, goal_id)
logger.info(f"分析结果: {analysis}")
# 生成医疗质量报告
report = analyzer.generate_report(metrics, analysis, goal_id, goal_config["name"])
logger.info(f"报告生成完成")
# 可视化医疗质量指标
visualizer = QualityDataVisualizer()
visualizer.visualize_metrics(metrics, goal_id)
logger.info("指标可视化完成")
# 示例：模拟趋势数据（实际应从数据库或文件读取）
metrics_list = [
{**metrics, "report_date": "2024-01-01"},
{
"total_patients": 120,
"treated_patients": 70,
"treatment_rate": 58.33,
"treatment_effectiveness": {
"effective_count": 45,
"total_treated": 70,
"effectiveness_rate": 64.29
},
"report_date": "2024-02-01"
},
{
"total_patients": 150,
"treated_patients": 90,
"treatment_rate": 60.00,
"treatment_effectiveness": {
"effective_count": 55,
"total_treated": 90,
"effectiveness_rate": 61.11
},
"report_date": "2024-03-01"
}
]
# 可视化趋势
visualizer.visualize_trend(metrics_list, goal_id)
logger.info("趋势可视化完成")
# 输出最终报告
print("\n医疗质量报告")
print("=" * 50)
print(f"目标: {report['goal_name']}")
print(f"指标: {report['metrics']}")
print(f"分析: {report['analysis'].get('analysis', '')}")
if "warning" in report['analysis']:
print(f"警告: {report['analysis']['warning']}")
print("建议:")
for recommendation in report['recommendations']:
print(f"- {recommendation}")
return True
except Exception as e:
logger.error(f"程序运行出错: {e}", exc_info=True)
return False
if __name__ == "__main__":
success = main()
if success:
print("\n程序运行成功！")
else:
print("\n程序运行失败，请检查日志获取详细信息。")系统使用说明
1. 环境配置
代码语言：txt复制安装必要的 Python 库：pip install pandas pyodbc matplotlib seaborn- 配置数据库连接信息：
- 修改
DB_CONFIG字典中的数据库连接参数，包括服务器地址、数据库名称、用户名和密码。
2. 功能扩展
- 添加新的医疗质量目标：
- 在
NATIONAL_QUALITY_GOALS字典中添加新的目标定义。 - 编写相应的 SQL 查询语句。
- 在
QualityDataExtractor类中添加新的计算方法。 - 添加新的专业领域目标：
- 在
SPECIALTY_QUALITY_GOALS字典中添加专业领域目标定义。 - 编写对应的 SQL 查询语句。
- 在
QualityDataExtractor类中实现必要的计算逻辑。
3. 运行程序
- 直接运行主程序：python quality_management.py
- 程序将自动完成如下工作：
- 从 HIS 系统提取指定的医疗质量数据。
- 计算相关指标。
- 分析指标数据。
- 生成医疗质量报告。
- 创建可视化图表。
4. 结果输出
- 日志文件：程序运行过程中记录详细日志，包括数据提取、计算与分析流程，方便问题追踪。
- 报告输出：在控制台输出详细的医疗质量报告，包含指标、分析和建议。
- 可视化图表：生成指标及趋势图，保存在
visualizations目录下，直观展示医疗质量状况及变化趋势。
系统优势
- 自动化处理 — 自动完成从数据提取到报告生成的全流程，极大减少人工操作量。
- 模块化设计 — 采用模块化架构，便于快速集成、扩展新的医疗质量目标和分析功能。
- 可视化分析 — 提供多样化图表，帮助用户更直观理解医疗质量指标。
- 趋势监控 — 支持历史数据趋势分析，便于持续跟踪和评估医疗质量改进成效。
- 报告生成 — 自动生成结构化且富有洞察力的医疗质量报告，支持决策和管理。
扩展建议
- 集成更多数据源 — 支持融合电子病历、医学影像等多源异构数据，丰富质量指标维度。
- 实时监控 — 开发实时数据采集与监控模块，及时发现医疗质量异常并快速响应。
- 预测分析 — 结合机器学习技术，预测医疗质量趋势及潜在风险，实现前瞻性管理。
- 移动端支持 — 推出移动端应用，方便医务人员随时随地获取医疗质量相关信息。
- 预警系统 — 构建自动预警机制，对异常指标及时发出警报，促进医疗服务质量提升。
本系统为医疗质量管理提供强大技术支撑，助力医疗机构高效、精准地监测与提升质量水平。医疗机构可根据自身业务需求灵活扩展和定制，打造符合自身特点的科学化、精细化医疗质量管理解决方案。
附录：《2025年国家医疗质量安全改进目标》的核心内容
3月21日，国家卫生健康委官网发布《2025年国家医疗质量安全改进目标》和各专业2025年质控工作改进目标。其中，本年度将「提高医疗机构检查检验结果互认率」纳入国家医疗质量安全改进目标。
国家卫生健康委自2021年起，每年发布年度国家医疗质量安全改进目标和各专业质控工作改进目标。《2025年国家医疗质量安全改进目标》是根据《全面提升医疗质量行动计划（2023-2025年）》《关于进一步推进医疗机构检查检验结果互认的指导意见》等文件要求，结合2024年目标改进情况和变化趋势以及年度医疗质量安全监测数据情况而形成。
10项国家医疗质量安全改进目标
文件明确了以下10项核心改进目标[1][3]：
- 提高急性脑梗死再灌注治疗率undefined针对急性脑梗死患者，提升及时有效的再灌注治疗比例，以改善患者预后。
- 提高肿瘤治疗前临床TNM分期评估率undefined确保肿瘤患者在治疗前接受规范的TNM分期评估，为精准治疗提供依据。
- 提高静脉血栓栓塞症规范预防率undefined连续第5年将此目标纳入改进重点[2]，强调对静脉血栓栓塞症的规范化预防。
- 提高感染性休克集束化治疗完成率undefined推动感染性休克患者接受全面的集束化治疗，降低死亡率。
- 提高住院患者静脉输液规范使用率undefined针对静脉输液这一常见治疗方式，完善质量改进长效机制[0]，重点关注住院患者静脉输液使用率、每床日静脉输液使用频次、液体总量和药品品种数量等指标，采取综合措施予以干预，以维护医疗安全和患者权益[0]。
- 提高医疗质量安全不良事件报告率undefined鼓励医疗机构和医务人员主动报告医疗质量安全不良事件，促进经验共享和持续改进。
- 提高四级手术术前多学科讨论完成率undefined确保复杂四级手术前进行充分的多学科讨论，降低手术风险。
- 提高关键诊疗行为相关记录完整率undefined提升关键诊疗行为的记录质量，为医疗质量和安全提供数据支持。
- 降低非计划重返手术室再手术率undefined减少非计划重返手术室再手术的发生，提升手术质量和安全。
- 提高医疗机构检查检验结果互认率undefined这一目标旨在实现不同医疗机构间的检查检验结果互认，有助于提高医疗资源的利用率，降低医疗费用、提高诊疗效率，进一步改善人民群众就医体验[1][2][3]。医疗机构应建立及完善本机构检查检验结果互认的监测评估机制，按月进行本机构数据分析、反馈。对互认率进行监测评估，分析不予互认的具体原因[0]。
专业领域质控工作改进目标
除上述10项国家目标外，《2025年国家医疗质量安全改进目标》还涉及40个专业领域的质控工作改进目标[1][3]，包括呼吸内科、产科、儿科及小儿外科、罕见病等[2]。这些专业目标是对国家目标的细化和补充，旨在针对不同专科的特点和需求，制定更具针对性的改进措施。
政策背景与实施要点
《2025年国家医疗质量安全改进目标》的制定延续了国家卫生健康委从2021年起连续4年组织制定国家医疗质量安全改进目标的做法[3]，体现了医疗质量管理的连续性和发展性。2025年目标文件进一步优化了改进工作策略，创新工作机制和方式方法，深入推进目标管理，实现医疗质量安全持续改进[3]。
文件强调，实现不同医疗机构间的检查检验结果互认，一方面有助于提高医疗资源的利用率，降低医疗费用、提高诊疗效率；另一方面贯彻落实2025年卫生健康系统为民办实事具体要求，进一步改善人民群众就医体验[4]。这反映了医疗质量管理从单纯关注医疗行为规范向医疗资源优化和患者体验提升的拓展。
在实施层面，目标文件要求医疗机构建立及完善本机构检查检验结果互认的监测评估机制，按月进行本机构数据分析、反馈。对互认率进行监测评估，分析不予互认的具体原因[0]。这种数据驱动的改进模式，体现了医疗质量管理的科学化和精细化趋势。
《2025年国家医疗质量安全改进目标》的发布标志着我国医疗质量管理进入了以目标为导向、多专业协同、数据驱动的系统化改进新阶段，为医疗机构提升医疗质量提供了明确的政策指引和行动框架。医疗机构应结合自身实际情况，将国家目标和专业目标转化为具体的改进措施，持续推动医疗质量安全管理水平提升。
发布评论