② 怎么把「笨重的整件事」拆成可替换的环节——管道流水线架构

目标：学软件架构最核心的思维——拆解 + 可替换。yt-dlp 把"下载一个视频"这件复杂事拆成 4 个可独立替换的环节，这是一个能套到很多场景的通用分解法。

一、问题：事件为什么"笨重"

"下载一个视频"看着简单，真拆开其实混着一堆事：

• 得先搞懂 URL 是哪个网站；
• 得解析页面拿到各种清晰度的下载地址；
• 得选一个格式去下载；
• 下载完可能要转码、合并音视频、改名……

如果全部堆在一个函数里写，你会得到一个几千行、牵一发动全身的函数——这就是"笨重的整件事"。

二、解法：拆成一条可替换的流水线

yt-dlp 把它拆成 4 个阶段，每段只管一件事，用一个总调度器串起来：

URL ─▶ [提取 extract] ─▶ [选择 select] ─▶ [下载 download] ─▶ [后处理 postprocess] ─▶ 文件
        抓页面拿所有清晰度    挑一个格式       真正拉取字节       转码/合并/改名

每个阶段：
  - 有自己独立的"接口"（进来什么、出去什么）
  - 可以被替换成另一个实现

三、为什么这套拆法好（4 个收益）

四、关键设计：接口是什么 + 谁调度

接口（每个阶段的进出约定）：

# 提取：输入 URL，输出 info dict
def extract(url) -> info_dict

# 选择：输入 info dict(含 formats)，输出选中的 format
def select(info) -> format

# 下载：输入 format，输出 filepath
def download(format) -> filepath

# 后处理：输入 filepath+info，输出最终文件
def postprocess(filepath, info) -> final_file

调度器（把四段串起来的人）：

def run(url):
    info   = extract(url)        # ①
    fmt    = select(info)        # ②
    path   = download(fmt)       # ③
    return postprocess(path, info)   # ④

[!important] 调度器不实现任何一段，它只负责"按顺序调"。真正的实现（你要哪个形态的"选择"）靠传入/注册进来。这就是"控制反转"的雏形。

五、yt-dlp 里对应的真实接口

调度器 = YoutubeDL（dl/extract_info/process_video_result…）。详见学习库 → [[Core-Dispatcher]]。

六、可迁移的用法：自己的项目怎么套

适用于"一个流程要做多步、每步可能有多种做法"的场景，例如：数据传输管线、图片处理链、数据清洗管道、爬虫流程。

class Pipeline:
    def __init__(self, stages=None):
        self.stages = stages or []      # 里塞一串"工位"
    def add(self, stage): self.stages.append(stage); return self
    def run(self, data):
        for s in self.stages: data = s(data)   # 上一个的输出喂给下一个
        return data

用的时候：Pipeline().add(去重).add(清洗).add(入库) —— 想加一步/换一步只动列表，不碰别的代码。

七、一句话带走

把"一件笨重的事"拆成一条环节可替换、接口清晰、数据单向流动的管道，谁调都由一个中央调度器统一安排——这是软件架构最核心的思维：单一职责 + 可插拔 + 集中编排。