医数智联·第 21 期:Python 综合实战 —— 首页质控全流程自动化,一个脚本搞定所有

[!ABSTRACT] 核心摘要

• 本期主题:Python 综合实战——把第 14-20 期所有技能串成「首页质控自动化流水线」
• 核心结论:真正的「自动化」不是单个脚本,而是「数据 → 校验 → 分析 → 图表 → 报告 → 邮件」的完整流水线
• 6 大模块:数据提取(SQL/Pandas)→ 首页校验(承 17 期)→ DRG 筛查(承 18 期)→ 图表生成(承 16 期)→ PDF 报告(承 20 期)→ 自动邮件(yagmail)
• 典型效果:从「8 小时人工」变成「30 秒自动」
• 阅读时长:约 10 分钟

---

一、场景引入:为什么需要「全流程自动化」?

[!example] 一个真实的月底
月底,质管办小张要做月度质控报告,他按照前面 6 期学的技能:

1. 用 SQL 跑数据 → 30 分钟(第 09 期)
2. Python 校验首页 → 30 分钟(第 17 期)
3. Python 筛 DRG 异常 → 30 分钟(第 18 期)
4. matplotlib 画图 → 30 分钟(第 16 期)
5. reportlab 生成 PDF → 30 分钟(第 20 期)
6. 手动发邮件 → 10 分钟

3 小时过去了,8 个脚本手动跑,中间漏一个就出错。

如果院长说:「这个月数据有更新,重做。」

又 3 小时。

这是「分散技能」的典型困境——每个脚本单独看都对,但串起来手动跑很痛苦。

分散技能	全流程自动化
8 个脚本手动跑	1 个脚本自动跑
中间环节易漏	全流程无遗漏
数据更新重做难	重跑一次就行
每月重复劳动	1 次配置,永久自动

[!quote] 狼叔的判断
“学了 6 个技能不算’会 Python’,能串成流水线才是’真会’。” 这一期,带你把 6 个技能打包成一个完整脚本,一键全跑。

---

二、工具原理:流水线的 6 大模块

2.1 流水线架构

数据源(HIS/MySQL)
    │
    ▼
[模块 1] 数据提取(pandas/SQL)
    │
    ▼
[模块 2] 首页校验(承 17 期)
    │
    ▼
[模块 3] DRG 筛查(承 18 期)
    │
    ▼
[模块 4] 图表生成(承 16 期)
    │
    ▼
[模块 5] PDF 报告(承 20 期)
    │
    ▼
[模块 6] 自动邮件(yagmail)
    │
    ▼
院长/科主任邮箱

2.2 模块化设计原则

[!TIP] 4 大原则

1. 每个模块一个函数:extract_data()、validate()、analyze_drg() 等
2. 模块之间用 DataFrame 传数据,不要传文件路径
3. 主函数串联,类似流水线
4. 日志记录,每一步打印状态,出错了好排查

---

三、医院实战:完整流水线脚本

3.1 完整代码

# -*- coding: utf-8 -*-
"""
首页质控全流程自动化脚本
作者:白衣狼
功能:从 HIS 取数 → 校验 → DRG 分析 → 图表 → PDF → 邮件
"""

import pandas as pd
import pymysql
import matplotlib.pyplot as plt
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle
from reportlab.platypus import (
    SimpleDocTemplate, Paragraph, Table, TableStyle,
    Image, Spacer, PageBreak
)
from reportlab.lib import colors
from reportlab.lib.units import cm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
import yagmail
from datetime import datetime
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('质控自动化.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 数据库配置
DB_CONFIG = {
    'host': '192.168.1.100',
    'port': 3306,
    'user': 'quality_user',
    'password': 'your_password',
    'database': 'his_db',
    'charset': 'utf8mb4'
}

# 邮件配置
EMAIL_CONFIG = {
    'sender': 'quality@hospital.com',
    'password': 'your_email_password',
    'recipients': ['director@hospital.com', 'office@hospital.com']
}

# 注册中文字体
pdfmetrics.registerFont(TTFont('SimHei', 'simhei.ttf'))

# 报告期(上月)
报告月份 = (datetime.now().replace(day=1) - pd.Timedelta(days=1)).strftime('%Y-%m')


# ============================================
# 模块 1:数据提取
# ============================================
def 模块1_数据提取():
    """从 HIS 提取首页数据"""
    logger.info('=' * 50)
    logger.info('[模块 1] 开始数据提取')
    logger.info('=' * 50)

    try:
        conn = pymysql.connect(**DB_CONFIG)

        # 提取首页主表
        首页_sql = f'''
            SELECT 病案号, 出院日期, 科室, 责任医生工号,
                   主诊ICD, 主诊错误标记, 住院天数, 总费用
            FROM 首页主表
            WHERE 出院日期 >= '{报告月份}-01'
              AND 出院日期 < '{报告月份}-32'
        '''
        首页 = pd.read_sql(首页_sql, conn)
        logger.info(f'✓ 首页主表:{len(首页)} 条')

        # 提取诊断表
        诊断 = pd.read_sql(f'''
            SELECT 病案号, ICD编码, 诊断类型
            FROM 诊断表
            WHERE 病案号 IN (SELECT 病案号 FROM 首页主表
                            WHERE 出院日期 >= '{报告月份}-01')
        ''', conn)
        logger.info(f'✓ 诊断表:{len(诊断)} 条')

        # 提取手术表
        手术 = pd.read_sql(f'''
            SELECT 病案号, 手术ICD, 手术名称, 主要操作标记
            FROM 手术表
            WHERE 病案号 IN (SELECT 病案号 FROM 首页主表
                            WHERE 出院日期 >= '{报告月份}-01')
        ''', conn)
        logger.info(f'✓ 手术表:{len(手术)} 条')

        # 提取 DRG 入组结果
        drg = pd.read_sql(f'''
            SELECT 病案号, 入组编码, DRG权重, MDC组
            FROM DRG入组结果
            WHERE 病案号 IN (SELECT 病案号 FROM 首页主表
                            WHERE 出院日期 >= '{报告月份}-01')
        ''', conn)
        logger.info(f'✓ DRG 入组:{len(drg)} 条')

        conn.close()
        return 首页, 诊断, 手术, drg

    except Exception as e:
        logger.error(f'✗ 数据提取失败:{e}')
        raise


# ============================================
# 模块 2:首页校验(承第 17 期)
# ============================================
def 模块2_首页校验(首页, 诊断):
    """首页数据校验"""
    logger.info('=' * 50)
    logger.info('[模块 2] 开始首页校验')
    logger.info('=' * 50)

    错误 = []

    # 必填校验
    必填字段 = ['病案号', '出院日期', '主诊ICD', '责任医生工号']
    for idx, row in 首页.iterrows():
        for field in 必填字段:
            if pd.isna(row[field]) or str(row[field]).strip() == '':
                错误.append({
                    '病案号': row.get('病案号', f'行{idx}'),
                    '类型': '必填校验',
                    '说明': f'{field} 为空',
                    '等级': '严重'
                })

    # 关联一致性:有诊断的病案必须有主诊
    无主诊 = 首页[~首页['主诊ICD'].notna()]['病案号'].tolist()
    for 病案号 in 无主诊:
        错误.append({
            '病案号': 病案号,
            '类型': '关联校验',
            '说明': '病案无主诊 ICD',
            '等级': '严重'
        })

    错误清单 = pd.DataFrame(错误)
    logger.info(f'✓ 校验完成,发现 {len(错误清单)} 条错误')

    return 错误清单


# ============================================
# 模块 3:DRG 异常筛查(承第 18 期)
# ============================================
def 模块3_DRG筛查(首页, 手术, drg):
    """DRG 入组异常筛查"""
    logger.info('=' * 50)
    logger.info('[模块 3] 开始 DRG 筛查')
    logger.info('=' * 50)

    # 误入组:实际 MDC ≠ 理论 MDC
    主要操作 = 手术[手术['主要操作标记'] == 1][['病案号', '手术ICD']] if '主要操作标记' in 手术.columns else 手术[['病案号', '手术ICD']]
    合并 = 首页.merge(主要操作, on='病案号', how='left').merge(drg, on='病案号', how='left')

    def 推算MDC(icd):
        if pd.isna(icd):
            return 'MDCE'
        章节 = str(icd)[:2]
        if 章节 in ['01', '02', '03']: return 'MDCA'
        elif 章节 in ['04', '05', '06']: return 'MDCB'
        elif 章节 in ['07', '08']: return 'MDCC'
        else: return 'MDCD'

    合并['理论MDC'] = 合并['手术ICD'].apply(推算MDC)
    误入组 = 合并[合并['理论MDC'] != 合并['MDC组']].copy()
    误入组['类型'] = '误入组'

    logger.info(f'✓ 误入组:{len(误入组)} 例')

    return 误入组


# ============================================
# 模块 4:图表生成(承第 16 期)
# ============================================
def 模块4_生成图表(首页, drg):
    """生成 4 类核心图表"""
    logger.info('=' * 50)
    logger.info('[模块 4] 开始生成图表')
    logger.info('=' * 50)

    # 1. 科室错误率柱状图
    错误率 = 首页.groupby('科室').agg(
        总数=('病案号', 'count'),
        错误数=('主诊错误标记', 'sum')
    ).reset_index()
    错误率['错误率'] = (错误率['错误数'] / 错误率['总数'] * 100).round(2)
    错误率 = 错误率.sort_values('错误率', ascending=False)

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.bar(错误率['科室'], 错误率['错误率'], color='steelblue')
    ax.set_title('各科室主诊错误率', fontsize=14, fontweight='bold')
    ax.set_xlabel('科室')
    ax.set_ylabel('错误率(%)')
    plt.xticks(rotation=45, ha='right')
    plt.tight_layout()
    plt.savefig('科室错误率.png', dpi=200, bbox_inches='tight')
    plt.close()
    logger.info('✓ 科室错误率.png')

    # 2. 月度趋势折线图
    首页['出院日期'] = pd.to_datetime(首页['出院日期'])
    首页['月'] = 首页['出院日期'].dt.to_period('M')
    月度 = 首页.groupby('月').agg(
        总数=('病案号', 'count'),
        错误数=('主诊错误标记', 'sum')
    ).reset_index()
    月度['错误率'] = (月度['错误数'] / 月度['总数'] * 100).round(2)

    fig, ax = plt.subplots(figsize=(10, 5))
    ax.plot(月度['月'].astype(str), 月度['错误率'],
            marker='o', linewidth=2, color='coral')
    ax.set_title('月度错误率趋势', fontsize=14, fontweight='bold')
    ax.set_xlabel('月份')
    ax.set_ylabel('错误率(%)')
    ax.grid(True, linestyle='--', alpha=0.5)
    plt.tight_layout()
    plt.savefig('月度趋势.png', dpi=200, bbox_inches='tight')
    plt.close()
    logger.info('✓ 月度趋势.png')

    return 错误率, 月度


# ============================================
# 模块 5:PDF 报告(承第 20 期)
# ============================================
def 模块5_生成PDF(首页, 错误清单, 误入组, 错误率, 月度, 输出文件):
    """生成 PDF 报告"""
    logger.info('=' * 50)
    logger.info('[模块 5] 开始生成 PDF')
    logger.info('=' * 50)

    doc = SimpleDocTemplate(输出文件, pagesize=A4)
    故事 = []

    中文样式 = ParagraphStyle('C', fontName='SimHei', fontSize=12, leading=18)
    标题样式 = ParagraphStyle('T', fontName='SimHei', fontSize=20, alignment=1)

    # 封面
    故事.append(Spacer(1, 5 * cm))
    故事.append(Paragraph(f'{报告月份} 月质控报告', 标题样式))
    故事.append(PageBreak())

    # 摘要
    故事.append(Paragraph('一、本月核心指标', 标题样式))
    故事.append(Spacer(1, 0.5 * cm))

    总数 = len(首页)
    错误数 = 首页['主诊错误标记'].sum()
    错误率_总 = round(错误数 / 总数 * 100, 2)

    指标表 = Table([
        ['指标', '数值'],
        ['出院病案总数', f'{总数}'],
        ['主诊错误数', f'{错误数}'],
        ['主诊错误率', f'{错误率_总}%'],
        ['校验问题数', f'{len(错误清单)}'],
        ['DRG 误入组数', f'{len(误入组)}']
    ], colWidths=[6 * cm, 6 * cm])

    指标表.setStyle(TableStyle([
        ('FONT', (0, 0), (-1, -1), 'SimHei', 12),
        ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#1e7e34')),
        ('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
        ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
        ('GRID', (0, 0), (-1, -1), 0.5, colors.grey)
    ]))
    故事.append(指标表)
    故事.append(PageBreak())

    # 图表
    故事.append(Paragraph('二、科室错误率', 标题样式))
    故事.append(Spacer(1, 0.5 * cm))
    故事.append(Image('科室错误率.png', width=14 * cm, height=9 * cm))
    故事.append(PageBreak())

    故事.append(Paragraph('三、月度趋势', 标题样式))
    故事.append(Spacer(1, 0.5 * cm))
    故事.append(Image('月度趋势.png', width=14 * cm, height=7 * cm))
    故事.append(PageBreak())

    # Top 错误科室
    故事.append(Paragraph('四、Top 错误科室', 标题样式))
    故事.append(Spacer(1, 0.5 * cm))

    Top表数据 = [['科室', '总数', '错误数', '错误率']]
    for _, row in 错误率.head(5).iterrows():
        Top表数据.append([
            row['科室'], str(row['总数']),
            str(row['错误数']), f"{row['错误率']}%"
        ])

    Top表 = Table(Top表数据, colWidths=[4 * cm, 3 * cm, 3 * cm, 3 * cm])
    Top表.setStyle(TableStyle([
        ('FONT', (0, 0), (-1, -1), 'SimHei', 11),
        ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#1e7e34')),
        ('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
        ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
        ('GRID', (0, 0), (-1, -1), 0.5, colors.grey)
    ]))
    故事.append(Top表)

    doc.build(故事)
    logger.info(f'✓ PDF 报告:{输出文件}')


# ============================================
# 模块 6:自动邮件
# ============================================
def 模块6_发送邮件(收件人, 主题, 内容, 附件):
    """自动发送邮件"""
    logger.info('=' * 50)
    logger.info('[模块 6] 开始发送邮件')
    logger.info('=' * 50)

    try:
        yag = yagmail.SMTP(
            EMAIL_CONFIG['sender'],
            EMAIL_CONFIG['password']
        )
        yag.send(
            to=收件人,
            subject=主题,
            contents=内容,
            attachments=附件
        )
        logger.info(f'✓ 邮件已发送至:{收件人}')
    except Exception as e:
        logger.error(f'✗ 邮件发送失败:{e}')


# ============================================
# 主函数:流水线串联
# ============================================
def 主流程():
    """首页质控全流程自动化"""
    logger.info('🚀 启动首页质控自动化流水线')
    开始时间 = datetime.now()

    try:
        # 模块 1
        首页, 诊断, 手术, drg = 模块1_数据提取()

        # 模块 2
        错误清单 = 模块2_首页校验(首页, 诊断)

        # 模块 3
        误入组 = 模块3_DRG筛查(首页, 手术, drg)

        # 模块 4
        错误率, 月度 = 模块4_生成图表(首页, drg)

        # 模块 5
        输出文件 = f'质控报告_{报告月份}.pdf'
        模块5_生成PDF(首页, 错误清单, 误入组, 错误率, 月度, 输出文件)

        # 模块 6
        模块6_发送邮件(
            收件人=EMAIL_CONFIG['recipients'],
            主题=f'【{报告月份} 月】医院质控报告',
            内容=f'您好,\n\n本月质控报告已生成,详见附件。\n\n本月出院病案 {len(首页)} 份,主诊错误率 {(首页["主诊错误标记"].sum() / len(首页) * 100):.2f}%。\n\n白衣狼',
            附件=[输出文件]
        )

        耗时 = (datetime.now() - 开始时间).total_seconds()
        logger.info(f'🎉 全流程完成,耗时 {耗时:.1f} 秒')

    except Exception as e:
        logger.error(f'💥 流水线失败:{e}')
        raise


# ============================================
# 入口
# ============================================
if __name__ == '__main__':
    主流程()

3.2 配置定时任务(让脚本每天/每周自动跑)

[!tip] Windows 任务计划程序

Step 1:打开「任务计划程序」(Win+R 输入 taskschd.msc)

Step 2:创建基本任务

• 名称:「首页质控自动化」
• 触发器:每月 1 日 02:00
• 操作:启动程序 → 选 python.exe → 参数填 main.py → 起始位置填脚本目录

效果:每月 1 日凌晨 2 点,脚本自动跑,醒来时报告已在邮箱。

---

四、避坑指南:5 个让流水线「崩溃」的常见错误

[!warning] 雷区 ①:数据库连接超时
症状:pymysql.err.OperationalError: (2003, "Can't connect to MySQL server")。
解决:加连接超时 + 重试机制:

import time
for attempt in range(3):
    try:
        conn = pymysql.connect(connect_timeout=10, **DB_CONFIG)
        break
    except:
        time.sleep(60)  # 1 分钟后重试

[!warning] 雷区 ②:文件路径在脚本目录外
症状:FileNotFoundError。
解决:用绝对路径或 os.path.dirname(__file__):

import os
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
df = pd.read_excel(os.path.join(BASE_DIR, '本月首页.xlsx'))

[!warning] 雷区 ③:邮件密码明文写在脚本里
症状:脚本泄露,邮件密码被别人拿到。
解决:用环境变量:

import os
EMAIL_CONFIG = {
    'sender': os.environ['EMAIL_USER'],
    'password': os.environ['EMAIL_PASS']
}

[!warning] 雷区 ④:大文件内存爆炸
症状:加载 100 万行病历,MemoryError。
解决:分块读取:

chunks = pd.read_sql(sql, conn, chunksize=10000)
for chunk in chunks:
    # 逐块处理
    pass

[!warning] 雷区 ⑤:异常没捕获,整个流水线死
症状:模块 3 出错,模块 4、5、6 全部不跑。
解决:每个模块 try/except + 继续跑:

try:
    错误清单 = 模块2(首页, 诊断)
except Exception as e:
    logger.error(f'模块 2 失败:{e}')
    错误清单 = pd.DataFrame()  # 空 DataFrame 继续

---

五、模块总结:Python 8 期学完了,你获得了什么?

[!ABSTRACT] Python 模块 8 期回顾

期号	主题	核心技能
14	Python 入门	5 个最小例子:变量/列表/字典/函数/文件
15	pandas 入门	DataFrame 10 大操作
16	matplotlib 入门	4 类核心图表
17	首页数据自动校验	5 大校验规则
18	DRG 入组分析	5 类异常筛查
19	病历关键词提取	jieba + TF-IDF
20	质控报告 PDF	reportlab + matplotlib
21	全流程自动化	6 模块流水线

学完这 8 期,你能独立搭建:

• 首页质控自动化流水线(数据 → 校验 → 分析 → 报告)
• 月度质控报告自动生成 + 自动发送
• DRG 异常筛查 + 整改追踪
• 病历关键词分析与科研数据提取

下个模块预告:AI Agent(共 8 期,见开篇规划)—— Claude Code 把 Python 效率放大 5-10 倍。

---

[!TIP] 互动
如果你觉得这篇「医数智联·第 21 期」对你有用,请点赞、在看、转发给科里的同事。

文中完整流水线脚本 + 配置文件 + 定时任务设置指南,我已经整理成「Python 自动化工具包」。点击「阅读原文」可直接下载。

我是白衣狼,一个用数字化工具把医院管理做「轻」的实战派。咱们下期见。

---

[!quote] Python 模块结语
“Python 不是程序员的专利,是医院人的’瑞士军刀’。”

学会了 Python,你就不再被「加班到深夜」绑架——让脚本替你干活,你去做「高价值判断」。

下一模块 AI Agent,我们用 Claude Code 把 Python 的效率再放大 5-10 倍。敬请期待。