第 20 章 · 模板方法（Template Method）

一句话：父类把一件事的步骤和顺序写死在一个方法里，只把会变的那几步留给子类填。 解决的痛：几段流程只有中间一步不同，却各抄了一整份；加一个公共步骤要每份都改，漏改的那份照样跑，报告看不出问题。

20.1 一个场景

你在电商公司做商品后台。供应商每次到货都会发一份到货清单，仓库要把它导进系统：每行一个商品，写着 SKU、名称和这次到货的数量，导入时把数量加到库存上。SKU（stock keeping unit）是商品的编号，本章用 A1、B1 这样的短编号。

清单有两种格式。小供应商用表格软件导出 CSV，第一行是表头；大供应商的采购系统直接给 JSON。不管哪种格式，导入都是同样五步：

读取文件内容；
解析成一行行商品；
校验：SKU 不能为空，数量必须大于 0，不合格的行跳过；
写库：数量加到库存上；
生成导入报告，告诉仓库有效几行、跳过了哪些。

本章用字符串模拟文件内容，用字典 stock 模拟库存表，show_stock() 打印当前库存。

20.2 先按直觉写

两种格式，写两个函数，各自从头做到尾：

import json
from dataclasses import dataclass


@dataclass
class Row:
    sku: str
    name: str
    qty: int  # 本次到货数量


stock: dict[str, int] = {}  # 模拟库存表：SKU → 库存数量


def show_stock() -> str:
    return "库存：" + "，".join(f"{sku} {qty}" for sku, qty in stock.items())


def import_csv(content: str) -> str:
    rows: list[Row] = []
    for line in content.strip().splitlines()[1:]:  # 读取、解析：第一行是表头
        sku, name, qty = line.split(",")
        rows.append(Row(sku, name, int(qty)))
    valid = [r for r in rows if r.sku and r.qty > 0]  # 校验
    skipped = [r.sku for r in rows if not (r.sku and r.qty > 0)]
    for r in valid:  # 写库：到货数量加到库存上
        stock[r.sku] = stock.get(r.sku, 0) + r.qty
    return f"CSV：有效 {len(valid)} 行，跳过：{'、'.join(skipped) or '无'}"  # 报告


def import_json(content: str) -> str:
    rows = [Row(d["sku"], d["name"], d["qty"]) for d in json.loads(content)]  # 读取、解析
    valid = [r for r in rows if r.sku and r.qty > 0]  # 校验：和上面一样
    skipped = [r.sku for r in rows if not (r.sku and r.qty > 0)]
    for r in valid:  # 写库：和上面一样
        stock[r.sku] = stock.get(r.sku, 0) + r.qty
    return f"JSON：有效 {len(valid)} 行，跳过：{'、'.join(skipped) or '无'}"  # 报告：和上面一样


if __name__ == "__main__":
    print(import_csv("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n"))
    print(import_json('[{"sku": "A1", "name": "蓝牙耳机", "qty": 50}]'))
    print(show_stock())

# 输出：
# CSV：有效 1 行，跳过：B2
# JSON：有效 1 行，跳过：无
# 库存：B1 30，A1 50

TS 版结构一样：

interface Row {
  sku: string;
  name: string;
  qty: number; // 本次到货数量
}

const stock = new Map<string, number>(); // 模拟库存表：SKU → 库存数量

function showStock(): string {
  return "库存：" + [...stock].map(([sku, qty]) => `${sku} ${qty}`).join("，");
}

function importCsv(content: string): string {
  const rows: Row[] = [];
  for (const line of content.trim().split("\n").slice(1)) { // 读取、解析：第一行是表头
    const [sku, name, qty] = line.split(",");
    rows.push({ sku, name, qty: Number(qty) });
  }
  const valid = rows.filter((r) => r.sku && r.qty > 0); // 校验
  const skipped = rows.filter((r) => !(r.sku && r.qty > 0)).map((r) => r.sku);
  for (const r of valid) stock.set(r.sku, (stock.get(r.sku) ?? 0) + r.qty); // 写库：到货数量加到库存上
  return `CSV：有效 ${valid.length} 行，跳过：${skipped.join("、") || "无"}`; // 报告
}

function importJson(content: string): string {
  const rows = (JSON.parse(content) as Row[]).map((d) => ({ sku: d.sku, name: d.name, qty: d.qty })); // 读取、解析
  const valid = rows.filter((r) => r.sku && r.qty > 0); // 校验：和上面一样
  const skipped = rows.filter((r) => !(r.sku && r.qty > 0)).map((r) => r.sku);
  for (const r of valid) stock.set(r.sku, (stock.get(r.sku) ?? 0) + r.qty); // 写库：和上面一样
  return `JSON：有效 ${valid.length} 行，跳过：${skipped.join("、") || "无"}`; // 报告：和上面一样
}

console.log(importCsv("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n"));
console.log(importJson('[{"sku": "A1", "name": "蓝牙耳机", "qty": 50}]'));
console.log(showStock());

// 输出：
// CSV：有效 1 行，跳过：B2
// JSON：有效 1 行，跳过：无
// 库存：B1 30，A1 50

两个函数只有解析那一两行不同，校验、写库、报告是照着抄的。只有两种格式、流程也不再变的时候，这样写一眼就能看懂，大多数人第一版都这么写。

20.3 需求变了

仓库提了三件事：有的同事用 Excel 填清单，要能导入 Excel；正式导入前要能“试运行”，只校验、出报告，不写库；供应商的文件偶尔会把同一行导出两遍，要按 SKU 去重。

Excel 文件本章用 Excel 另存出来的制表符分隔文本（TSV，tab-separated values）模拟。要改的地方有三处：

Excel 导入。把 import_csv 整个复制一份叫 import_tsv，分隔符换成制表符，报告里的 “CSV” 换成 “Excel”。现在同一套流程有三份。
试运行。每个函数加 dry_run 参数，写库那段套上 if not dry_run，报告末尾加“试运行，未写库”。
去重。每个函数在解析之后插一段：按 SKU 建字典，同一 SKU 只留最后一行，报告里加上去重了几行。

流程有三份，②③ 就得在三份里各写一遍，每份要动解析后、写库、报告三个地方。

小李先改好了 import_csv，import_tsv 是从改好的版本复制的，所以两个都有去重。import_json 是单独改的，加了试运行，漏了去重。改完是这样（Row、stock、show_stock 和 20.2 相同）：

def import_csv(content: str, dry_run: bool = False) -> str:  # ② 加参数
    parsed: list[Row] = []
    for line in content.strip().splitlines()[1:]:
        sku, name, qty = line.split(",")
        parsed.append(Row(sku, name, int(qty)))
    rows = list({r.sku: r for r in parsed}.values())  # ③ 去重：同一 SKU 只留最后一行
    dup = len(parsed) - len(rows)
    valid = [r for r in rows if r.sku and r.qty > 0]
    skipped = [r.sku for r in rows if not (r.sku and r.qty > 0)]
    if not dry_run:  # ② 试运行不写库
        for r in valid:
            stock[r.sku] = stock.get(r.sku, 0) + r.qty
    note = "（试运行，未写库）" if dry_run else ""  # ② 报告注明
    return f"CSV：有效 {len(valid)} 行，去重 {dup} 行，跳过：{'、'.join(skipped) or '无'}{note}"


def import_tsv(content: str, dry_run: bool = False) -> str:  # ① 整个复制 import_csv
    parsed: list[Row] = []
    for line in content.strip().splitlines()[1:]:
        sku, name, qty = line.split("\t")  # ① 只有这里和下面的 "Excel" 不同
        parsed.append(Row(sku, name, int(qty)))
    rows = list({r.sku: r for r in parsed}.values())
    dup = len(parsed) - len(rows)
    valid = [r for r in rows if r.sku and r.qty > 0]
    skipped = [r.sku for r in rows if not (r.sku and r.qty > 0)]
    if not dry_run:
        for r in valid:
            stock[r.sku] = stock.get(r.sku, 0) + r.qty
    note = "（试运行，未写库）" if dry_run else ""
    return f"Excel：有效 {len(valid)} 行，去重 {dup} 行，跳过：{'、'.join(skipped) or '无'}{note}"


def import_json(content: str, dry_run: bool = False) -> str:  # ② 加参数
    rows = [Row(d["sku"], d["name"], d["qty"]) for d in json.loads(content)]
    valid = [r for r in rows if r.sku and r.qty > 0]  # ③ 去重：这一步漏了
    skipped = [r.sku for r in rows if not (r.sku and r.qty > 0)]
    if not dry_run:  # ② 试运行不写库
        for r in valid:
            stock[r.sku] = stock.get(r.sku, 0) + r.qty
    note = "（试运行，未写库）" if dry_run else ""  # ② 报告注明
    return f"JSON：有效 {len(valid)} 行，跳过：{'、'.join(skipped) or '无'}{note}"

供应商的采购系统这次把 A1 导出了两遍，实际只到了 50 件。仓库先试运行，看报告没问题，再正式导入。把上面三个函数换进 before.py，入口换成下面这段（第一行拿同样重复的 CSV 对照）：

if __name__ == "__main__":
    supplier = """[
        {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
        {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
        {"sku": "C1", "name": "手机壳", "qty": 20}
    ]"""  # 供应商导出时 A1 重复了一行
    print(import_csv("sku,name,qty\nA1,蓝牙耳机,50\nA1,蓝牙耳机,50\n", dry_run=True))  # 同样重复，CSV 版
    print(import_json(supplier, dry_run=True))  # 先试运行
    print(import_json(supplier))  # 看着没问题，正式导入
    print(show_stock())

运行结果如下（TS 版改的是同样的地方，输出相同）：

CSV：有效 1 行，去重 1 行，跳过：无（试运行，未写库）
JSON：有效 3 行，跳过：无（试运行，未写库）
JSON：有效 3 行，跳过：无
库存：A1 100，C1 20

同样两行重复的 A1，CSV 版报告“去重 1 行”，JSON 版报告“有效 3 行”，A1 被加了两次，库存 100，比实际到货多了一倍。试运行也没拦住：它走的是同一个漏了去重的函数，报告里没有“去重”这一项，看不出少了什么。前台显示 A1 有 100 件，卖到第 51 件就超卖了。

问题的形状是：同一套流程写了三份，加一个公共步骤要改三处；漏改一份不报错，那一份的报告也照样好看。再加一种格式，又是整份复制，第四份。

直觉写法：三份流程，三处要改，JSON 那份漏了去重

图中 ①–③ 就是上面三处，黄色是为新需求加的，浅红是被改动的步骤。红色加粗的框是那次事故：import_json 里没有去重这一步。

20.4 用模板方法模式重写

换个思路：五个步骤和它们的顺序只写一次，写在基类 Importer 的 run() 里；三种格式唯一不同的“读取、解析”留成抽象方法 parse_rows，由子类实现。去重、校验、写库、报告都是基类自己的方法，子类不用管，也不该覆写。

再留一个钩子（hook）before_save：写库之前调用，试运行时不调，基类里什么也不做。Excel 清单是仓库同事手填的，容易填错，TsvImporter 覆写它，写库前先把库存备份一份；CSV 和 JSON 不需要，就不覆写。

import json
from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class Row:
    sku: str
    name: str
    qty: int  # 本次到货数量


stock: dict[str, int] = {}  # 模拟库存表：SKU → 库存数量
backups: list[dict[str, int]] = []  # 库存备份，Excel 导入前存一份


def show_stock() -> str:
    return "库存：" + "，".join(f"{sku} {qty}" for sku, qty in stock.items())


class Importer(ABC):
    label = ""

    def run(self, content: str, dry_run: bool = False) -> str:  # 模板方法：流程只写在这里
        rows = self.parse_rows(content)  # 子类实现
        rows, dup = self._dedupe(rows)
        valid, skipped = self._validate(rows)
        if not dry_run:  # 试运行：不调钩子，不写库
            self.before_save(valid)  # 钩子：默认什么也不做
            self._save(valid)
        return self._report(valid, dup, skipped, dry_run)

    @abstractmethod
    def parse_rows(self, content: str) -> list[Row]: ...  # 读取、解析：每种格式不同

    def before_save(self, rows: list[Row]) -> None:  # 钩子：子类想覆写就覆写
        pass

    def _dedupe(self, rows: list[Row]) -> tuple[list[Row], int]:  # 同一 SKU 只留最后一行
        by_sku = {r.sku: r for r in rows}
        return list(by_sku.values()), len(rows) - len(by_sku)

    def _validate(self, rows: list[Row]) -> tuple[list[Row], list[str]]:
        valid = [r for r in rows if r.sku and r.qty > 0]
        return valid, [r.sku for r in rows if r not in valid]

    def _save(self, rows: list[Row]) -> None:  # 到货数量加到库存上
        for r in rows:
            stock[r.sku] = stock.get(r.sku, 0) + r.qty

    def _report(self, valid: list[Row], dup: int, skipped: list[str], dry_run: bool) -> str:
        note = "（试运行，未写库）" if dry_run else ""
        return f"{self.label}：有效 {len(valid)} 行，去重 {dup} 行，跳过：{'、'.join(skipped) or '无'}{note}"


def split_lines(content: str, sep: str) -> list[Row]:  # CSV、TSV 共用：第一行是表头
    rows: list[Row] = []
    for line in content.strip().splitlines()[1:]:
        sku, name, qty = line.split(sep)
        rows.append(Row(sku, name, int(qty)))
    return rows


class CsvImporter(Importer):
    label = "CSV"

    def parse_rows(self, content: str) -> list[Row]:
        return split_lines(content, ",")


class JsonImporter(Importer):
    label = "JSON"

    def parse_rows(self, content: str) -> list[Row]:
        return [Row(d["sku"], d["name"], d["qty"]) for d in json.loads(content)]


class TsvImporter(Importer):  # 新加的 Excel 导入：Excel 另存为制表符分隔的文本
    label = "Excel"

    def parse_rows(self, content: str) -> list[Row]:
        return split_lines(content, "\t")

    def before_save(self, rows: list[Row]) -> None:  # Excel 是仓库同事手填的，写库前先备份
        backups.append(dict(stock))
        print(f"Excel：写库前备份库存（{len(stock)} 个 SKU）")


if __name__ == "__main__":
    supplier = """[
        {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
        {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
        {"sku": "C1", "name": "手机壳", "qty": 20}
    ]"""  # 和 20.3 同一份文件，A1 重复了一行
    print(JsonImporter().run(supplier, dry_run=True))
    print(JsonImporter().run(supplier))
    print(CsvImporter().run("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n"))
    print(TsvImporter().run("sku\tname\tqty\nD1\t手机支架\t15\n"))
    print(show_stock())

# 输出：
# JSON：有效 2 行，去重 1 行，跳过：无（试运行，未写库）
# JSON：有效 2 行，去重 1 行，跳过：无
# CSV：有效 1 行，去重 0 行，跳过：B2
# Excel：写库前备份库存（3 个 SKU）
# Excel：有效 1 行，去重 0 行，跳过：无
# 库存：A1 50，C1 20，B1 30，D1 15

TS 版结构相同，固定的步骤标成 private，留给子类的标成 protected：

interface Row {
  sku: string;
  name: string;
  qty: number; // 本次到货数量
}

const stock = new Map<string, number>(); // 模拟库存表：SKU → 库存数量
const backups: Map<string, number>[] = []; // 库存备份，Excel 导入前存一份

function showStock(): string {
  return "库存：" + [...stock].map(([sku, qty]) => `${sku} ${qty}`).join("，");
}

abstract class Importer {
  abstract readonly label: string;

  run(content: string, dryRun = false): string { // 模板方法：流程只写在这里
    const parsed = this.parseRows(content); // 子类实现
    const [rows, dup] = this.dedupe(parsed);
    const [valid, skipped] = this.validate(rows);
    if (!dryRun) { // 试运行：不调钩子，不写库
      this.beforeSave(valid); // 钩子：默认什么也不做
      this.save(valid);
    }
    return this.report(valid, dup, skipped, dryRun);
  }

  protected abstract parseRows(content: string): Row[]; // 读取、解析：每种格式不同

  protected beforeSave(_rows: Row[]): void {} // 钩子：子类想覆写就覆写

  private dedupe(rows: Row[]): [Row[], number] { // 同一 SKU 只留最后一行
    const bySku = new Map(rows.map((r) => [r.sku, r]));
    return [[...bySku.values()], rows.length - bySku.size];
  }

  private validate(rows: Row[]): [Row[], string[]] {
    const valid = rows.filter((r) => r.sku && r.qty > 0);
    return [valid, rows.filter((r) => !valid.includes(r)).map((r) => r.sku)];
  }

  private save(rows: Row[]): void { // 到货数量加到库存上
    for (const r of rows) stock.set(r.sku, (stock.get(r.sku) ?? 0) + r.qty);
  }

  private report(valid: Row[], dup: number, skipped: string[], dryRun: boolean): string {
    const note = dryRun ? "（试运行，未写库）" : "";
    return `${this.label}：有效 ${valid.length} 行，去重 ${dup} 行，跳过：${skipped.join("、") || "无"}${note}`;
  }
}

function splitLines(content: string, sep: string): Row[] { // CSV、TSV 共用：第一行是表头
  return content.trim().split("\n").slice(1).map((line) => {
    const [sku, name, qty] = line.split(sep);
    return { sku, name, qty: Number(qty) };
  });
}

class CsvImporter extends Importer {
  readonly label = "CSV";

  protected parseRows(content: string): Row[] {
    return splitLines(content, ",");
  }
}

class JsonImporter extends Importer {
  readonly label = "JSON";

  protected parseRows(content: string): Row[] {
    return (JSON.parse(content) as Row[]).map((d) => ({ sku: d.sku, name: d.name, qty: d.qty }));
  }
}

class TsvImporter extends Importer { // 新加的 Excel 导入：Excel 另存为制表符分隔的文本
  readonly label = "Excel";

  protected parseRows(content: string): Row[] {
    return splitLines(content, "\t");
  }

  protected beforeSave(_rows: Row[]): void { // Excel 是仓库同事手填的，写库前先备份
    backups.push(new Map(stock));
    console.log(`Excel：写库前备份库存（${stock.size} 个 SKU）`);
  }
}

const supplier = `[
  {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
  {"sku": "A1", "name": "蓝牙耳机", "qty": 50},
  {"sku": "C1", "name": "手机壳", "qty": 20}
]`; // 和 20.3 同一份文件，A1 重复了一行
console.log(new JsonImporter().run(supplier, true));
console.log(new JsonImporter().run(supplier));
console.log(new CsvImporter().run("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n"));
console.log(new TsvImporter().run("sku\tname\tqty\nD1\t手机支架\t15\n"));
console.log(showStock());

// 输出：
// JSON：有效 2 行，去重 1 行，跳过：无（试运行，未写库）
// JSON：有效 2 行，去重 1 行，跳过：无
// CSV：有效 1 行，去重 0 行，跳过：B2
// Excel：写库前备份库存（3 个 SKU）
// Excel：有效 1 行，去重 0 行，跳过：无
// 库存：A1 50，C1 20，B1 30，D1 15

模板方法版：流程在 Importer.run() 里，子类只写解析

看黄色的 TsvImporter：加 Excel 导入就是加这一个类，Importer 一行没动。

一次 Excel 导入：run() 依次调自己的步骤和子类的实现

看箭头的方向：始终是基类的 run() 去调子类的 parse_rows 和 before_save，子类从不调 run() 里的步骤。

逐条对照 20.3：

① Excel 导入：写一个 TsvImporter，实现 parse_rows，需要备份就再覆写 before_save。CSV 和 TSV 的解析只差一个分隔符，放进共用的 split_lines，不用复制。
② 试运行：只在 run() 里写一个 if not dry_run，三种格式都有。
③ 去重：只在 run() 里调一次 _dedupe，三种格式都有。
事故：同一份 JSON，试运行就报告“去重 1 行”，正式导入后 A1 是 50。现在没有“哪一份忘了改”这回事，因为只有一份。
再加一种格式（比如供应商的 XML）：写一个子类，实现 parse_rows。别的都不动。

有三件事要说清楚：

流程顺序被基类定死了。所有子类都是先解析、再去重、再校验。哪天来了一个供应商，同一 SKU 按箱分成好几行，要把数量加起来而不是只留最后一行，_dedupe 就会把前几箱悄悄丢掉。这个供应商用不了 Importer，要么改基类（影响三种格式），要么覆写 run() 再抄一遍流程，又回到 20.3。20.8 会再说这种情况。
“子类不该覆写”多半只是约定。Python 的下划线只是告诉读代码的人“这不是给你改的”，子类照样能覆写 run() 和 _save。TS 的 private 能挡住子类覆写 save，却挡不住覆写 run()，因为 TS 没有 Java 那样的 final。
读子类看不出流程。打开 TsvImporter 只看到两个方法，before_save 什么时候调、试运行时调不调，要回到 Importer.run() 才知道。

模板方法换来的是流程只写一份，加一种格式只加一个子类。代价是流程和步骤分在两个类里，读的时候要来回跳；所有子类共用一个顺序，顺序不一样的需求放不进来。

20.5 模式定义

GoF 对模板方法的定义是：定义一个操作中的算法的骨架，而将一些步骤延迟到子类中。模板方法使得子类可以不改变一个算法的结构即可重定义该算法的某些特定步骤。

换成本章的话：“算法的骨架”就是 run() 里那五步和它们的顺序；“延迟到子类”的步骤是 parse_rows，基类只声明、不实现。子类能决定怎么解析，但决定不了先去重还是先校验。

基类里留给子类的步骤有两种。一种是抽象的，子类必须实现，比如 parse_rows；另一种就是钩子，基类给了默认实现，通常什么也不做，子类需要时才覆写，比如 before_save。钩子让基类在流程里多留几个“插一脚”的位置，又不逼每个子类都去写。

这种结构常被叫作“好莱坞原则”（Hollywood Principle）：别调用我们，我们会调用你。意思是子类不去调基类的流程，只写好自己那几步，等基类的 run() 在该调的时候来调它。

模板方法的结构：templateMethod() 定顺序，子类只实现其中几步

角色	在本章案例里是谁	职责
抽象类（AbstractClass）	`Importer`	在 `run()`（模板方法）里定好步骤和顺序；声明抽象步骤 `parse_rows`，给出钩子 `before_save` 的默认实现，自己实现不变的步骤
具体类（ConcreteClass）	`CsvImporter`、`JsonImporter`、`TsvImporter`	实现抽象步骤，按需覆写钩子，不碰流程

常见场景：

框架的生命周期。框架定好“创建、初始化、处理请求、销毁”的顺序，你继承它的基类，只覆写其中几个方法。
测试夹具（test fixture）。每个测试前准备数据、测试后清理，顺序由测试框架定，20.7 的例一就是。
数据处理流水线。导入、导出、报表都是“读、转换、校验、写、汇报”，只有读和写因格式而不同，本章就是一例。
协议处理器。基类负责读请求、分派、写响应，子类只写各个请求怎么处理。Python 标准库的 http.server.BaseHTTPRequestHandler 读完请求行后，按方法名去调子类的 do_GET、do_POST。

20.6 更简洁的写法

本章的子类大多只实现了 parse_rows 一个方法。和 17.6、19.6 一样，只有一个方法要填，就可以不写子类，把这一步当作函数参数传给一个流程函数：

import json
from collections.abc import Callable
from dataclasses import dataclass


@dataclass
class Row:
    sku: str
    name: str
    qty: int  # 本次到货数量


stock: dict[str, int] = {}  # 模拟库存表：SKU → 库存数量
Parse = Callable[[str], list[Row]]  # 唯一会变的一步：一个函数类型


def parse_csv(content: str) -> list[Row]:
    return [Row(s, n, int(q)) for s, n, q in (line.split(",") for line in content.strip().splitlines()[1:])]


def parse_json(content: str) -> list[Row]:
    return [Row(d["sku"], d["name"], d["qty"]) for d in json.loads(content)]


def run_import(content: str, parse: Parse, label: str, dry_run: bool = False) -> str:  # 流程函数
    parsed = parse(content)  # 变的那一步，由参数传进来
    rows = list({r.sku: r for r in parsed}.values())  # 去重
    valid = [r for r in rows if r.sku and r.qty > 0]
    skipped = [r.sku for r in rows if r not in valid]
    if not dry_run:
        for r in valid:
            stock[r.sku] = stock.get(r.sku, 0) + r.qty
    note = "（试运行，未写库）" if dry_run else ""
    return f"{label}：有效 {len(valid)} 行，去重 {len(parsed) - len(rows)} 行，跳过：{'、'.join(skipped) or '无'}{note}"


if __name__ == "__main__":
    supplier = '[{"sku": "A1", "name": "蓝牙耳机", "qty": 50}, {"sku": "A1", "name": "蓝牙耳机", "qty": 50}]'
    print(run_import(supplier, parse_json, "JSON"))
    print(run_import("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n", parse_csv, "CSV", dry_run=True))
    print("库存：" + "，".join(f"{sku} {qty}" for sku, qty in stock.items()))

# 输出：
# JSON：有效 1 行，去重 1 行，跳过：无
# CSV：有效 1 行，去重 0 行，跳过：B2（试运行，未写库）
# 库存：A1 50

TS 版同样把解析函数传进去：

interface Row {
  sku: string;
  name: string;
  qty: number; // 本次到货数量
}

const stock = new Map<string, number>(); // 模拟库存表：SKU → 库存数量
type Parse = (content: string) => Row[]; // 唯一会变的一步：一个函数类型

const parseCsv: Parse = (content) =>
  content.trim().split("\n").slice(1).map((line) => line.split(",")).map(([s, n, q]) => ({ sku: s, name: n, qty: Number(q) }));

const parseJson: Parse = (content) =>
  (JSON.parse(content) as Row[]).map((d) => ({ sku: d.sku, name: d.name, qty: d.qty }));

function runImport(content: string, parse: Parse, label: string, dryRun = false): string { // 流程函数
  const parsed = parse(content); // 变的那一步，由参数传进来
  const rows = [...new Map(parsed.map((r) => [r.sku, r])).values()]; // 去重
  const valid = rows.filter((r) => r.sku && r.qty > 0);
  const skipped = rows.filter((r) => !valid.includes(r)).map((r) => r.sku);
  if (!dryRun) {
    for (const r of valid) stock.set(r.sku, (stock.get(r.sku) ?? 0) + r.qty);
  }
  const note = dryRun ? "（试运行，未写库）" : "";
  return `${label}：有效 ${valid.length} 行，去重 ${parsed.length - rows.length} 行，跳过：${skipped.join("、") || "无"}${note}`;
}

const supplier = '[{"sku": "A1", "name": "蓝牙耳机", "qty": 50}, {"sku": "A1", "name": "蓝牙耳机", "qty": 50}]';
console.log(runImport(supplier, parseJson, "JSON"));
console.log(runImport("sku,name,qty\nB1,充电宝,30\nB2,数据线,0\n", parseCsv, "CSV", true));
console.log("库存：" + [...stock].map(([sku, qty]) => `${sku} ${qty}`).join("，"));

// 输出：
// JSON：有效 1 行，去重 1 行，跳过：无
// CSV：有效 1 行，去重 0 行，跳过：B2（试运行，未写库）
// 库存：A1 50

流程还是只有一份，去重、试运行也只写一次。但变的那一步不再是子类覆写的方法，而是调用时传进来的对象，这就是第 17 章的策略：run_import 是上下文，parse_csv、parse_json 是可互换的策略。语言替你省掉的是 Importer 基类和每种格式一个子类。

什么时候继承仍然更合适：

要填的步骤不止一个。TsvImporter 既要解析，又要覆写 before_save，还要给报告一个名字。写成函数参数就是 run_import(content, parse, before_save, label)，再多一步，参数还得加，调用的地方每处都要改。子类把这几步放在一起，有默认实现的钩子不写就是不用。
几个步骤要共享状态。比如某个供应商的导入要先在 parse_rows 里读出批次号，再在 before_save 里拿批次号去核对采购单。子类可以把批次号存在 self 上；分开传的几个函数之间没有地方放它，只能再包一层闭包或类。

20.7 真实系统里的例子

例一：unittest 的 TestCase（Python）

Python 标准库的 unittest 里，每个测试类继承 unittest.TestCase。你写 setUp（准备）、tearDown（清理）和若干 test_ 开头的方法，自己从不调它们：测试运行器对每个测试方法调一次 TestCase.run()，run() 按固定顺序调 setUp、测试方法、tearDown。setUp、tearDown 在 TestCase 里默认什么也不做，就是钩子。测试方法失败，tearDown 照样跑；setUp 抛异常，测试方法和 tearDown 都不跑。为了“准备到一半失败也要清理”，还有 addCleanup 注册的清理函数，setUp 失败也会执行。真实的 run() 还要把结果交给 TestResult、处理跳过和 subTest，大致如此。下面模拟一个 MiniTestCase.run()：

class MiniTestCase:  # 模拟 unittest.TestCase：真实的 run() 还要汇报结果、处理跳过和 addCleanup，这里省略
    def setUp(self) -> None:  # 钩子：默认什么也不做
        pass

    def tearDown(self) -> None:  # 钩子：默认什么也不做
        pass

    def run(self, method_name: str) -> str:  # 模板方法：顺序固定，子类不覆写它
        try:
            self.setUp()
        except Exception as e:  # setUp 失败：测试方法和 tearDown 都不跑
            return f"{method_name} ERROR（setUp：{e}）"
        try:
            getattr(self, method_name)()  # 子类写的测试方法
            outcome = "ok"
        except AssertionError as e:
            outcome = f"FAIL（{e}）"
        except Exception as e:  # 断言以外的异常算 ERROR
            outcome = f"ERROR（{e}）"
        finally:
            self.tearDown()  # 测试失败也会跑
        return f"{method_name} {outcome}"


class TestStock(MiniTestCase):
    def setUp(self) -> None:
        self.stock = {"A1": 10}
        print("setUp：准备库存表")

    def tearDown(self) -> None:
        print("tearDown：清空库存表")

    def test_add(self) -> None:
        self.stock["A1"] += 50
        assert self.stock["A1"] == 60, "A1 应为 60"

    def test_double_write(self) -> None:
        self.stock["A1"] += 50
        self.stock["A1"] += 50  # 20.3 的事故：同一行写了两次
        assert self.stock["A1"] == 60, f"A1 应为 60，实际 {self.stock['A1']}"


class TestNoDb(MiniTestCase):
    def setUp(self) -> None:
        raise ConnectionError("连不上测试库")

    def test_add(self) -> None:
        print("不会运行")


if __name__ == "__main__":
    print(TestStock().run("test_add"))
    print(TestStock().run("test_double_write"))
    print(TestNoDb().run("test_add"))

# 输出：
# setUp：准备库存表
# tearDown：清空库存表
# test_add ok
# setUp：准备库存表
# tearDown：清空库存表
# test_double_write FAIL（A1 应为 60，实际 110）
# test_add ERROR（setUp：连不上测试库）

run() 是模板方法，setUp、tearDown 是钩子，测试方法是子类必须提供的那一步，和本章的 Importer 一一对应。test_double_write 顺带用测试复现了 20.3 的事故。

例二：Node 的 stream.Transform（TypeScript）

Node 的 node:stream 里，Transform 是“一边写进去、一边读出来”的流，比如压缩、加解密、按行切分。写一个转换流，继承 Transform，实现 _transform(chunk, encoding, callback)：拿到一块数据，处理完交给 callback；需要在结尾补点什么，再实现可选的 _flush(callback)。其余的事都在基类里：接收写入、缓冲数据、在写端结束时调 _flush 并发出 "end"。读的一方跟不上时，基类会暂停调 _transform，让写的一方等着，这叫背压（backpressure），细节由 highWaterMark 等参数决定，大致如此。Node 文档写明，_transform 这类下划线方法由基类内部调用，应用代码不要直接调。下面用真实的 node:stream 写一个转大写的流：

import { Transform, type TransformCallback } from "node:stream";

const calls: string[] = []; // 记下基类在什么时候调了子类的哪个方法

class UpperCase extends Transform { // 只写两个步骤，流程归 Transform
  override _transform(chunk: Buffer, _encoding: BufferEncoding, callback: TransformCallback): void {
    calls.push(`_transform 收到 ${JSON.stringify(chunk.toString())}`);
    callback(null, chunk.toString().toUpperCase()); // 交回基类，基类放进可读端的缓冲区
  }

  override _flush(callback: TransformCallback): void { // 可选：写端结束后、发 "end" 之前调一次
    calls.push("_flush");
    callback(null, "-- END --\n");
  }
}

const upper = new UpperCase();
const out: string[] = [];
upper.on("data", (chunk: Buffer) => out.push(chunk.toString()));
upper.on("end", () => {
  calls.forEach((line) => console.log(line));
  console.log(out.join("").trimEnd());
});
upper.write("sku,name\n"); // 往写端写，基类决定什么时候调 _transform
upper.write("a1,earbuds\n");
upper.end(); // 写端结束：基类调 _flush，再发 "end"

// 输出：
// _transform 收到 "sku,name\n"
// _transform 收到 "a1,earbuds\n"
// _flush
// SKU,NAME
// A1,EARBUDS
// -- END --

Transform 内部的写入、缓冲、结束流程就是模板方法，_transform 是子类必须实现的步骤，_flush 是钩子。输出里 _transform 和 _flush 的调用都不在我们的代码里，是基类在 write 和 end 之后调的，这就是好莱坞原则。

20.8 什么时候别用

只有一种实现。如果只有 CSV 一种格式，写一个 Importer 基类再写一个 CsvImporter，读代码的人要在两个类之间跳，才看懂一个函数就能写完的事。等第二种格式真的来了，再把公共流程提到基类，不迟。
子类需要改流程顺序。20.4 说的按箱分行的供应商，要的是“合并数量”而不是“只留最后一行”；还有的供应商要求整个文件先校验，有一行不合格就整批拒收，而 run() 是逐行跳过。顺序不同，模板就太死了：要么给基类加开关，基类越来越多 if；要么子类覆写 run() 再抄一份流程。这时把“去重”“校验”这些步骤也做成可替换的对象，由调用方组装（第 17 章的策略），比继承合适。
继承层级已经很深。比如 Importer 下面有 FileImporter，再下面是 SupplierImporter，最下面才是某个供应商的 AcmeCsvImporter，每一层都覆写过一两个步骤。现在要改 AcmeCsvImporter 的 before_save，得先读完上面三个文件：哪一层的 run() 在调它，SupplierImporter 是不是已经覆写过，要不要调 super().before_save()。漏调了，上一层写在那里的库存备份就不跑了，不报错，等哪天要还原时才发现没有备份。

20.9 容易混淆的模式

模式	意图	区别
模板方法（Template Method）	父类定流程骨架，子类填具体步骤	靠继承：变的那步是子类覆写的方法，对象创建出来就定了；变的是流程里的一两步
策略（Strategy）	把一族可互换的算法各自封装	靠组合：变的那步是持有的对象或传进来的函数，运行时也能换；换掉的可以是一整个算法，也可以只是流程里的一步（20.6）。见第 17 章
工厂方法（Factory Method）	定义创建对象的接口，让子类决定实例化哪个类	结构和模板方法一样，只是子类填的那一步恰好是“创建一个对象”，常常就是某个模板方法里的一步。见第 3 章 3.5

两组都能用一个问题当场判断：

模板方法还是策略：问“换算法要不要新建子类”。本章加 Excel 导入写了 TsvImporter 子类，导入器创建出来就只会解析 TSV，是模板方法。20.6 的 run_import 换格式只要传另一个函数，运行时随时能换，是策略。第 17 章用同一个问题区分过两者。
模板方法还是工厂方法：问“子类填的那一步是不是在创建对象”。本章的 parse_rows 返回的是解析结果，是模板方法里的普通步骤。第 3 章的 Notifier.notify() 是一段固定流程，其中 create_sender() 由子类决定创建哪个 Sender，这一步是工厂方法，整个 notify() 是模板方法。

20.10 练一练

报表生成器：写一个基类 Report，构造时传入标题和若干行 (商品名, 金额)。render() 是模板方法，按顺序生成页眉、每一行正文、合计行、页脚，用换行连起来返回。写两个子类：MarkdownReport 生成 Markdown 表格，页脚写”（金额单位：元）“；CsvReport 生成 name,amount 开头的 CSV，没有页脚。CSV 有时要交给别的程序读，合计行会被当成一件商品，所以 CsvReport 要能关掉合计行。

用“蓝牙耳机 3950、充电宝 2670”两行数据，分别打印 Markdown 报表、带合计的 CSV、不带合计的 CSV。Python 和 TS 各写一版。

提示：页眉、正文行、合计行是抽象步骤；页脚和“要不要合计行”是钩子，基类给默认值（没有页脚、要合计），子类按需覆写。TS 里 Report 和浏览器的一个全局类型重名，文件开头加一行 export {}; 把它变成模块。

20.11 小结

模板方法把步骤和顺序写在基类的一个方法里，会变的步骤留成抽象方法或钩子，由子类填。
公共步骤只写一份，加一个步骤改一处，加一种变体加一个子类；不会再有“哪一份漏改了”。
代价是顺序被基类定死，顺序不同的需求放不进来；读子类要回到基类才看得懂流程，层级越深越难读。
只有一步会变时，把它当函数参数传进流程函数就够了，那就是策略；几步一起变、要共享状态时，继承更合适。
和策略的区别看换算法要不要新建子类；工厂方法是“子类那一步恰好是创建对象”的模板方法。