当前位置：首页 > news >正文

基于Python的医疗质量管理指标智能提取系统【2025代码版】

news 来源：原创 2025/5/21 15:29:21

在这里插入图片描述

系统概述

本系统旨在帮助医疗质量管理部从医院信息系统(HIS)中智能提取《2025年国家医疗质量安全改进目标》中的关键指标数据。系统采用Python编程语言，结合现代数据处理库，实现高效、准确的数据提取与分析功能。

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: strname: strdescription: strsql_query: strdef 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 = configself.conn = Nonedef 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}")raisedef 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 dfexcept Exception as e:logger.error(f"SQL执行失败: {e}")raisedef 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. 数据提取模块

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_configself.goal_id = goal_idself.goal_name = goal_nameself.sql_query = sql_query@abstractmethoddef extract_data(self) -> pd.DataFrame:"""从HIS系统提取医疗质量数据"""pass@abstractmethoddef calculate_metrics(self, data: pd.DataFrame) -> Dict:"""计算医疗质量指标"""passclass 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 = Nonedef _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 Trueexcept Exception as e:logger.error(f"连接数据库失败: {e}")return Falsedef 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 dfexcept Exception as e:logger.error(f"提取数据失败: {e}")raisefinally: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