1.1 数据流与模块化

1.1 数据流与模块化

本节摘要：本节给流水线立第一块地基：数据流。把每段加工看成一个函数——输入一个产物目录，输出一个产物目录，除此之外不依赖任何隐含状态。先给出七段的输入输出契约表，它是全书所有模块的接口定义；再讲模块可替换性的三个实现条件（接口稳定、路径约定、零隐含状态），用「换 TTS 供应商不动剪辑代码」的适配层示例落地；最后讲失败隔离与重跑：每段落盘完成标记，执行器只重跑没完成的段，一条 30 分钟的片子改一个镜头只需重跑两个段。本节的结论很朴素：模块化的单位不是代码文件，而是产物契约。

学习目标

• 写出七段流水线的输入输出契约表，标出每段的典型失败模式。
• 说出模块可替换性的三个条件，并写出一个供应商适配层。
• 解释「完成标记加幂等执行」如何实现只重跑失败段。
• 用三十行以内的分段执行器把失败隔离落进代码。

一、流水线即函数链

把 0.2 节的最小闭环放大，每段加工都长成同一个形状：读输入目录、写输出目录、成功后留下完成标记。整条流水线就是这样的函数链：

 run(script) ──▶ run(storyboard) ──▶ run(assets) ──▶ run(render)
                                                    │
 成片 ◀── run(export) ◀── run(subs) ◀── run(audio) ◀─┘（剪辑段在 render 后接续）

 每个函数的三条纪律：
 1. 只读自己的输入目录，只写自己的输出目录；
 2. 不读别的段的内部状态，段间交流只走契约文件；
 3. 同输入必同输出（幂等），重跑不产生副作用。
​

这个形状的好处是把「复杂系统」降维成「一组可以独立测试的函数」：怀疑哪段有问题，单独重跑那段；要看中间结果，直接打开产物目录。代价是纪律——任何一段偷懒读了别的段的内部变量，模块化就在那个瞬间失效。

二、七段的输入输出契约表

表里的「典型失败模式」不是吓唬人——它们会在后续章节逐个出现并逐个处理。契约表的本质是一份接口文档：任何一段的实现换了人、换了语言、换了供应商，只要输入输出不变，上下游就不受影响。

三、模块可替换性：换 TTS 不动剪辑

可替换性需要三个条件同时成立。接口稳定：上游调用的是「文本进、音频文件出」的接口，而不是某个供应商的专有 SDK 形状。路径约定：音频文件落盘的位置与命名由约定给出（build/audio 目录加镜头号），下游剪辑与字幕段只认文件，不认供应商。零隐含状态：配音段不往剪辑段偷传「这句感情强烈一点」之类的内存提示，一切信息进 JSON。

# tts_adapter.py —— 供应商可替换的 TTS 适配层（接口示意，实现见第 6 章后半本）
from pathlib import Path

class TTSProvider:
    """共同接口：一句台词进，一个音频文件出。"""
    name = "base"

    def synth(self, text, out_path):
        raise NotImplementedError("子类必须实现 synth")

class CloudTTS(TTSProvider):
    name = "cloud"

    def synth(self, text, out_path):
        pass  # 云服务商接口调用：文本上传、音频下载、写到 out_path

class LocalTTS(TTSProvider):
    name = "local"

    def synth(self, text, out_path):
        pass  # 本地模型推理：加载检查点、合成、写到 out_path

def make_voiceover(storyboard, provider):
    audio_dir = Path("build/audio")
    audio_dir.mkdir(parents=True, exist_ok=True)
    for shot in storyboard["shots"]:
        if shot.get("line"):
            out = audio_dir / ("shot_%03d.wav" % shot["shot_id"])
            provider.synth(shot["line"], out)
​

剪辑段消费的是 build/audio 里的文件与分镜 JSON 里的镜头号，至于 provider 是谁它既不知道也不关心——把 CloudTTS 换成 LocalTTS 只改配置文件一处，剪辑代码零改动。同一个模式适用于文生视频供应商、字幕引擎、甚至整段换语言实现：契约在，替换就是局部手术。

四、失败隔离与重跑

流水线越长，一次全绿的概率越低。假设每段成功率九成五，八段连跑一次全绿的概率只剩约六成六——所以「重跑」不是异常处理，是常态设计。方案是三件套：每段一个产物目录、每段一个完成标记、执行器按标记跳过已完成段。

 build/status/ 里的标记决定执行行为：

 script.done ─┐ storyboard.done ─┐ render.done ×（缺）
              ▼                  ▼            ▼
          [跳过 script]     [跳过分镜]   [重跑 render 及其后]
                                          └─ 只花生成段的钱
​

# run_stage.py —— 分段执行、跳过已完成、只重跑失败段（骨架示意）
import subprocess
import sys
from pathlib import Path

STAGES = ["script", "storyboard", "assets", "render", "edit", "audio", "subs", "export"]
STATUS = Path("build/status")

def main():
    targets = sys.argv[1:] or STAGES
    for stage in targets:
        marker = STATUS / (stage + ".done")
        if marker.exists():
            print("跳过已完成段：", stage)
            continue
        print("执行段：", stage)
        subprocess.run([sys.executable, "stages/run_%s.py" % stage], check=True)
        STATUS.mkdir(parents=True, exist_ok=True)
        marker.write_text("ok\n", encoding="utf-8")

if __name__ == "__main__":
    main()
​

两个工程细节。其一，段脚本必须幂等：render 段重跑会覆盖旧的片段文件，不会出现新旧混杂——这也是「每镜一个文件、以镜头号命名」的价值。其二，强制重跑某段用删标记实现：删掉 render.done 再跑执行器，render 及其下游未完成的段会重新执行，而已完成的下游段（如字幕）若产物仍有效可保留标记跳过。判断「下游是否受影响」的依据永远是契约内容变没变，这个话题在第 2.3 节的修改粒度里展开。

本节要点回顾

• 每段流水线是一个函数：读输入目录、写输出目录、留完成标记，段间只走契约文件。
• 七段契约表是全书的接口文档，典型失败模式预告了后续章节的坑位。
• 可替换性三条件：接口稳定、路径约定、零隐含状态；适配层示例给出标准形状。
• 失败隔离靠产物目录加完成标记加重跑执行器，八段连跑的低全绿率让重跑成为常态设计。

契约表标好了每段传什么，下一节把「传的东西」本身写清楚——三个 JSON schema 的字段设计与校验。