第 05 章 · 文件与批处理自动化

本章目标：这是 Python 相对 SAS 优势最明显、见效最快的领域。学完本章你就能写出自动化脚本，把"复制粘贴改文件名"这类苦力活消灭掉。对应案例：cases/case06_批处理自动化.py

5.1 为什么这是最值得先学的部分

PharmaSUG 2018 的论文 Why SAS Programmers Should Learn Python Too（AD-12）统计过：临床程序员日常最耗时的非统计分析工作是——

从 QC 跟踪表里把几百个程序名/输出名一个个敲出来
批量重命名、批量改程序头、批量归档
从几十个日志文件里提取 ERROR / WARNING
汇总多个文件夹里的输出清单

这些事情用 SAS 做很别扭（SAS 不擅长文件系统操作），用 Python 做是几行代码。 PharmaSUG China 2019 的 CC-036 论文就是纯讲"用 Python 自动创建 SAS 程序文件"的。

💡 务实建议：不要一上来就想用 Python 做统计分析（SAS 在这块更成熟）。 先用 Python 干掉你的文书性工作量——这一步的 ROI 最高，而且能让你在不影响主要交付的前提下快速积累 Python 手感。

5.2 路径处理：`pathlib` 是你的好朋友

Python
from pathlib import Path

# 定义路径（跨平台，Windows/Linux 都能跑）
data_dir = Path("data/raw")
fpath = data_dir / "dm.xpt"              # 用 / 拼接路径（不是字符串相加！）
print(fpath)                             # data\raw\dm.xpt

# 路径的组成部分
fpath.name          # 'dm.xpt'            文件名
fpath.stem          # 'dm'                不含扩展名
fpath.suffix        # '.xpt'             扩展名
fpath.parent        # 'data/raw'          父目录
fpath.resolve()     # 绝对路径（很重要，出问题时用它确认）

# 判断与创建
fpath.exists()                  # 文件存在吗
fpath.is_file()                 # 是文件吗
fpath.is_dir()                  # 是目录吗
data_dir.mkdir(parents=True, exist_ok=True)   # 建目录（已存在不报错）

⚠️ 不要用字符串拼路径： "data\\raw\\" + "dm.xpt" 在 Windows 上凑巧能用，在 Linux 上会挂。用 Path 对象，/ 运算符会自动处理分隔符。

绝对路径的坑（很多新手卡在这里）

Python
# ❌ 危险：相对路径取决于"你从哪个目录运行脚本"
df = pd.read_csv("data/samples/dm.csv")

# ✅ 稳妥：以脚本自身位置为基准
BASE = Path(__file__).resolve().parent.parent     # 项目根目录
df = pd.read_csv(BASE / "data" / "samples" / "dm.csv")

原因：你从 cases/ 运行脚本和从项目根目录运行脚本， "当前工作目录"是不同的，相对路径就会失效。

Python
# 一个通用写法（放在每个脚本开头）
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
SAMPLE_DIR = DATA_DIR / "samples"
OUT_DIR = BASE_DIR / "outputs"
OUT_DIR.mkdir(exist_ok=True)

5.3 遍历文件：`glob` 与 `rglob`

Python
from pathlib import Path

# 当前目录下所有 .xpt
list(RAW_DIR.glob("*.xpt"))

# 递归查找所有 .csv（含子目录）
list(BASE_DIR.rglob("*.csv"))

# 多模式匹配
list(RAW_DIR.glob("ad*.xpt"))         # adsl, adae, adtte ...

# 只取文件名（不带路径）
[f.name for f in RAW_DIR.glob("*.xpt")]

# 只取域名（去掉扩展名）
[f.stem for f in RAW_DIR.glob("*.xpt")]

# 按模块名排序，便于输出稳定
files = sorted(RAW_DIR.glob("sdtm/*.xpt"))

💡 为什么要 sorted()：glob 返回的顺序依赖文件系统，不稳定。临床程序要求输出可复现，凡是遍历文件都要显式排序。

5.4 实战 1：批量重命名（对应 SAS 里最烦的活）

场景：一堆文件叫 dm_final_v2.xpt、ae_FINAL_v3.xpt，要统一改成 SDTM 规范的小写域名格式 dm.xpt、ae.xpt。

Python
import re
from pathlib import Path

def normalize_domain_name(filename: str) -> str:
    """从杂乱文件名中提取域名，统一成小写。

    'dm_final_v2.xpt' -> 'dm.xpt'
    'AE-FINAL-v3.csv' -> 'ae.csv'
    'LbFinal.xpt'     -> 'lb.xpt'
    """
    stem = Path(filename).stem.lower()
    # 匹配开头连续的字母（域名一般是 2-4 个字母）
    m = re.match(r"^([a-z]{2,4})", stem)
    return f"{m.group(1)}.xpt" if m else filename

for f in sorted(RAW_DIR.glob("*.xpt")):
    new_name = normalize_domain_name(f.name)
    if new_name != f.name:
        print(f"{f.name:30s} -> {new_name}")
        # f.rename(f.with_name(new_name))     # 确认无误后再取消注释

⚠️ 安全第一：任何批量重命名/删除脚本，先打印（dry-run），确认清单无误后再执行。这是本教程贯穿始终的原则。

5.5 实战 2：批量转换 XPT → CSV

这是你日常会用到的：把 SAS 数据集批量导出成 Excel/CSV 给非编程同事。

Python
#!/usr/bin/env python
"""把 data/raw/ 下所有 XPT 批量转换成 CSV 到 data/samples/。"""
import logging
from pathlib import Path
import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")

BASE_DIR = Path(__file__).resolve().parent.parent
RAW_DIR = BASE_DIR / "data" / "raw"
OUT_DIR = BASE_DIR / "data" / "samples"

def xpt_to_csv(xpt_path: Path, out_dir: Path) -> tuple[int, int]:
    df = pd.read_sas(xpt_path, format="xport", encoding="utf-8")
    out_path = out_dir / f"{xpt_path.stem}.csv"
    df.to_csv(out_path, index=False, encoding="utf-8-sig")   # utf-8-sig 让 Excel 不乱码
    return df.shape

def main():
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    files = sorted(RAW_DIR.glob("*.xpt"))
    if not files:
        logging.warning("在 %s 下没找到 .xpt 文件", RAW_DIR)
        logging.warning("请先运行: python scripts/download_data.py --core")
        return

    ok = 0
    for f in files:
        try:
            rows, cols = xpt_to_csv(f, OUT_DIR)
            logging.info("%-10s -> %-12s %6d 行 x %d 列", f.name, f"{f.stem}.csv", rows, cols)
            ok += 1
        except Exception as e:
            logging.error("转换 %s 失败: %s", f.name, e)
    logging.info("完成：%d/%d 个文件转换成功", ok, len(files))

if __name__ == "__main__":
    main()

💡 encoding="utf-8-sig" 的意义：Windows 版 Excel 打开纯 UTF-8 的 CSV 会把中文显示成乱码。加 BOM 头（utf-8-sig）就能正常打开。这是"给同事发的 CSV 打不开/乱码"问题的标准解法。

5.6 实战 3：批量扫描 SAS 日志，提取 ERROR / WARNING

PharmaSUG 论文里反复出现的场景。核心是用正则 + 状态机逐行扫描。

Python
import re
from pathlib import Path
from collections import Counter

# SAS 日志里错误行的典型模式
PATTERNS = {
    "ERROR":   re.compile(r"^ERROR", re.I),
    "WARNING": re.compile(r"^WARNING", re.I),
    "UNinit":  re.compile(r"NOTE: .*uninitialized", re.I),
    "MERGE":   re.compile(r"NOTE: .*MERGE statement has more than one data set", re.I),
    "重复":    re.compile(r"NOTE: .*duplicate", re.I),
}

def scan_log(log_path: Path) -> dict:
    """扫描单个日志，返回各问题类别的行号和内容。"""
    findings = {k: [] for k in PATTERNS}
    with log_path.open(encoding="utf-8", errors="replace") as fh:
        for lineno, line in enumerate(fh, start=1):
            text = line.strip()
            for kind, pat in PATTERNS.items():
                if pat.search(text):
                    findings[kind].append((lineno, text[:160]))
    return findings

def scan_all(log_dir: Path) -> None:
    totals = Counter()
    for log in sorted(log_dir.glob("*.log")):
        findings = scan_log(log)
        issues = sum(len(v) for v in findings.values())
        if issues:
            print(f"\n=== {log.name} （{issues} 处问题）===")
            for kind, items in findings.items():
                for lineno, text in items:
                    print(f"  [{kind}] L{lineno}: {text}")
                    totals[kind] += 1
        else:
            print(f"=== {log.name} 无异常 ===")
    print("\n--- 汇总 ---")
    for kind, n in totals.most_common():
        print(f"{kind:10s} {n}")

扩展：同时生成一份 Excel 巡检报告

Python
def scan_all_to_excel(log_dir: Path, out_xlsx: Path) -> None:
    rows = []
    for log in sorted(log_dir.glob("*.log")):
        for kind, items in scan_log(log).items():
            for lineno, text in items:
                rows.append({"程序": log.name, "类别": kind, "行号": lineno, "内容": text})
    df = pd.DataFrame(rows)
    with pd.ExcelWriter(out_xlsx, engine="openpyxl") as w:
        df.to_excel(w, sheet_name="日志巡检", index=False)
    print(f"已输出 {out_xlsx}（{len(df)} 条）")

5.7 实战 4：从 Excel 跟踪表批量生成 SAS 程序骨架

这是 PharmaSUG China 2019 CC-036 论文的方案——自动化最高价值的应用。你不再需要一个个复制粘贴程序头和 %let 语句。

Python
"""读取 QC 跟踪表，为每个 TLF 批量生成带标准程序头的 .sas 骨架。"""
from pathlib import Path
import pandas as pd

TEMPLATE = """/* ============================================================
 * 项目/研究  : {study}
 * 程序名      : {pgm}.sas
 * 程序用途    : {purpose}
 * 输入数据集  : {indata}
 * 输出        : {output}
 * 创建人      : {author}
 * 创建日期    : {date}
 * ============================================================ */

%let pgm    = {pgm};
%let out    = {output};
%let indata = {indata};

/* --- 主逻辑 --- */
"""

def generate_skeletons(tracking_xlsx: Path, out_dir: Path, author="", date="") -> int:
    spec = pd.read_excel(tracking_xlsx)
    # 期望列：PGM | PURPOSE | INDATA | OUTPUT | STUDY
    out_dir.mkdir(parents=True, exist_ok=True)
    n = 0
    for _, row in spec.iterrows():
        content = TEMPLATE.format(
            study=row.get("STUDY", ""), pgm=row["PGM"], purpose=row.get("PURPOSE", ""),
            indata=row.get("INDATA", ""), output=row.get("OUTPUT", ""),
            author=author, date=date,
        )
        (out_dir / f"{row['PGM']}.sas").write_text(content, encoding="utf-8")
        n += 1
    return n

这段代码的价值：一个 CSR 可能有 几百个 TLF 程序。手工创建 + 填表头，一个人要花好几天，还容易打错字。 Python 版本几秒钟跑完，且零打字错误。

5.8 文件读写：几个必知要点

Python
from pathlib import Path

# 文本读写
p = Path("outputs/note.txt")
p.parent.mkdir(parents=True, exist_ok=True)

p.write_text("第一行\n第二行\n", encoding="utf-8")     # 写（覆盖）
text = p.read_text(encoding="utf-8")                   # 读全部
lines = p.read_text(encoding="utf-8").splitlines()     # 按行

# 追加（对应 SAS 的 mod 模式）
with p.open("a", encoding="utf-8") as fh:
    fh.write("第三行\n")

# 逐行读大文件（避免一次性进内存）
with p.open(encoding="utf-8", errors="replace") as fh:
    for line in fh:
        if "ERROR" in line:
            print(line.rstrip())

Python
# 表格读写
import pandas as pd

df = pd.read_csv("data/samples/dm.csv")                 # 默认 utf-8
df = pd.read_csv("x.csv", encoding="gbk")                # 中文 Excel 导出的常见编码
df = pd.read_excel("x.xlsx", sheet_name="Sheet1")
df = pd.read_sas("data/raw/dm.xpt", format="xport")      # SAS XPORT
df = pd.read_sas("x.sas7bdat", format="sas7bdat")        # SAS 数据集（注意编码）

df.to_csv("out.csv", index=False, encoding="utf-8-sig")
df.to_excel("out.xlsx", index=False)

⚠️ index=False 必加。否则 pandas 会把行号当成一列写进去，下次读回来就多了一列 Unnamed: 0。

5.9 动手练习

dry-run 重命名：写一个脚本，列出 data/samples/ 下所有 CSV，把文件名中的下划线换成中划线（如 ad_sl.csv → ad-sl.csv）， 先只打印、不执行。
批量转 CSV：运行 5.5 节的脚本，把 data/raw/ 里的 XPT 全部转成 CSV。
日志巡检：自己造两个假的 .log 文件（里面各写几行 ERROR / WARNING），用 5.6 节的脚本扫描，确认能正确提取。
程序骨架生成：手工造一个 5 行的 Excel 跟踪表（列：PGM/PURPOSE/INDATA/OUTPUT），用 5.7 节的脚本生成 5 个 .sas 骨架文件。
（综合） 写一个脚本：遍历 data/samples/ 下所有 CSV，输出一份清单表格（文件名、行数、列数、文件大小），存成 outputs/data_inventory.xlsx。

5.10 本章小节：SAS ↔ Python 文件操作对照

任务	SAS	Python
列目录	`%sysfunc(dread())` + 循环	`Path.glob()`
文件是否存在	`%sysfunc(fileexist())`	`Path.exists()`
复制文件	系统命令 / `%sysfunc`	`shutil.copy2()`
重命名	系统命令	`Path.rename()`
建目录	`%sysfunc(dcreate())`	`Path.mkdir()`
读文本	`data _null_; infile; input;`	`Path.read_text()` / `open()`
写文本	`data _null_; file; put;`	`Path.write_text()` / `open("w")`
读 Excel	`PROC IMPORT dbms=xlsx`	`pd.read_excel()`
写 Excel	`PROC EXPORT dbms=xlsx`	`df.to_excel()`
读 SAS 数据集	`set adsl;`	`pd.read_sas(..., format="sas7bdat")`
正则匹配	`PRXMATCH`	`re.search`

上一章 ← 第 04 章 · 控制流、函数、模块与异常 下一章 → 第 06 章 · NumPy 与向量化思维