┌──────────────────────────────────────────────────────────┐
│                    Web 管理后台 (单页)                     │
│                  Cloudflare Workers + KV                  │
│                                                          │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│  │ 账号凭证  │ │ AI 服务   │ │ 频道管理  │ │ 运行状态  │   │
│  │ B站Cookie │ │ ASR API  │ │ 搜索频道  │ │ 处理记录  │   │
│  │ YT Cookie │ │ 翻译API  │ │ 选合集    │ │ 立即执行  │   │
│  │ GitHub    │ │          │ │ 选分区    │ │ 连通测试  │   │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
│                       │                                  │
│         KV 存储：config / channels / processed / status  │
└───────────────────────┬──────────────────────────────────┘
                        │ 拉取配置 / 回写状态
                        ▼
┌──────────────────────────────────────────────────────────┐
│              GitHub Actions Runner (定时/手动)            │
│                                                          │
│  Cookie续期 → 轮询YT RSS → yt-dlp下载视频+音频            │
│  → ASR转录 → LLM翻译 → B站投稿 → CC字幕 → 合集归档        │
│  → 批量回写KV                                            │
│                                                          │
│  外部服务：YouTube RSS / 用户ASR API / 用户翻译API /      │
│           B站创作者API                                    │
└──────────────────────────────────────────────────────────┘

1. Runner 启动
   ↓
2. 从 Worker KV 拉取全部配置（1次 HTTP 请求，带重试）
   - B站凭证 (SESSDATA / bili_jct / buvid3 / ac_time_value)
   - ASR API 地址 + 密钥
   - 翻译 API 地址 + 密钥
   - YouTube Cookie (可选，存于 config 内)
   - 所有已启用频道列表
   - 已处理视频 ID 去重表
   ↓
3. Cookie 自动续期检查（详见 §8）
   - check_refresh() → 判断是否需要刷新
   - 需要则 refresh() → 获取新凭证
   - 刷新成功 → 回写 KV（带多次重试，失败则兜底保存到 Artifact）
   ↓
4. 逐个频道轮询 YouTube RSS
   - 请求 /feeds/videos.xml?channel_id=xxx
   - 拿到该频道最近约15条视频
   - 与去重表做差集 → 得到"新视频"列表
   ↓
5. 汇总待处理视频
   - 优先处理手动队列中的视频（区块 5 添加的，不受 MAX_VIDEOS_PER_RUN 限制）
   - 然后处理 RSS 发现的新视频
   - 上限保护：RSS 发现的视频单次最多处理 MAX_VIDEOS_PER_RUN 个（默认 5）
   - 按预估总时长动态裁剪：长视频减少单次处理量
   - 手动队列 + RSS 新视频超出 GitHub Actions 配置超时（300 分钟）的部分留到下次 cron
   ↓
6. 逐个视频处理（根据频道配置的 subtitle_mode 决定是否翻译）：

   ┌─ ① 下载视频 + 提取音频
   │   yt-dlp 下载最佳视频+音频合并 → video.mp4（供投稿）
   │   ffmpeg -i video.mp4 -vn -acodec libmp3lame audio.mp3（供 ASR）
   │   如有 YouTube Cookie，加 --cookies 参数
   │
   ├─ ② 语音识别 + 时间戳对齐
   │   MiMo ASR 转录 → 纯文本
   │   stable-ts 强制对齐 → 词级时间戳 → 句子级 SRT
   │
   ├─ ③ 翻译（根据 subtitle_mode 决定）
   │   - subtitle_mode = "translated" 或 "both"：
   │     POST 翻译API → 中文字幕(保持时间轴) + 中文标题
   │     时间戳严格保留 ASR 原始值，仅替换文本（详见 §16.3.4）
   │   - subtitle_mode = "original" 或 "none"：
   │     跳过翻译，标题保留原文
   │
   ├─ ④ 投稿B站
   │   bilibili-api-python 上传视频文件
   │   - 标题 = 翻译后的中文标题（如翻译）/ 原标题（如不翻译）
   │   - 分区 = 频道配置的 tid
   │   - 标签 = 频道配置的默认标签（详见 §6.6）
   │   - 简介 = 按模板生成（含原视频链接）
   │   - 封面 = YouTube 缩略图 URL
   │   - copyright = 频道配置（自有内容建议=1 自制，搬运=2 转载）
   │   → 返回 bvid + aid + cid
   │
   ├─ ⑤ 上传CC字幕（根据 subtitle_mode 决定）
   │   - "translated"：上传中文字幕（lan="zh-Hans"）
   │   - "original"：上传原语言字幕（lan="en" 等）
   │   - "both"：先上传原语言字幕，再上传中文字幕
   │   - "none"：跳过字幕上传
   │
   ├─ ⑥ 加入合集
   │   season/section/episodes/add
   │   - section_id = 频道配置的合集小节
   │   - episodes = [{aid, cid, title, charging_pay}]
   │   → 视频自动归入指定合集
   │
   └─ ⑦ 记录结果到本地累积列表（含处理阶段 stage 字段）

   ※ 单视频失败不阻塞后续视频，记录失败原因和中断阶段后继续
   ※ 投稿间加随机延迟 5-15 秒，避免触发 B 站风控
   ※ 各环节重试策略：
     - ① 下载：yt-dlp 失败重试 2 次（内置 --retries 3）
     - ② ASR + 对齐：call_with_retry 包装，3 次指数退避（见 §16.4）
     - ③ 翻译：call_with_retry 包装，3 次指数退避 + 条数不匹配自动拆分（见 §16.3.4）
     - ④ 投稿上传：失败重试 2 次，间隔 30 秒
     - ⑤ 字幕上传：失败重试 2 次，B 站改版类错误不重试
     - ⑥ 合集追加：失败重试 1 次（非关键步骤，失败仅记录不影响投稿结果）

   ↓
7. 全部处理完毕，批量回写 Worker KV
   - POST /api/pipeline/processed (批量更新去重表，含裁剪逻辑保留近500条)
   - POST /api/pipeline/status (更新运行状态)
   = 仅 2 次写入，远低于 KV 限额

待处理视频 = 手动队列视频（不受限制，优先处理） + RSS 新视频

RSS 新视频处理数量 = min(所有频道 RSS 新视频之和, MAX_VIDEOS_PER_RUN)

示例：
  手动队列: 2 个视频（优先处理，不计入 MAX_VIDEOS_PER_RUN）
  频道A: RSS有2条新视频
  频道B: 无新视频
  频道C: 1条新视频
  → 本次处理 2(手动) + min(3, 5) = 2 + 3 = 5 个视频

输入关键词（如公司频道名）→ 点"搜索"
→ 系统调用 YouTube Data API v3 搜索频道
→ 显示匹配的 YouTube 频道卡片（头像 + 名称 + 订阅数 + 描述）
→ 点"关注"
→ 系统自动填好 channel_id 和频道名
→ 选择：
    ① 投稿到哪个 B 站合集（下拉框，自动拉取 B 站账号下的所有合集）
    ② 投稿分区（下拉框，通过 /api/tids 拉取 B 站分区列表）
    ③ 默认标签（文本框，逗号分隔）
    ④ 投稿类型（自制 copyright=1 / 转载 copyright=2）
    ⑤ 字幕模式（下拉框）：
       - 翻译后上传中文字幕（默认）
       - 上传原语言字幕（不翻译）
       - 双字幕（原语言+中文翻译）
       - 不上传字幕
→ 点"保存"

粘贴 YouTube 频道 ID + 填名称
→ 选合集 → 选分区 → 填标签 → 选投稿类型 → 选字幕模式 → 保存

输入 YouTube 视频 URL（每行一个，支持多行批量添加）
  → 选择目标频道配置（复用已有频道的合集/分区/标签/字幕模式设置）
  → 或单独指定合集/分区/字幕模式
  → 点"添加到队列"
  → 视频进入手动处理队列，下次流水线执行时优先处理

# 安装 wrangler CLI（如尚未安装）
npm install -g wrangler

# 登录 Cloudflare
wrangler login

# 删除 admin_password 键，使后台回到初始化状态
wrangler kv:key delete --binding=YT2BILI_KV "config"

# 或只重置密码（需先读取 config JSON，修改 password 字段后写回）
wrangler kv:key get --binding=YT2BILI_KV "config" > config.json
# 编辑 config.json，将 admin_password 改为新密码的 bcrypt 哈希
wrangler kv:key put --binding=YT2BILI_KV "config" --path=config.json

Cookie: SESSDATA=xxx; bili_jct=xxx; buvid3=xxx

{
  "title": "翻译后的中文标题",
  "tid": 122,
  "tag": "科技,翻译,YouTube",
  "desc": "原视频：https://youtube.com/watch?v=xxx\n频道：公司官方频道\n本视频由自动翻译系统处理",
  "copyright": 1,
  "source": "https://youtube.com/watch?v=xxx",
  "cover": "https://img.youtube.com/vi/xxx/maxresdefault.jpg",
  "no_reprint": 0,
  "videos": [
    {
      "filename": "video.mp4",
      "title": "翻译后的中文标题",
      "cid": 0
    }
  ]
}

{
  "lan": "zh-Hans",
  "data": {
    "body": [
      {"from": 0.0, "to": 3.0, "content": "第一句字幕"},
      {"from": 3.0, "to": 7.0, "content": "第二句字幕"}
    ]
  }
}

{
  "sectionId": 3954033,
  "episodes": [
    {
      "title": "视频标题",
      "aid": 1906473802,
      "cid": 1625992822,
      "charging_pay": 0
    }
  ]
}

检测到 Cookie 即将过期
  ↓
check_refresh() → 判断是否需要刷新
  ↓
refresh() → 调用 B 站刷新接口
  ├─ 1. 获取刷新 CSRF 令牌 (_get_refresh_csrf)
  ├─ 2. 提交刷新请求
  └─ 3. 确认刷新操作 (_confirm_refresh)
  ↓
获取新凭证，覆盖旧的 SESSDATA / bili_jct / ac_time_value
 （buvid3 不参与刷新，保持不变）
  ↓
回写 KV（带多次重试，详见 §8.4）
  ↓
用新凭证继续执行投稿/字幕/合集操作

每次流水线执行时：
  1. 从 KV 拉取 4 个凭证字段（SESSDATA / bili_jct / buvid3 / ac_time_value）
  2. 创建 Credential 对象（含 ac_time_value）
  3. 调用 check_refresh() 检查是否需要刷新
  4. 需要则 refresh() → 获取新凭证（含新的 SESSDATA / bili_jct / ac_time_value）
  5. 刷新成功 → 回写 KV（带重试，详见 §8.4）
  6. 用新凭证执行投稿/字幕/合集
  7. 失败则标红提示用户重新填 Cookie

def refresh_and_save(credential, worker_url, pipeline_token):
    """刷新 Cookie 并安全回写 KV"""
    # 1. 执行刷新
    new_cred = credential.refresh()  # 旧凭证在此步立即失效
    new_data = {
        "bili_sessdata": new_cred.sessdata,
        "bili_jct": new_cred.bili_jct,
        "ac_time_value": new_cred.ac_time_value,
    }

    # 2. 回写 KV（指数退避重试，最多 5 次）
    for attempt in range(5):
        try:
            resp = requests.post(
                f"{worker_url}/api/pipeline/cookies",
                headers={"Authorization": f"Bearer {pipeline_token}"},
                json=new_data,
                timeout=(10, 30),
            )
            resp.raise_for_status()
            return True  # 回写成功
        except Exception as e:
            delay = 3 * (2 ** attempt)
            print(f"KV 回写失败({attempt+1}/5)，{delay}秒后重试: {e}")
            time.sleep(delay)

    # 3. 全部重试失败 → 兜底：保存到 GitHub Actions Artifact
    print("警告：KV 回写全部失败！新凭证保存到 Artifact 作为兜底")
    with open("new_cookies.json", "w") as f:
        json.dump(new_data, f)
    # Artifact 会在 workflow 中自动上传
    # 注意：Artifact 含敏感凭证，仓库必须设为 Private；公共仓库不建议使用此兜底
    # 同时发送通知（如配置了 notify_webhook）
    return False

concurrency:
  group: pipeline-singleton
  cancel-in-progress: false

on:
  schedule:
    - cron: '0 */4 * * *'        # 自动：每4小时
  repository_dispatch:
    types: [process]              # 手动：Web后台触发
  workflow_dispatch:              # 手动：GitHub页面触发

concurrency:
  group: pipeline-singleton       # 确保同一时刻只有一个流水线运行
  cancel-in-progress: false

jobs:
  process:
    runs-on: ubuntu-latest
    timeout-minutes: 300          # 5小时超时（系统上限 360 分钟，留 1 小时余量）
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install ffmpeg
        run: sudo apt-get update && sudo apt-get install -y ffmpeg
      - name: Cache Whisper model
        uses: actions/cache@v4
        with:
          path: ~/.cache/whisper
          key: whisper-model-small
          restore-keys: whisper-model-
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run pipeline
        env:
          WORKER_URL: ${{ secrets.WORKER_URL }}
          PIPELINE_TOKEN: ${{ secrets.PIPELINE_TOKEN }}
        run: python scripts/main.py
      - name: Upload cookie artifact (if exists)
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: cookie-backup
          path: new_cookies.json
          if-no-files-found: ignore
          retention-days: 7

{
  "results": [
    {
      "video_id": "dQw4w9WgXcQ",
      "status": "success",
      "bvid": "BV1xx411c7mD",
      "title": "视频标题",
      "channel": "频道名",
      "stage": "completed",
      "retryable": false
    },
    {
      "video_id": "abc123",
      "status": "failed",
      "bvid": "",
      "title": "视频标题",
      "channel": "频道名",
      "stage": "uploaded",
      "message": "Cookie 可能已过期",
      "retryable": false
    }
  ]
}

{
  "last_run": "2026-07-06T12:00:00Z",
  "total_processed": 42,
  "records": [
    {
      "channel": "公司官方频道",
      "title": "视频标题",
      "status": "success",
      "stage": "completed",
      "time": "2026-07-06T12:05:00Z",
      "message": ""
    }
  ]
}

{
  "bili_sessdata": "新SESSDATA",
  "bili_jct": "新bili_jct",
  "bili_buvid3": "新buvid3",
  "ac_time_value": "新ac_time_value"
}

{
  "config": {
    "bili_sessdata": "xxx",
    "bili_jct": "xxx",
    "bili_buvid3": "xxx",
    "ac_time_value": "xxx",
    "asr_api": "https://...",
    "asr_key": "sk-xxx",
    "translate_api": "https://...",
    "translate_key": "sk-xxx",
    "yt_cookies": "Netscape格式Cookie内容(可选)",
    "notify_webhook": "https://...(可选)"
  },
  "channels": [
    {
      "id": "uuid",
      "channel_id": "UCxxxx",
      "name": "频道名",
      "season_id": 3541247,
      "section_id": 3954033,
      "tid": 122,
      "tags": "科技,翻译",
      "copyright": 1,
      "subtitle_mode": "translated",
      "enabled": true
    }
  ],
  "manual_queue": [
    {
      "video_id": "dQw4w9WgXcQ",
      "url": "https://youtube.com/watch?v=dQw4w9WgXcQ",
      "title": "Never Gonna Give You Up",
      "channel_config_id": "a1b2c3d4-uuid-of-channel-config",
      "config": {
        "season_id": 3541247,
        "section_id": 3954033,
        "tid": 122,
        "tags": "科技,翻译",
        "copyright": 1,
        "subtitle_mode": "translated"
      },
      "added_at": "2026-07-06T12:00:00Z",
      "status": "pending",
      "processing_since": null,
      "retry_count": 0,
      "last_error": null
    }
  ],
  "processed": {
    "videoId1": {
      "bvid": "BVxxx",
      "title": "标题",
      "channel": "频道名",
      "status": "success",
      "stage": "completed",
      "processed_at": 1783344000
    },
    "videoId2": {
      "bvid": "",
      "title": "标题",
      "channel": "频道名",
      "status": "failed",
      "stage": "uploaded",
      "message": "Cookie 可能已过期",
      "processed_at": 1783343000
    }
  }
}

{
  "last_run": "2026-07-06T12:00:00Z",
  "total_processed": 42,
  "records": [
    {
      "channel": "公司官方频道",
      "title": "视频标题",
      "status": "success",
      "stage": "completed",
      "time": "2026-07-06T12:05:00Z",
      "message": ""
    }
  ]
}

优化前（每视频1次写入）:
  5个视频 → 5次写processed + 1次写status = 6次写入/执行

优化后（批量写入 + 裁剪）:
  5个视频 → Runner本地累积
         → 1次批量写processed（含裁剪逻辑，保留近500条）
         → 1次写status
         = 2次写入/执行

Cookie 刷新回写（如有）: +1次写入/执行
手动队列清理（如有手动视频处理完）: +1次写入/执行（manual_queue 是独立 KV Key，需单独写入清理结果）
用户操作（改配置/增删频道/添加手动视频）: 按需，通常每天 <5 次

# 每次回写时裁剪
if len(processed) > 500:
    # 按时间排序，保留最近500条
    sorted_items = sorted(
        processed.items(),
        key=lambda x: x[1].get('processed_at') or 0,
        reverse=True
    )
    processed = dict(sorted_items[:500])

步骤 0：前置准备
  ├─ 注册 GitHub / Cloudflare / Google Cloud 账号
  ├─ 准备 ASR 服务和翻译服务
  ├─ 安装本地开发工具（可选，用于步骤 3 的 secret 配置）：
  │    ├─ Node.js 18+（含 npm）
  │    ├─ npm install -g wrangler  # 安装 Cloudflare CLI
  │    └─ wrangler login            # 登录 Cloudflare 账号
  └─ 在 B 站创作者后台手动创建所需合集与小节（需人工审核通过）

步骤 1：准备 GitHub 仓库
  ├─ Fork 或克隆项目模板到自己的 GitHub 私有仓库
  ├─ 本地克隆：git clone https://github.com/yourname/yt2bili.git
  └─ （Secrets 稍后在步骤 6 回填）

步骤 2：创建 Cloudflare KV
  ├─ 登录 Cloudflare Dashboard
  ├─ Workers & Pages → KV → 创建命名空间
  └─ 复制命名空间 ID

步骤 3：配置 wrangler.toml + 加密密钥
  ├─ 找到项目根目录的 wrangler.toml
  ├─ 将 KV 命名空间 ID 填入 kv_namespaces.id 字段
  ├─ 执行 wrangler secret put ENCRYPTION_KEY，输入 32 字节随机字符串
  └─ push 到主分支

步骤 4：连接 Git 仓库
  ├─ Cloudflare Dashboard → Workers → 连接 GitHub 仓库
  ├─ 选择私有仓库 → 授权
  └─ 主分支推送后自动部署，获得 Worker 域名

步骤 5：初始化 Web 后台
  ├─ 访问 Worker 域名
  ├─ 设置管理密码（首次初始化）
  └─ 在"系统设置"中查看自动生成的 pipeline_token，复制保存

步骤 6：回填 GitHub Secrets
  ├─ GitHub 仓库 → Settings → Secrets and variables → Actions
  ├─ WORKER_URL = Worker 域名（如 https://yt2bili.xxx.workers.dev）
  └─ PIPELINE_TOKEN = 步骤 5 中复制的 pipeline_token

步骤 7：填写配置
  ├─ 后台填写 B 站 Cookie（4 个字段，ac_time_value 从 Console 获取）
  ├─ 后台填写 ASR API + 翻译 API
  ├─ 后台填写 GitHub Token + 仓库
  ├─ 后台添加 YouTube 频道（搜索或手动）
  └─ 每项填完后点"测试"按钮验证连通性

步骤 8：验证部署
  ├─ 后台点"立即执行"
  ├─ 等 5-10 分钟
  └─ 刷新状态页，确认出现处理记录

# 安装依赖
npm install

# 本地运行 Worker
npm run dev

# 手动部署（也可通过 Git 自动部署）
npm run deploy

yt2bili/
├── src/
│   └── index.ts              # Cloudflare Worker 入口（Hono 框架）
├── public/
│   └── index.html            # Web 管理后台（单文件前端）
├── scripts/
│   ├── main.py               # 流水线主入口
│   ├── youtube.py            # YouTube RSS 轮询 + yt-dlp 下载
│   ├── asr.py                # MiMo ASR 调用 + stable-ts 强制对齐
│   ├── translate.py          # 翻译 API 调用（根据 subtitle_mode 决定是否执行）
│   ├── bilibili.py           # B 站投稿 + 字幕 + 合集
│   ├── cookie_refresh.py     # Cookie 自动续期
│   └── srt_utils.py          # SRT 字幕处理工具（parse_srt, entries_to_srt 等）
├── .github/
│   └── workflows/
│       └── process.yml       # GitHub Actions 工作流
├── wrangler.toml             # Cloudflare Workers 配置
├── package.json              # Node.js 依赖
├── tsconfig.json             # TypeScript 配置
├── requirements.txt          # Python 依赖锁定
├── .gitignore
└── README.md                 # 本文档

yt-dlp 下载视频 (mp4) → ffmpeg 提取音频 (mp3)
  ↓
MiMo ASR API → 返回纯文本转录（无时间戳）
  ↓
stable-ts 强制对齐 → 将文本对齐到音频 → 词级时间戳（±0.2s 精度）
  ↓
stable-ts 句子级分段 → 输出 SRT
  ↓
（可选）翻译 API → 返回中文字幕（仅文本，时间戳保留原始值）+ 中文标题
  ↓
组装为 B 站字幕提交格式 (body 数组)

阶段 1：转录（MiMo ASR）
  音频 → MiMo API → 纯文本（无时间戳）

阶段 2：对齐（stable-ts）
  纯文本 + 音频 → stable-ts align() → 词级时间戳（±0.2s）
  → 按句子分组 → 输出 SRT

POST https://api.xiaomimimo.com/v1/chat/completions
api-key: {asr_key}
Content-Type: application/json

import base64
import os
import requests

def call_mimo_asr(audio_path, asr_api, asr_key, language="auto"):
    """
    调用 MiMo ASR API，返回纯文本转录结果。
    音频文件需 wav 或 mp3 格式，Base64 编码后不超过 10MB。
    """
    with open(audio_path, "rb") as f:
        audio_bytes = f.read()

    audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
    if len(audio_b64) > 10 * 1024 * 1024:
        raise ValueError(f"音频 Base64 编码后超过 10MB 限制: {len(audio_b64)} bytes")

    # 根据文件扩展名动态选择 MIME 类型
    ext = os.path.splitext(audio_path)[1].lower()
    mime_map = {".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4"}
    mime_type = mime_map.get(ext, "audio/mpeg")

    resp = requests.post(
        asr_api,
        headers={
            "api-key": asr_key,                    # MiMo 使用 api-key Header
            "Content-Type": "application/json",
        },
        json={
            "model": "mimo-v2.5-asr",
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "input_audio",
                            "input_audio": {
                                "data": f"data:{mime_type};base64,{audio_b64}"
                            }
                        }
                    ]
                }
            ],
            "asr_options": {
                "language": language       # "auto" / "zh" / "en"
            }
        },
        timeout=(10, 600),                 # (connect 10s, read 600s)
    )
    resp.raise_for_status()
    data = resp.json()
    return data["choices"][0]["message"]["content"].strip()

{
  "id": "chatcmpl-xxx",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "Hello everyone, welcome to this video. Today we are going to talk about..."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 0,
    "completion_tokens": 100,
    "total_tokens": 100
  }
}

pip install stable-ts

import stable_whisper

# 全局模型实例，避免重复加载（模型约 461MB，加载需 5-10 秒）
_align_model = None

def get_align_model(model_size="small"):
    """获取或初始化对齐模型（单例模式，整个流水线只加载一次）"""
    global _align_model
    if _align_model is None:
        _align_model = stable_whisper.load_model(model_size)
    return _align_model

def align_text_to_audio(audio_path, transcript_text, language="en", model=None):
    """
    使用 stable-ts 将已有文本对齐到音频，生成带时间戳的字幕。
    
    参数:
        audio_path: 音频文件路径（mp3/wav）
        transcript_text: MiMo ASR 返回的纯文本
        language: 音频语言（"en"/"zh" 等，用于选择对齐模型）
        model: 已加载的 stable-ts 模型实例（可选，不传则自动加载）
    
    返回: stable-ts Result 对象（含词级时间戳）
    """
    # 使用传入的模型或自动加载（推荐在流水线入口处加载一次后传入）
    if model is None:
        model = get_align_model("small")

    # 强制对齐：将文本对齐到音频
    result = model.align(audio_path, transcript_text, language=language)

    # 按句子分组：用 max_chars 控制每条字幕长度（非 max_words，因中文分词不依赖空格）
    if language == "zh":
        result.split_by_length(max_chars=30)    # 中文每条最多 30 字符
    else:
        result.split_by_length(max_chars=80)    # 英文每条最多 80 字符

    # 限制每条字幕最长时长，合并过短段
    result.clamp_max(max_dur=8.0)               # 任何段落不超过 8 秒

    return result

def transcribe_and_align(audio_path, asr_api, asr_key, language="auto"):
    """
    完整流程：MiMo ASR 转录 → stable-ts 对齐 → 输出 SRT
    """
    # 阶段 1：MiMo ASR 转录（获取纯文本）
    transcript = call_mimo_asr(audio_path, asr_api, asr_key, language)

    # 映射语言代码：MiMo 用 "auto"/"zh"/"en"，stable-ts 需明确语言
    # "auto" 默认按英文对齐（Whisper 模型对英文支持最好）；如已知其他语言应显式传入
    align_lang = "zh" if language == "zh" else language if language != "auto" else "en"

    # 阶段 2：stable-ts 强制对齐（生成时间戳）
    model = get_align_model("small")
    result = align_text_to_audio(audio_path, transcript, align_lang, model=model)

    # 输出 SRT 文本
    srt_content = result.to_srt_vtt(word_level=False)  # 句子级 SRT
    return srt_content

if align_lang == "zh":
    result.split_by_length(max_chars=30)
else:
    result.split_by_length(max_chars=80)
result.clamp_max(max_dur=8.0)

import subprocess

def split_audio_if_needed(audio_path, max_size_mb=7):
    """
    如果音频文件超过 max_size_mb，按固定时长切分。
    返回: [(path, start_offset), ...] 切分后的片段及时间偏移
    """
    import os
    file_size = os.path.getsize(audio_path)
    if file_size <= max_size_mb * 1024 * 1024:
        return [(audio_path, 0.0)]  # 无需切分

    # 获取总时长
    cmd = ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
           "-of", "default=noprint_wrappers=1:nokey=1", audio_path]
    duration = float(subprocess.run(cmd, capture_output=True, text=True).stdout.strip())

    # 按目标大小估算每段时长
    seg_duration = (max_size_mb * 1024 * 1024 / file_size) * duration * 0.9  # 留 10% 余量
    n_segments = int(duration // seg_duration) + 1

    segments = []
    for i in range(n_segments):
        start = i * seg_duration
        seg_path = f"audio_part_{i:03d}.mp3"
        # 使用 -accurate_seek + 重新编码（非 -acodec copy），确保切分时间戳精确
        # -acodec copy 在关键帧边界切分，会导致实际起始时间偏移
        cmd = ["ffmpeg", "-i", audio_path, "-ss", str(start),
               "-t", str(seg_duration), "-accurate_seek",
               "-acodec", "libmp3lame", "-q:a", "4", seg_path]
        subprocess.run(cmd, capture_output=True, timeout=120)
        segments.append((seg_path, start))

    return segments


def transcribe_long_audio(audio_path, asr_api, asr_key, language="auto"):
    """
    长音频处理：切分 → 逐段转录+对齐 → 合并 SRT
    模型在循环外加载一次，避免每段重复加载（~461MB 模型加载需 5-10 秒）
    """
    segments = split_audio_if_needed(audio_path)
    all_srt_entries = []

    # 预加载对齐模型（整个函数只加载一次）
    align_lang = "zh" if language == "zh" else language if language != "auto" else "en"
    model = get_align_model("small")

    for seg_path, offset in segments:
        # 阶段 1：MiMo ASR 转录
        transcript = call_mimo_asr(seg_path, asr_api, asr_key, language)

        # 阶段 2：stable-ts 对齐（传入已加载的模型）
        result = align_text_to_audio(seg_path, transcript, align_lang, model=model)

        # 提取条目并加上时间偏移
        for seg in result.segments:
            all_srt_entries.append({
                "start_seconds": seg.start + offset,
                "end_seconds": seg.end + offset,
                "text": seg.text,
            })

    # 重新编号并输出 SRT
    lines = []
    for i, entry in enumerate(all_srt_entries, 1):
        start_str = format_timestamp(entry["start_seconds"])
        end_str = format_timestamp(entry["end_seconds"])
        lines.append(f"{i}\n{start_str} --> {end_str}\n{entry['text']}\n")
    return "\n".join(lines)

def format_timestamp(seconds):
    """秒数转 SRT 时间格式 HH:MM:SS,mmm（修复浮点进位 bug）"""
    if seconds is None or seconds < 0:
        raise ValueError(f"无效的时间戳: {seconds}")
    # 统一转为毫秒整数，避免 round((seconds - int(seconds)) * 1000) 在 0.9999 时进位丢失
    total_ms = round(seconds * 1000)
    hrs = total_ms // 3_600_000
    mins = (total_ms % 3_600_000) // 60_000
    secs = (total_ms % 60_000) // 1_000
    ms = total_ms % 1_000
    return f"{hrs:02d}:{mins:02d}:{secs:02d},{ms:03d}"

def parse_srt(srt_content):
    """解析 SRT 为条目列表 [{index, start, end, start_seconds, end_seconds, text}, ...]"""
    entries = []
    blocks = srt_content.strip().split("\n\n")
    for block in blocks:
        lines = block.strip().split("\n")
        if len(lines) < 3:
            continue
        index = int(lines[0])
        time_line = lines[1]
        start_str, end_str = time_line.split(" --> ")
        text = "\n".join(lines[2:])
        entries.append({
            "index": index,
            "start": start_str,
            "end": end_str,
            "start_seconds": parse_time_to_seconds(start_str),
            "end_seconds": parse_time_to_seconds(end_str),
            "text": text,
        })
    return entries

def entries_to_srt(entries):
    """将条目列表转回 SRT 格式"""
    lines = []
    for i, entry in enumerate(entries, 1):
        lines.append(f"{i}\n{entry['start']} --> {entry['end']}\n{entry['text']}\n")
    return "\n".join(lines)

def parse_time_to_seconds(time_str):
    """SRT 时间格式 HH:MM:SS,mmm 转秒数"""
    h, m, s = time_str.replace(",", ".").split(":")
    return int(h) * 3600 + int(m) * 60 + float(s)

POST {translate_api}
Authorization: Bearer {translate_key}
Content-Type: application/json

import requests
import json

def call_translate(srt_content, original_title, translate_api, translate_key, source_language="auto", context=""):
    resp = requests.post(
        translate_api,
        headers={
            "Authorization": f"Bearer {translate_key}",
            "Content-Type": "application/json",
        },
        json={
            "model": "your-model-name",       # 模型名（可选）
            "source_language": source_language,  # 源语言（"auto"/"en"/"ja"/"ko" 等，由 ASR language 决定）
            "target_language": "zh-Hans",
            "title": original_title,           # 需要翻译的标题
            "subtitle": srt_content,           # SRT 格式字幕全文
            "context": context,                # 上一批末尾的上下文（可选）
            "instructions": "请将字幕和标题翻译为简体中文。仅翻译文本内容，不要修改时间戳、编号或格式。保持语气自然流畅，专业术语使用中文通用译法。",
        },
        timeout=(10, 300),                    # (connect超时10s, read超时300s)
    )
    resp.raise_for_status()
    return resp.json()

{
  "translated_title": "大家好，欢迎观看这个视频",
  "translated_subtitle": "1\n00:00:01,860 --> 00:00:04,459\n大家好，欢迎观看这个视频。\n\n2\n00:00:04,460 --> 00:00:08,250\n今天我们要讨论的是...\n\n...",
  "usage": {
    "prompt_tokens": 5000,
    "completion_tokens": 4500
  }
}

def translate_in_batches(srt_content, original_title, translate_api, translate_key, batch_size=50, source_language="auto"):
    """
    将 SRT 按条数分批翻译，保持上下文连贯。
    时间戳保护：翻译后按顺序逐条配对，强校验条数一致，仅替换 text，时间戳沿用 ASR 原始值。
    条数不匹配时自动拆分重试，避免直接放弃整批。
    batch_size: 每批字幕条数，默认50条
    source_language: 源语言（由 ASR language 决定，透传给翻译 API）
    """
    entries = parse_srt(srt_content)  # 解析为 [{index, start, end, text, ...}, ...]
    batches = [entries[i:i+batch_size] for i in range(0, len(entries), batch_size)]

    all_translated = []
    translated_title = ""

    for i, batch in enumerate(batches):
        batch_srt = entries_to_srt(batch)
        context = ""
        if i > 0:
            # 提供前一批最后几条的原文+译文作为上下文，保持术语一致
            context = "\n".join([
                f"原文: {e['original_text']}\n译文: {e['text']}"
                for e in all_translated[-3:]
            ])

        resp = call_translate(
            batch_srt,
            original_title if i == 0 else "",
            translate_api, translate_key,
            source_language=source_language,
            context=context
        )

        if i == 0:
            translated_title = resp["translated_title"]

        # 解析 LLM 返回的翻译结果
        translated_entries = parse_srt(resp["translated_subtitle"])

        # 时间戳保护：按行号配对，仅替换 text，时间戳沿用 ASR 原始值
        # 条数不匹配时，缩小 batch_size 重试（而非直接报错放弃整批）
        if len(translated_entries) != len(batch):
            if len(batch) > 10:
                # 递归拆分：将当前 batch 再分两半分别翻译
                mid = len(batch) // 2
                left_title, left_srt = translate_in_batches(
                    entries_to_srt(batch[:mid]),
                    original_title if i == 0 else "",
                    translate_api, translate_key,
                    batch_size=batch_size, source_language=source_language
                )
                right_title, right_srt = translate_in_batches(
                    entries_to_srt(batch[mid:]),
                    "",
                    translate_api, translate_key,
                    batch_size=batch_size, source_language=source_language
                )
                if i == 0:
                    translated_title = left_title or right_title
                # 合并两侧结果到 all_translated（省略具体合并逻辑）
                continue
            raise ValueError(
                f"翻译条数不匹配且已无法再拆分：原 {len(batch)} 条，译 {len(translated_entries)} 条"
            )

        for orig, trans in zip(batch, translated_entries):
            all_translated.append({
                "index": orig["index"],
                "start": orig["start"],           # 保留 ASR 原始时间戳
                "end": orig["end"],               # 保留 ASR 原始时间戳
                "start_seconds": orig["start_seconds"],
                "end_seconds": orig["end_seconds"],
                "text": trans["text"],            # 仅替换翻译文本
                "original_text": orig["text"],    # 保留原文供上下文使用
            })

    return translated_title, entries_to_srt(all_translated)

import time

def call_with_retry(func, validator=None, max_retries=3, base_delay=5):
    """
    带指数退避的重试包装，含响应结构校验。
    
    参数:
        func: 无参 callable，执行实际 API 调用并返回结果
        validator: 可选的校验函数，接收结果返回 bool。不传则跳过结构校验
        max_retries: 最大重试次数
        base_delay: 基础退避延迟（秒）
    """
    for attempt in range(max_retries):
        try:
            result = func()
            # 响应结构校验（防止 HTTP 200 但响应体异常）
            if validator is not None and not validator(result):
                raise ValueError("响应结构校验失败：缺少必要字段")
            return result
        except requests.exceptions.Timeout:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"超时，{delay}秒后重试 ({attempt+1}/{max_retries})")
            time.sleep(delay)
        except requests.exceptions.HTTPError as e:
            status = e.response.status_code
            if status == 429:  # 限流
                retry_after = parse_retry_after(e.response.headers.get("Retry-After", "60"))
                retry_after = min(retry_after, 120)  # 最大等待120秒
                print(f"限流，等待{retry_after}秒")
                time.sleep(retry_after)
            elif status >= 500:  # 服务端错误
                if attempt == max_retries - 1:
                    raise
                time.sleep(base_delay * (2 ** attempt))
            else:  # 4xx 客户端错误，不重试
                raise
        except requests.exceptions.ConnectionError:
            if attempt == max_retries - 1:
                raise
            time.sleep(base_delay * (2 ** attempt))
        except ValueError:
            # 响应结构校验失败，可重试
            if attempt == max_retries - 1:
                raise
            time.sleep(base_delay * (2 ** attempt))


def validate_asr_response(result):
    """校验 MiMo ASR 响应：choices[0].message.content 必须为非空字符串"""
    if not isinstance(result, dict):
        return False
    choices = result.get("choices")
    if not isinstance(choices, list) or len(choices) == 0:
        return False
    content = choices[0].get("message", {}).get("content")
    return isinstance(content, str) and len(content.strip()) > 0


def validate_translate_response(result):
    """校验翻译响应：translated_title 和 translated_subtitle 必须非空"""
    if not isinstance(result, dict):
        return False
    return bool(result.get("translated_title")) and bool(result.get("translated_subtitle"))


# 使用示例：
#   transcript = call_with_retry(
#       lambda: call_mimo_asr(path, api, key, lang),
#       validator=validate_asr_response
#   )
#   result = call_with_retry(
#       lambda: call_translate(srt, title, api, key),
#       validator=validate_translate_response
#   )

def parse_retry_after(value):
    """解析 Retry-After 头（支持数字秒和 HTTP-date 格式）"""
    try:
        return int(value)
    except ValueError:
        from email.utils import parsedate_to_datetime
        from datetime import datetime, timezone
        try:
            dt = parsedate_to_datetime(value)
            return max(int((dt - datetime.now(timezone.utc)).total_seconds()), 1)
        except Exception:
            return 60