[vscs] Nushell 视频中转项目设计文档

Nov 25, 2025 - ⧖ 9 min
🗒️

1. 项目目标

从一批 百度网盘链接 中：

  • 解析出可下载链接（或构造下载命令）

  • 在 VPS 上以流式方式 进行：

    • 百度网盘 → VPS（下载流）
    • VPS → CDN（上传流）

  • 全过程中 不需要大磁盘空间（VPS 仅作为中转）

  • 通过 任务状态文件 + 日志文件 管理：

    • 任务是否成功
    • 失败原因
    • 数据是否通过校验（可用性）

并且全部逻辑尽量 模块化 + 可替换，用 Nushell 作为主调度和 glue。

2. 总体模块划分

模块拆分如下：

  1. config.nu 全局配置模块：CDN、并发、路径等。
  2. tasks.nu 任务加载与基本操作：从 CSV/JSON 读取任务列表等。
  3. status.nu 任务状态管理：
  • 读写 tasks.sqlite 中的 tasks 表
  • 更新 status / error_type / downloaded_ok / verified_ok 等字段
  • 记录时间戳
  1. baidu.nu 百度盘相关逻辑：
  • 解析分享链接为直链
  • 或构造下载命令参数
  1. cdn.nu CDN 上传逻辑：
  • 对外暴露一个 upload-stream 接口
  • 内部用 rclone / aws s3 / ossutil 等

这里我们选择使用 rclone

  1. transfer.nu 核心传输模块：
  • 针对单个任务执行：下载 → 上传 → 校验 → 状态更新
  • 捕获错误并写入状态 & 日志
  1. main.nu 入口模块：
  • 加载任务列表
  • 并发调度 transfer-one
  • 控制整体流程

可选扩展模块：

  • logger.nu：独立管理 JSONL 日志写入
  • verify.nu：封装各种内容校验逻辑（CDN HEAD / hash / ffprobe 等）

3. 项目目录结构（建议）

project/
  config.nu        # 配置模块
  tasks.nu         # 任务加载模块
  status.nu        # 任务状态管理模块
  baidu.nu         # 百度盘解析模块
  cdn.nu           # CDN 上传模块
  transfer.nu      # 核心传输逻辑
  main.nu          # 入口脚本

  verify.nu        # （可选）校验逻辑
  logger.nu        # （可选）日志模块

  tasks.sqlite     # 任务主表（任务状态的单一事实来源）
  errors.jsonl     # 详细错误日志（追加写）

4. 数据结构设计

4.1 tasks 表字段设计（任务主表）

建议字段：

  • id：任务 ID（唯一）
  • baidu_url：原始百度盘分享链接
  • filename：目标文件名（CDN 中的对象名）
  • direct_url：解析得到的直链（若解析失败可为空）
    • build_tasks.nu 会强制生成 https://pan.baidu.com/... 形式，并在缺少 ?pwd= 时自动拼上提取码。
  • status：任务状态
    • pending / running / success / failed
  • error_type：错误类别
    • 例如：invalid_url / missing_url / download_error / upload_error / verify_failed / unknown
  • error_message：错误详情（可裁剪）
  • broken_url：如果 direct_url 校验失败，把原始链接放到这里，方便人工查错。
  • downloaded_ok：bool（传输过程层面的成功）
  • verified_ok：string（unknown / true / false）
  • attempts：int（尝试次数）
  • last_updated_at：ISO 时间字符串（2025-11-24T10:00:00+08:00）

示例：

id,baidu_url,filename,direct_url,status,error_type,error_message,downloaded_ok,verified_ok,attempts,last_updated_at
task-001,https://pan.baidu.com/s/xxxx,video1.mp4,,pending,,,,0,
task-002,https://pan.baidu.com/s/yyyy,video2.mp4,,pending,,,,0,

后续由脚本填充 direct_url、更新 status 等。

4.2 数据来源（tasks.sqlite 生成流程）

  • 使用 build_tasks.nu 将 2025长三角 报名数据-独奏组.csv 解析为 tasks.sqlite 中的 tasks 表。
  • 入口位于 baidu/build_tasks.nu，依赖 scripts/parse_baidu_link.jq 对“作品网盘链接”字段做正则解析，以及 scripts/write_tasks_sqlite.py 负责落地 SQLite。
  • 示例命令：
nu build_tasks.nu \
  --input "2025长三角 报名数据-独奏组.csv" \
  --database tasks.sqlite

运行后会生成 360 行任务，并自动填充以下额外字段（方便后续过滤/校验）：

  • group_label / group_slug：报名原始组别 & 适合做路径的 slug。
  • performer_name：参赛者姓名（任务上下文快速查看）。
  • baidu_code：提取码（支持从文本或 ?pwd= 中自动提取）。
  • raw_link_text：原始 CSV 中的“作品网盘链接”原文，方便追溯。
  • issues：解析阶段发现的告警（如 missing_url / missing_code）。
  • notes：沿用 CSV 的 备注 字段，便于人工补充信息。
  • source_row：在原始 CSV 中的行号，定位原始数据用。

运行后 tasks.sqlite 中的 tasks 表会自动重建，具备：

  • PRIMARY KEY (id)，保证任务唯一性（同一个 “编号” 只会保留一条，脚本发现重复会直接报错）。
  • 针对 status、group_slug、direct_url 的索引，方便后续查询调度。
  • 与 CSV 相同的字段集合，便于在 Nushell / Python / sqlite3 CLI 里做复杂过滤。

后续模块默认读取 tasks.sqlite 即可，不需要再重新访问原始 CSV。

5. 各模块代码骨架

下面的 Nushell 代码都是 骨架级 示例，你可以根据实际命令、Nushell 版本细节进行微调。

5.1 config.nu – 全局配置

# config.nu

# 返回全局配置为一个 record
export def get-config [] {
  {
    cdn_type: "s3"                # "s3" / "oss" / "cos" / "qiniu" / ...
    bucket: "my-video-bucket"
    region: "ap-southeast-1"
    base_path: "videos/from-baidu" # CDN 里的前缀
    max_parallel: 2               # 并发任务数量
    tasks_db: "tasks.sqlite"      # 任务主表（SQLite）
    error_log_file: "errors.jsonl"# 错误日志文件
    verify_mode: "head"           # "none" / "head" / "hash" ...
  }
}

注意！！！

这里直接使用 rclone 本身已经配置好的cdn项即可

所以不需要以上配置，只需要直接写 type: rclone, remotes: r2, 之后就是 bucket 里的 path 之类的

5.2 tasks.nu – 任务加载模块

# tasks.nu
use config.nu [get-config]

# 加载任务表，返回表格数据
export def load-tasks [] {
  let cfg = get-config

  open $cfg.tasks_file
  | from csv
}

# 保存任务表（全表回写）
export def save-tasks [tasks] {
  let cfg = get-config

  $tasks
  | to csv
  | save -f $cfg.tasks_file
}

这里的 save-tasks 主要用于全量更新，也可以只在 status.nu 内部使用。

5.3 status.nu – 任务状态管理模块

# status.nu
use config.nu [get-config]
use tasks.nu [load-tasks save-tasks]

# 工具函数：获取当前时间的 ISO 字符串
def now-iso [] {
  (date now | format date "%Y-%m-%dT%H:%M:%S%:z")
}

# 内部辅助：更新指定 id 的任务
def update-task-internal [id updater] {
  let tasks = load-tasks

  let updated = $tasks
  | each {|row|
      if $row.id == $id {
        ($updater $row)
      } else {
        $row
      }
    }

  save-tasks $updated
}

# 设置状态为 running（开始处理）
export def set-running [id: string] {
  update-task-internal $id {|row|
    $row
    | upsert status "running"
    | upsert last_updated_at (now-iso)
    | upsert attempts ( ( $row.attempts | default 0 ) + 1 )
  }
}

# 标记成功（downloaded_ok + verified_ok）
export def set-success [id: string downloaded_ok: bool verified_ok: string] {
  update-task-internal $id {|row|
    $row
    | upsert status "success"
    | upsert error_type ""
    | upsert error_message ""
    | upsert downloaded_ok $downloaded_ok
    | upsert verified_ok $verified_ok
    | upsert last_updated_at (now-iso)
  }
}

# 标记失败
export def set-failed [
  id: string
  error_type: string
  error_message: string
  downloaded_ok?: bool = false
  verified_ok?: string = "false"
] {
  update-task-internal $id {|row|
    $row
    | upsert status "failed"
    | upsert error_type $error_type
    | upsert error_message $error_message
    | upsert downloaded_ok $downloaded_ok
    | upsert verified_ok $verified_ok
    | upsert last_updated_at (now-iso)
  }
}

这里用 set-running / set-success / set-failed 来集中管理状态更新逻辑，避免每个模块重复改 CSV。

5.4 baidu.nu – 百度盘解析模块

# baidu.nu
use config.nu [get-config]

# 根据 baidu_url 解析 direct_url
# 假设你有外部工具 pan-cli get-link <url>
def resolve-direct-url [url: string] {
  ^pan-cli get-link $url
  | str trim
}

# 对任务表增加 direct_url 字段
export def with-direct-url []: [table -> table] {
  each {|row|
    let url = $row.baidu_url

    if ($url | is-empty) {
      $row
      | upsert direct_url ""
      | upsert error_type "missing_url"
    } else {
      let direct = (resolve-direct-url $url)

      $row
      | upsert direct_url $direct
    }
  }
}

实际解析逻辑可以根据你现有工具微调，这里只是占位骨架。

5.5 cdn.nu – CDN 上传模块

# cdn.nu
use config.nu [get-config]

# 从 stdin 读数据流，并上传到 CDN 指定路径
# path: 相对 base_path 的路径，如 "video1.mp4"
export def upload-stream [path: string] {
  let cfg = get-config

  # 示例：使用 rclone，remote 名为 "cdn"
  # 最终对象路径：cdn:bucket/base_path/path
  let full = $"cdn:($cfg.bucket)/($cfg.base_path)/($path)"

  ^rclone rcat $full
}

如果你改用其他工具，比如 aws s3 cp - s3://bucket/...，只需改这个模块内部实现即可。

5.6 verify.nu – 校验模块（可选）

# verify.nu
use config.nu [get-config]

# 根据配置选择校验方式，返回 "true" / "false" / "unknown"
export def verify-remote-object [filename: string] {
  let cfg = get-config

  if $cfg.verify_mode == "none" {
    "unknown"
  } else if $cfg.verify_mode == "head" {
    # 示例：用 rclone lsjson 或其他命令校验大小等
    # 这里先返回 "true"，你可以自己实现 HEAD 检查逻辑
    "true"
  } else {
    "unknown"
  }
}

5.7 logger.nu – 错误日志模块（可选）

# logger.nu
use config.nu [get-config]

# 追加写错误日志到 JSONL
export def log-error [
  id: string
  phase: string
  error_type: string
  error_message: string
] {
  let cfg = get-config
  let ts = (date now | format date "%Y-%m-%dT%H:%M:%S%:z")

  let record = {
    ts: $ts
    id: $id
    phase: $phase
    error_type: $error_type
    error_message: $error_message
  }

  $record
  | to json
  | append --raw $cfg.error_log_file
}

5.8 transfer.nu – 核心传输模块

# transfer.nu
use status.nu [set-running set-success set-failed]
use cdn.nu [upload-stream]
use verify.nu [verify-remote-object]
use logger.nu [log-error]

# 单任务传输：输入是一行任务 record
export def transfer-one []: [record -> record] {
  each {|row|
    let id = $row.id
    let fname = $row.filename
    let durl = $row.direct_url

    # 标记任务开始
    set-running $id

    if ($durl | is-empty) {
      let msg = "direct_url is empty"
      log-error $id "prepare" "invalid_direct_url" $msg
      set-failed $id "invalid_direct_url" $msg false "false"
      $row | upsert status "failed"
    } else {
      print $"[($id)] start transfer: ($fname)"

      # 下载 + 上传（流式）
      try {
        ^aria2c $durl --max-connection-per-server=16 --continue=true --stdout
        | upload-stream $fname

        # 传输层成功
        let verified = (verify-remote-object $fname)

        if $verified == "false" {
          let msg = "verification failed"
          log-error $id "verify" "verify_failed" $msg
          set-failed $id "verify_failed" $msg true "false"
          $row
          | upsert status "failed"
          | upsert downloaded_ok true
          | upsert verified_ok "false"
        } else {
          set-success $id true $verified
          $row
          | upsert status "success"
          | upsert downloaded_ok true
          | upsert verified_ok $verified
        }

      } catch {|err|
        let msg = ($err | to string)
        log-error $id "transfer" "download_or_upload_error" $msg
        set-failed $id "download_or_upload_error" $msg false "false"

        $row
        | upsert status "failed"
        | upsert downloaded_ok false
        | upsert verified_ok "false"
      }
    }
  }
}

这里的 try/catch、err 结构需要根据你当前 Nushell 版本略微调整，但整体框架就是： 先标记 running → 管道执行 → 根据结果更新状态 & 日志。

5.9 main.nu – 入口模块

# main.nu
use config.nu [get-config]
use tasks.nu [load-tasks save-tasks]
use baidu.nu [with-direct-url]
use transfer.nu [transfer-one]

def main [] {
  let cfg = get-config

  # 加载任务
  let tasks = load-tasks

  # 先解析 direct_url（可选：只对 pending 的任务解析）
  let tasks_with_direct = $tasks
  | with-direct-url

  # 回写更新后的 direct_url
  save-tasks $tasks_with_direct

  # 只处理 pending 的任务
  let pending = $tasks_with_direct
  | where status == "pending"

  # 并发数控制：简单做法是 par-each
  $pending
  | par-each {|row|
      transfer-one $row
    }
}

main

你也可以做更复杂的并发控制，比如分批 take cfg.max_parallel 循环执行等。

6. 总结

  • 项目通过 模块化拆分（config / tasks / status / baidu / cdn / transfer / main），让你可以：

    • 单独替换下载来源（不是百度盘了也没问题）
    • 单独替换 CDN 实现
    • 单独调整状态管理 / 校验策略

  • 任务状态通过 tasks 主表（tasks.sqlite） 管理：

    • status、error_type、error_message
    • downloaded_ok（传输过程是否顺利）
    • verified_ok（内容是否通过额外校验）

  • 详细错误记录通过 errors.jsonl 持久化，以便后期排查问题。

  • 核心传输流程采用 流式管道：

    • aria2c --stdout 下载
    • upload-stream 从 stdin 上传
    • VPS 几乎不占用磁盘空间，适合 40GB 空间的小机子。

这里的流式管道，直接使用 rclone 本身提供的

Nov 25, 2025 - ⧖ 9 min