From 2e93b84ff3f0fb267e57543de01df7d0f8a71063 Mon Sep 17 00:00:00 2001 From: Entropy-R <144404583+Entropy-R@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:40:06 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=80=9A=E7=94=A8=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E5=A4=9A=E6=A8=A1=E6=80=81=E8=A7=A3=E6=9E=90=E4=B8=8E?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E6=A8=A1=E5=9E=8B=E6=80=BB=E7=BB=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 20 +- CHANGELOG.md | 22 + README.md | 128 ++- SECURITY.md | 5 + docker-compose.yml | 11 + requirements.txt | 1 + tests/test_video_summary.py | 679 ++++++++++++ video_summary.py | 2038 ++++++++++++++++++++++++++++++++--- 8 files changed, 2725 insertions(+), 179 deletions(-) create mode 100644 tests/test_video_summary.py diff --git a/.env.example b/.env.example index cfe374b..864fcba 100644 --- a/.env.example +++ b/.env.example @@ -1,8 +1,26 @@ -# OpenAI-compatible API settings. Required unless you use --no-llm or --export-prompt. +# Final summary defaults to the local Ollama model. +SUMMARY_PROVIDER=ollama +SUMMARY_API_KEY=ollama +SUMMARY_BASE_URL=http://host.docker.internal:11434/v1 +SUMMARY_MODEL=qwen3-vl:8b-instruct-q4_K_M +SUMMARY_TIMEOUT=300 +# Local summary uses the Qwen model limit and Ollama runtime context automatically. +SUMMARY_LOCAL_CHUNK_CHARS=auto +SUMMARY_OLLAMA_CONTEXT_LENGTH=8192 + +# Optional OpenAI-compatible API settings. +# Used only with --summary-provider api or SUMMARY_PROVIDER=api. OPENAI_API_KEY=sk-... OPENAI_BASE_URL=https://api.deepseek.com OPENAI_MODEL=deepseek-v4-flash +# Optional local vision model settings. Used only with --with-vision. +# Run Ollama on the macOS host; Docker reaches it through host.docker.internal. +VISION_API_KEY=ollama +VISION_BASE_URL=http://host.docker.internal:11434/v1 +VISION_MODEL=qwen3-vl:8b-instruct-q4_K_M +VISION_TIMEOUT=300 + # Optional: path to a Netscape cookies.txt file mounted inside the container. # Leave empty to use the built-in default: /app/cookies/cookies.txt if it exists. VIDEO_SUMMARY_COOKIES= diff --git a/CHANGELOG.md b/CHANGELOG.md index 0bd993c..0a6bfa7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # 版本变更说明 +## v1.2（开发中） + +- 正式定义“快速解析”和“多模态解析”两种模式；默认快速解析以语音/字幕为主要信息源，多模态解析将语音/字幕与画面都作为重要证据。 +- `meta.json` 新增稳定的 `analysis_mode` 字段，并调整多模态总结提示词，要求核心观点和分段要点综合两类证据。 +- 新增显式 `--with-vision` 模式；默认仍保持字幕优先、Whisper 兜底的原有流程，不下载或解析视频画面。 +- 视觉模式支持宿主机 Ollama 中的 `Qwen3-VL 8B Q4`，Docker 通过 OpenAI 兼容接口调用。 +- 使用固定间隔候选帧、镜头变化检测、感知哈希去重和最大帧数限制实现自适应关键帧筛选。 +- 新增带时间戳的视觉结果、关键帧和多模态时间轴产物，并在视觉失败时严格终止总结。 +- 在线视频在无字幕时复用已下载的视频提取音频，避免为 Whisper 重复下载媒体。 +- 最终总结默认改用本机 Ollama；通过 `--summary-provider api` 继续保留 OpenAI 兼容 API 路径。 +- B 站字幕不可用时区分“没有字幕”和“字幕需要登录”，默认语言规则补充 `ai-zh`。 +- 新增阶段缓存和视觉断点续跑；相同参数重跑可复用转写、关键帧、视觉结果和最终总结。 +- Whisper 转写与 FFmpeg 关键帧抽取受控并行；优化后关键帧默认缩放到 768 像素宽。 +- 多模态总结会删除重复屏幕文字并限制单帧冗余描述；本地总结按较小分段运行，避免 Ollama 4K 上下文截断长视频前半段。 +- 本地总结改用 Ollama 原生接口显式设置上下文，并根据 Qwen 模型上限与 Ollama 运行参数自动计算均衡分段；上下文超限时自动缩小分段重试。 +- 字幕下载后根据实际文件名记录所选语言和字幕类型，B 站 `ai-zh` 可正确标记为自动字幕。 +- 视觉模式根据视频时长自适应候选密度和实际帧预算，增强主体区域近重复过滤，并将默认批量提高到 4 帧。 +- 视觉请求使用相邻字幕和视频术语校正 OCR，只提取字幕之外的新增证据；多模态上下文进一步移除固定页眉、重复文字和字幕重合内容。 +- `visual_context.json` 与 `meta.json` 增加视觉批次耗时、采样计划和帧来源指标。 +- 视觉批次超出上下文时自动拆分并记住有效批量；多模态材料完整保留字幕，同时按总结模型上下文动态限制视觉补充预算。 +- 修复最终总结失败时错误覆盖已完成视觉状态的问题。 + ## v1.1 本版本围绕“转写质量、总结结构、运行可追踪性和文档可用性”做了增强。 diff --git a/README.md b/README.md index bdc2355..acba876 100644 --- a/README.md +++ b/README.md @@ -6,8 +6,9 @@ 1. 优先用 `yt-dlp` 抓取字幕。 2. 字幕不可用时，下载音频并用 `faster-whisper` 转写。 -3. 最后调用 OpenAI 兼容接口（例如 DeepSeek、OpenAI、OpenRouter 等）生成中文总结。 -4. 如果没有 API 额度，也可以导出 `chatgpt_prompt.md`，复制到 ChatGPT 手动总结。 +3. 默认不解析画面；显式使用 `--with-vision` 时，才会抽取关键帧并调用宿主机上的本地视觉模型。 +4. 默认调用宿主机 Ollama 中的 Qwen3-VL 生成中文总结；也可显式切换到 OpenAI 兼容 API。 +5. 如果没有 API 额度，也可以导出 `chatgpt_prompt.md`，复制到 ChatGPT 手动总结。 适合这些场景： @@ -18,6 +19,17 @@ 推荐使用 Docker 运行，避免污染本地 Python 环境。 +## 解析模式 + +| 模式 | 信息来源 | 适用场景 | 特点 | +| --- | --- | --- | --- | +| 快速解析（默认） | 平台字幕；没有字幕时使用 Whisper 转写语音 | 访谈、播客、口播、知识讲解等主要内容由讲述承载的视频 | 速度较快，但不会识别 PPT、代码、图表和操作画面，可能遗漏仅在画面中出现的信息。 | +| 多模态解析 | 语音/字幕 + 关键帧画面 | 录屏教程、课程、演示、评测以及画面信息较多的视频 | 语音与画面都是重要证据，准确度通常更高，但会增加下载、抽帧和本地模型推理时间。 | + +快速解析并不表示视频画面本身不重要，而是明确采用“主要内容由语音或字幕承载”的处理假设。多模态解析中， +语音/字幕主要提供讲述逻辑、观点和因果关系，画面主要提供屏幕文字、PPT、代码、图表、操作步骤和场景变化； +最终总结会综合两类证据，发生冲突时保守标记不确定性。 + ## 路径说明 README 中会同时出现两类路径： @@ -52,14 +64,15 @@ docker compose build copy .env.example .env ``` -编辑 `.env`，填入 OpenAI 兼容接口配置： +默认总结也使用本机 Ollama。先安装 Ollama 并拉取模型： -```text -OPENAI_API_KEY=sk-... -OPENAI_BASE_URL=https://api.deepseek.com -OPENAI_MODEL=deepseek-v4-flash +```bash +ollama pull qwen3-vl:8b-instruct-q4_K_M ``` +`.env.example` 已将最终总结配置为本地 Ollama。默认流程仍不下载或解析视频画面；只有显式使用 +`--with-vision` 时才会启用关键帧分析。 + ## 常用场景 ### 1. 总结本地视频 @@ -85,7 +98,7 @@ docker compose run --rm video-summary "/app/outputs/local/demo.mp4" 默认行为： - 使用 Whisper 转写，默认语言 `zh` -- 调用 `.env` 里的大模型接口 +- 默认调用本机 Qwen3-VL 总结 - 输出 `transcript.txt` 和 `summary.md` ### 2. 总结 B 站视频 @@ -139,15 +152,51 @@ docker compose run --rm video-summary --summary-from-file /app/outputs/example/t 适合已经有 `transcript.txt`，不想重新下载或转写的情况。 +### 6. 使用本地 Qwen3-VL 进行多模态解析 + +当前方案要求 macOS 14 或更高版本。在 macOS 宿主机安装并启动 Ollama，然后拉取模型： + +```bash +ollama pull qwen3-vl:8b-instruct-q4_K_M +``` + +`.env` 中配置： + +```text +VISION_API_KEY=ollama +VISION_BASE_URL=http://host.docker.internal:11434/v1 +VISION_MODEL=qwen3-vl:8b-instruct-q4_K_M +VISION_TIMEOUT=300 +``` + +运行： + +```powershell +docker compose run --rm video-summary "视频链接或本地视频路径" --with-vision +``` + +多模态模式会根据视频时长自动计算候选帧密度和实际帧预算，并补充镜头变化帧。短视频保留基础覆盖， +长视频约按每 30 秒增长一帧，最终受 `--vision-max-frames` 硬上限约束。画面差异去重会忽略边缘 +页眉、水印和播放器装饰；关键帧默认缩放到 768 像素宽，视觉模型每批分析 4 帧，并结合相邻字幕 +校正 OCR，只提取字幕之外的 +新增画面证据。原始图片不会发送给最终的文本总结接口。 + +如果当前 Ollama 上下文无法一次处理默认批量，程序会自动拆成更小批次，并在后续请求中记住已经 +验证可用的批量。最终多模态材料会完整保留字幕，并根据总结模型的安全上下文限制视觉补充预算； +短中视频会尽量使用单次总结，长视频仍采用均衡分段。 + +为保证旧参数语义不变，`--with-vision` 暂不能与 `--no-llm`、`--export-prompt`、`--summary-from-file` 同时使用。 + ## B 站 Cookies -B 站经常会对未登录或容器网络请求返回： +B 站经常会对未登录或容器网络请求返回 412，或者只允许登录用户获取字幕： ```text HTTP Error 412: Precondition Failed ``` -这通常不是程序错误，而是缺少有效登录态。建议导出 cookies。 +这通常不是程序错误，而是缺少有效登录态。没有 cookies 时，即使视频网页上能看到 AI 字幕， +`yt-dlp` 也可能只能获得弹幕轨道，程序将提示登录要求并回退到 Whisper。建议导出 cookies。 默认 Docker 用法会自动尝试读取容器内的 `/app/cookies/cookies.txt`，所以宿主机项目目录下的文件名必须是： @@ -220,6 +269,10 @@ outputs/ transcript.txt summary.md chatgpt_prompt.md + visual_context.json + visual_context.md + multimodal_context.txt + frames/run-*/ *.srt ``` @@ -231,6 +284,9 @@ outputs/ | `transcript.txt` | 最终用于总结的文本。可能来自字幕，也可能来自 Whisper 转写；可识别时间时会保留为 `[00:01:23] 文本`。 | | `summary.md` | 大模型生成的中文总结。只有调用 LLM 成功时生成，默认会尽量按转写稿中的时间节点组织分段要点。 | | `chatgpt_prompt.md` | 可复制到 ChatGPT 的提示词。使用 `--export-prompt` 或 LLM 失败兜底时生成。 | +| `visual_context.json` / `visual_context.md` | 视觉模式生成的带时间戳画面解析结果。 | +| `multimodal_context.txt` | 按时间轴合并语音/字幕与画面描述的最终总结输入。 | +| `frames/run-*/` | 视觉模式筛选出的关键帧；调试阶段默认保留。 | | `*.srt` | 下载到的字幕文件，如果视频有字幕才会出现。 | | `*.mp3` / `*.m4a` | 下载的音频文件。默认转写后删除，使用 `--keep-audio` 时保留。 | @@ -244,7 +300,9 @@ outputs/ "uploader": "up-name", "duration": 120, "subtitle_langs": [], - "automatic_caption_langs": [], + "automatic_caption_langs": ["ai-zh"], + "selected_subtitle_lang": "ai-zh", + "selected_subtitle_type": "automatic", "whisper_context_terms": ["LLaMA-Factory", "AI"], "transcript_source": "whisper", "warnings": [] @@ -270,7 +328,7 @@ outputs/ | `--output` | `outputs` | 输出目录。 | | `--model-size` | `small` | Whisper 模型大小，影响转写速度和准确率。 | | `--language` | `zh` | Whisper 识别语言。中文视频默认不用配置；英文视频建议传 `en`；中英文不确定或多语言内容可传 `auto`。 | -| `--sub-langs` | `zh-Hans,zh-CN,zh,en` | 字幕语言优先级。 | +| `--sub-langs` | `zh.*,ai-zh,en.*` | 字幕语言匹配规则，覆盖常见中文、B 站 AI 中文字幕和英文轨道。 | | `--cookies` | 自动尝试 `/app/cookies/cookies.txt` | cookies 文件路径。默认文件不存在时会忽略。 | | `--cookies-from-browser` | 空 | 从浏览器读取 cookies，例如 `edge`、`chrome`、`firefox`。Docker 中通常需要额外挂载浏览器 profile。 | | `--no-llm` | `false` | 只生成 `transcript.txt`，不调用大模型。 | @@ -282,6 +340,14 @@ outputs/ | `--prompt` | 空 | 直接传入自定义总结 prompt 模板文本。 | | `--prompt-file` | 空 | 从文件读取自定义总结 prompt 模板。 | | `--max-chars` | `12000` | LLM 或 prompt 分段最大字符数。 | +| `--summary-provider` | `ollama` | 最终总结提供方；`api` 使用 `OPENAI_*` 配置。 | +| `--no-resume` | `false` | 忽略阶段缓存，强制重新处理。 | +| `--with-vision` | `false` | 启用多模态解析（语音/字幕 + 关键帧）；默认使用快速解析，仅处理语音/字幕。 | +| `--vision-scan-interval` | `5` | 候选帧扫描间隔秒数。 | +| `--vision-max-gap` | `45` | 自适应采样允许的相似画面最长保留间隔上限秒数。 | +| `--vision-max-frames` | `60` | 单个视频最多解析的关键帧数。 | +| `--vision-batch-size` | `4` | 每次视觉模型请求包含的图片数。 | +| `--vision-frame-width` | `768` | 送入视觉模型前的关键帧宽度。 | ## Whisper 模型大小 @@ -316,7 +382,7 @@ distil-small.en, distil-medium.en, distil-large-v2, distil-large-v3 这个过程默认启用，不需要额外参数。关键词只来自视频自身元数据，不使用固定内置词表；如果没有提取到有效关键词，就不会传上下文提示。实际使用的关键词会记录到 `meta.json` 的 `whisper_context_terms` 字段。视频有字幕时会优先使用字幕，这个 Whisper 上下文不会生效。 -## 运行元数据和旧产物 +## 运行元数据、缓存和断点续跑 每次运行会在 `meta.json` 中记录本次参数和阶段耗时： @@ -340,7 +406,9 @@ distil-small.en, distil-medium.en, distil-large-v2, distil-large-v3 } ``` -如果同一个视频输出目录已经存在，程序会在本次运行开始前清理旧的 `transcript.txt`、`summary.md`、`chatgpt_prompt.md`，避免不同运行产生的旧结果混在一起。其他文件（例如字幕文件、手动放入的资料、保留的音频）不会被自动删除。 +程序默认在 `pipeline_cache.json` 中记录转写、关键帧、视觉解析和最终总结的签名。同一视频以相同参数重跑时， +会复用已完成阶段；视觉处理中断后也会从缺失帧继续。修改模型、抽帧参数、总结模型或输入文件后，相应阶段会自动失效。 +需要全部重新处理时使用 `--no-resume`。调试用的 `frames/run-*/` 不会自动删除。 ## Prompt 模板 @@ -381,17 +449,36 @@ docker compose run --rm video-summary "/app/outputs/local/demo.mp4" --prompt-fil ## 配置说明 -`.env.example` 默认使用 DeepSeek 兼容接口示例： +`.env.example` 默认使用本机 Ollama 完成视觉解析和最终总结： ```text +SUMMARY_PROVIDER=ollama +SUMMARY_API_KEY=ollama +SUMMARY_BASE_URL=http://host.docker.internal:11434/v1 +SUMMARY_MODEL=qwen3-vl:8b-instruct-q4_K_M +SUMMARY_TIMEOUT=300 +SUMMARY_LOCAL_CHUNK_CHARS=auto +SUMMARY_OLLAMA_CONTEXT_LENGTH=8192 OPENAI_API_KEY=sk-... OPENAI_BASE_URL=https://api.deepseek.com OPENAI_MODEL=deepseek-v4-flash +VISION_API_KEY=ollama +VISION_BASE_URL=http://host.docker.internal:11434/v1 +VISION_MODEL=qwen3-vl:8b-instruct-q4_K_M +VISION_TIMEOUT=300 VIDEO_SUMMARY_COOKIES= VIDEO_SUMMARY_COOKIES_FROM_BROWSER= ``` -如果使用 OpenAI，把 `OPENAI_BASE_URL` 和 `OPENAI_MODEL` 改成对应值即可。 +需要调用云端 OpenAI 兼容接口时，设置 `SUMMARY_PROVIDER=api`，或者传入 +`--summary-provider api`，再配置 `OPENAI_API_KEY`、`OPENAI_BASE_URL` 和 `OPENAI_MODEL`。 +程序不会在本地总结失败后自动调用云端接口，避免意外费用和数据外发。 +本地总结默认读取 Qwen 模型向 Ollama 声明的最大上下文，并在每次原生 Ollama 请求中显式设置 +`SUMMARY_OLLAMA_CONTEXT_LENGTH`（默认 8192）。程序会为提示词和输出预留空间，自动计算不超过 +当前上下文安全容量的分段；文本能安全放入一次请求时直接总结，超过时才均衡分段。若 Ollama返回 +上下文超限，程序会缩小分段后重试。需要固定分段时可把 `SUMMARY_LOCAL_CHUNK_CHARS` 设置为正整数。 +长分段的中间摘要会保留更充足的输出预算，避免为了减少调用次数而丢失分段后半部分。提高 Ollama +上下文会增加内存占用。 注意：`docker compose config` 会展开 `.env` 中的真实 API key，不要把它的输出截图或发到公开场合。 @@ -415,10 +502,17 @@ docker compose run --rm video-summary "/app/outputs/local/sample.mp4" --model-si docker compose run --rm video-summary "/app/outputs/local/sample.mp4" --model-size tiny --no-llm --json ``` +视觉连通和短视频验证： + +```powershell +docker compose run --rm video-summary "/app/outputs/local/sample.mp4" --model-size tiny --with-vision --vision-max-frames 6 +``` + ## 注意事项 - 首次 Whisper 转写会下载模型，模型缓存在 Docker volume `video_summary_cache`。 -- ChatGPT Plus 不等于 OpenAI API 额度；API 需要单独配置 billing。 +- Qwen3-VL 运行在宿主机 Ollama 中，不占用 Docker volume；16 GB 内存机器建议保持单任务运行。 +- 使用 `--summary-provider api` 时，ChatGPT Plus 不等于 OpenAI API 额度；API 需要单独配置 billing。 - B 站链接可能需要 cookies、代理或换网络。 - `.env`、`cookies/`、`outputs/` 已加入忽略列表，避免误提交敏感数据或产物。 diff --git a/SECURITY.md b/SECURITY.md index 6d9d57a..03de5e9 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -22,3 +22,8 @@ If an API key appears in logs, screenshots, Git history, or shared chat: ## Responsible use Only process videos when you have the right to access and summarize the content. + +When `--with-vision` is enabled, extracted keyframes are sent only to the configured `VISION_BASE_URL`. +With the documented defaults, both vision analysis and final summary use local Ollama. If `SUMMARY_PROVIDER=api` or +`--summary-provider api` is selected, generated visual descriptions and transcript text are sent to `OPENAI_BASE_URL`; +the original keyframe images are not sent to the summary provider. diff --git a/docker-compose.yml b/docker-compose.yml index 35d1b94..f51632c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,9 +4,20 @@ services: image: video-summary:local working_dir: /app environment: + SUMMARY_PROVIDER: ${SUMMARY_PROVIDER:-ollama} + SUMMARY_API_KEY: ${SUMMARY_API_KEY:-ollama} + SUMMARY_BASE_URL: ${SUMMARY_BASE_URL:-http://host.docker.internal:11434/v1} + SUMMARY_MODEL: ${SUMMARY_MODEL:-qwen3-vl:8b-instruct-q4_K_M} + SUMMARY_TIMEOUT: ${SUMMARY_TIMEOUT:-300} + SUMMARY_LOCAL_CHUNK_CHARS: ${SUMMARY_LOCAL_CHUNK_CHARS:-auto} + SUMMARY_OLLAMA_CONTEXT_LENGTH: ${SUMMARY_OLLAMA_CONTEXT_LENGTH:-8192} OPENAI_API_KEY: ${OPENAI_API_KEY:-} OPENAI_BASE_URL: ${OPENAI_BASE_URL:-https://api.openai.com/v1} OPENAI_MODEL: ${OPENAI_MODEL:-gpt-4.1-mini} + VISION_API_KEY: ${VISION_API_KEY:-ollama} + VISION_BASE_URL: ${VISION_BASE_URL:-http://host.docker.internal:11434/v1} + VISION_MODEL: ${VISION_MODEL:-qwen3-vl:8b-instruct-q4_K_M} + VISION_TIMEOUT: ${VISION_TIMEOUT:-300} VIDEO_SUMMARY_COOKIES: ${VIDEO_SUMMARY_COOKIES:-} VIDEO_SUMMARY_COOKIES_FROM_BROWSER: ${VIDEO_SUMMARY_COOKIES_FROM_BROWSER:-} XDG_CACHE_HOME: /cache diff --git a/requirements.txt b/requirements.txt index f338089..d112405 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ yt-dlp>=2025.5.22 faster-whisper>=1.1.1 openai>=1.82.0 +Pillow>=11.2.1 python-dotenv>=1.1.0 diff --git a/tests/test_video_summary.py b/tests/test_video_summary.py new file mode 100644 index 0000000..69f60e1 --- /dev/null +++ b/tests/test_video_summary.py @@ -0,0 +1,679 @@ +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest.mock import patch + + +if importlib.util.find_spec("dotenv") is None: + dotenv = types.ModuleType("dotenv") + dotenv.load_dotenv = lambda: None + sys.modules["dotenv"] = dotenv + +import video_summary + + +class ArgumentTests(unittest.TestCase): + def test_default_path_keeps_vision_disabled(self): + args = video_summary.parse_args(["https://example.com/video"]) + + video_summary.validate_args(args) + + self.assertFalse(args.with_vision) + self.assertEqual(args.vision_scan_interval, 5) + self.assertEqual(args.vision_max_gap, 45) + self.assertEqual(args.vision_max_frames, 60) + self.assertEqual(args.vision_batch_size, 4) + self.assertEqual(args.vision_frame_width, 768) + self.assertEqual(args.summary_provider, "ollama") + self.assertEqual(args.sub_langs, "zh.*,ai-zh,en.*") + _processing, options = video_summary.build_processing_info(args, "zh") + self.assertEqual(options["analysis_mode"], "quick") + + def test_with_vision_uses_multimodal_analysis_mode(self): + args = video_summary.parse_args(["https://example.com/video", "--with-vision"]) + + _processing, options = video_summary.build_processing_info(args, "zh") + + self.assertEqual(options["analysis_mode"], "multimodal") + self.assertIn("两类信息都是理解视频的重要证据", video_summary.MULTIMODAL_PROMPT_TEMPLATE) + self.assertIn("## 关键画面信息", video_summary.MULTIMODAL_PROMPT_TEMPLATE) + + def test_meta_exposes_stable_analysis_mode(self): + meta = video_summary.VideoMeta("video", "id", "https://example.com/video") + + with tempfile.TemporaryDirectory() as temp_name: + output_dir = Path(temp_name) + video_summary.write_meta( + meta, + output_dir, + options={"analysis_mode": "multimodal", "with_vision": True}, + ) + payload = json.loads((output_dir / "meta.json").read_text(encoding="utf-8")) + + self.assertEqual(payload["analysis_mode"], "multimodal") + self.assertTrue(payload["options"]["with_vision"]) + + def test_vision_rejects_text_only_modes(self): + for flag in ("--no-llm", "--export-prompt", "--summary-from-file"): + argv = ["https://example.com/video", "--with-vision", flag] + if flag == "--summary-from-file": + argv.append("transcript.txt") + with self.subTest(flag=flag), self.assertRaises(video_summary.AppError): + video_summary.validate_args(video_summary.parse_args(argv)) + + def test_summary_provider_defaults_to_local_ollama(self): + with patch.dict(os.environ, {}, clear=True): + config = video_summary.resolve_summary_config("ollama") + + self.assertEqual(config.provider, "ollama") + self.assertEqual(config.model, "qwen3-vl:8b-instruct-q4_K_M") + self.assertEqual(config.base_url, "http://host.docker.internal:11434/v1") + self.assertIsNone(config.chunk_chars) + self.assertEqual(config.context_length, 8192) + + def test_api_summary_requires_key(self): + with patch.dict(os.environ, {}, clear=True), self.assertRaises(video_summary.AppError) as context: + video_summary.resolve_summary_config("api") + + self.assertEqual(context.exception.code, "missing_openai_api_key") + + +class SubtitleTests(unittest.TestCase): + def test_bilibili_ai_subtitle_language_and_type_are_detected_from_file(self): + meta = video_summary.VideoMeta("video", "BV1", "https://www.bilibili.com/video/BV1") + + source, warnings = video_summary.detect_subtitle_source(Path("video.ai-zh.srt"), meta) + + self.assertEqual(source, "auto_subtitle") + self.assertEqual(meta.selected_subtitle_lang, "ai-zh") + self.assertEqual(meta.selected_subtitle_type, "automatic") + self.assertIn("ai-zh", meta.automatic_caption_langs) + self.assertTrue(any("自动字幕（ai-zh）" in warning for warning in warnings)) + + def test_bilibili_login_requirement_is_reported_before_whisper_fallback(self): + result = subprocess.CompletedProcess( + args=["yt-dlp"], + returncode=0, + stdout="", + stderr="WARNING: [BiliBili] Subtitles are only available when logged in.", + ) + meta = video_summary.VideoMeta("video", "BV1", "https://www.bilibili.com/video/BV1") + warnings = [] + + with tempfile.TemporaryDirectory() as temp_name, patch.object( + video_summary, "require_tool" + ), patch.object(video_summary, "run_command", return_value=result): + subtitle = video_summary.download_subtitle( + meta.webpage_url, + Path(temp_name), + None, + "zh.*,ai-zh,en.*", + meta, + warnings, + ) + + self.assertIsNone(subtitle) + self.assertTrue(any("需要登录态" in warning for warning in warnings)) + + def test_cached_automatic_subtitle_restores_quality_warning(self): + meta = video_summary.VideoMeta("video", "BV1", "https://www.bilibili.com/video/BV1") + warnings = [] + + video_summary.restore_cached_subtitle_metadata( + { + "selected_subtitle_lang": "ai-zh", + "selected_subtitle_type": "automatic", + }, + meta, + warnings, + ) + video_summary.restore_cached_subtitle_metadata( + { + "selected_subtitle_lang": "ai-zh", + "selected_subtitle_type": "automatic", + }, + meta, + warnings, + ) + + self.assertEqual(meta.selected_subtitle_lang, "ai-zh") + self.assertEqual(meta.selected_subtitle_type, "automatic") + self.assertEqual(meta.automatic_caption_langs, ["ai-zh"]) + self.assertEqual( + warnings, + ["使用的是平台自动字幕（ai-zh），内容可能存在识别错误。"], + ) + + +class SummaryChunkTests(unittest.TestCase): + def test_ollama_context_determines_default_chunk_size(self): + config = video_summary.SummaryConfig( + provider="ollama", + api_key="ollama", + base_url="http://localhost:11434/v1", + model="qwen3-vl:8b", + timeout=300, + context_length=8192, + ) + + self.assertEqual(video_summary.determine_summary_chunk_chars(config, 12000), 6392) + + def test_balanced_split_uses_two_similar_chunks(self): + text = "\n".join(f"[00:00:{index:02d}] " + "内容" * 90 for index in range(50)) + + chunks = video_summary.split_text_balanced(text, 5000) + + self.assertEqual(len(chunks), 2) + self.assertLessEqual(max(map(len, chunks)), 5000) + self.assertLess(abs(len(chunks[0]) - len(chunks[1])), 500) + + def test_balanced_split_does_not_leave_tiny_subtitle_tail(self): + text = "\n".join(f"[00:00:{index:02d}] " + "内容" * 40 for index in range(58)) + + chunks = video_summary.split_text_balanced(text, 5000) + + self.assertEqual(len(chunks), 2) + self.assertLess(abs(len(chunks[0]) - len(chunks[1])), 300) + + def test_merge_material_requires_full_coverage_of_every_chunk(self): + material = video_summary.build_summary_merge_material(["第一段", "第二段"]) + + self.assertIn("第 1/2 段摘要", material) + self.assertIn("第 2/2 段摘要", material) + self.assertIn("开头、中间和结尾", material) + self.assertIn("不得声称“原文截断”", material) + + +class FrameSelectionTests(unittest.TestCase): + def test_sampling_plan_adapts_to_video_duration(self): + short = video_summary.build_vision_sampling_plan(60, 5, 45, 60) + medium = video_summary.build_vision_sampling_plan(510, 5, 45, 60) + long = video_summary.build_vision_sampling_plan(3600, 5, 45, 60) + + self.assertEqual(short.max_frames, 8) + self.assertEqual(medium.max_frames, 18) + self.assertEqual(long.max_frames, 60) + self.assertEqual(medium.scan_interval, 7.083) + self.assertEqual(long.scan_interval, 15) + self.assertEqual(long.max_gap, 45) + + def test_sampling_plan_uses_safe_budget_when_duration_is_unknown(self): + plan = video_summary.build_vision_sampling_plan(None, 5, 45, 60) + + self.assertEqual(plan.max_frames, 16) + self.assertEqual(plan.scan_interval, 5) + + def test_filters_similar_frames_but_keeps_changes_scenes_and_max_gap(self): + candidates = [ + video_summary.FrameCandidate(Path("0.jpg"), 0, "periodic"), + video_summary.FrameCandidate(Path("5.jpg"), 5, "periodic"), + video_summary.FrameCandidate(Path("10.jpg"), 10, "periodic"), + video_summary.FrameCandidate(Path("12.jpg"), 12, "scene"), + video_summary.FrameCandidate(Path("35.jpg"), 35, "periodic"), + ] + hashes = { + Path("0.jpg"): 0, + Path("5.jpg"): 0, + Path("10.jpg"): (1 << 64) - 1, + Path("12.jpg"): (1 << 64) - 1, + Path("35.jpg"): (1 << 64) - 1, + } + + with patch.object(video_summary, "image_difference_hash", side_effect=lambda path: hashes[path]): + selected = video_summary.filter_frame_candidates(candidates, max_gap=20, max_frames=60) + + self.assertEqual([frame.timestamp for frame in selected], [0, 10, 12, 35]) + + def test_selection_respects_maximum(self): + candidates = [ + video_summary.FrameCandidate(Path(f"{index}.jpg"), index * 5, "periodic") + for index in range(20) + ] + + with patch.object(video_summary, "image_difference_hash", side_effect=range(20)): + selected = video_summary.filter_frame_candidates( + candidates, + max_gap=5, + max_frames=6, + hash_threshold=0, + ) + + self.assertEqual(len(selected), 6) + self.assertEqual(selected[0].timestamp, 0) + self.assertEqual(selected[-1].timestamp, 95) + + def test_keyframe_cache_requires_all_files(self): + with tempfile.TemporaryDirectory() as temp_name: + output_dir = Path(temp_name) + frame_path = output_dir / "frames/one.jpg" + frame_path.parent.mkdir() + frame_path.write_bytes(b"frame") + frames = [video_summary.FrameCandidate(frame_path, 5, "scene", 123)] + cache = {"schema_version": 1, "stages": {}} + video_summary.update_cache_stage( + cache, + "keyframes", + "signature", + frames=video_summary.serialize_keyframes(frames, output_dir), + ) + + restored = video_summary.load_cached_keyframes(cache, "signature", output_dir) + frame_path.unlink() + missing = video_summary.load_cached_keyframes(cache, "signature", output_dir) + + self.assertEqual(restored[0].timestamp, 5) + self.assertIsNone(missing) + + +class VisionOutputTests(unittest.TestCase): + def test_nearby_transcript_context_uses_timestamp_window(self): + transcript = "[00:00:05] 开头\n[00:00:20] 当前内容\n[00:01:00] 远处内容\n" + + context = video_summary.nearby_transcript_context(transcript, 18, window_seconds=5) + + self.assertEqual(context, "当前内容") + + def test_normalizes_visual_response_with_timestamps(self): + output_dir = Path("/tmp/output") + frames = [ + video_summary.FrameCandidate(output_dir / "frames/one.jpg", 5, "periodic"), + video_summary.FrameCandidate(output_dir / "frames/two.jpg", 12, "scene"), + ] + payload = { + "frames": [ + { + "index": 1, + "description": "显示项目首页", + "visible_text": ["开始", "设置"], + "importance": "high", + "uncertainty": "", + }, + { + "index": 2, + "description": "打开设置页面", + "visible_text": "模型配置", + "importance": "medium", + "uncertainty": "小字无法辨认", + }, + ] + } + + results = video_summary.normalize_visual_batch(payload, frames, output_dir) + + self.assertEqual(results[0]["timestamp"], "00:00:05") + self.assertEqual(results[0]["visible_text"], "开始；设置") + self.assertEqual(results[1]["frame_source"], "scene") + + def test_multimodal_context_is_sorted_by_time(self): + transcript = "[00:00:10] 第二句话\n[00:00:02] 第一句话\n" + visuals = [ + { + "timestamp_seconds": 5, + "timestamp": "00:00:05", + "description": "显示流程图", + "visible_text": "输入→输出", + "uncertainty": "", + } + ] + + context = video_summary.build_multimodal_context(transcript, visuals) + + self.assertLess(context.index("00:00:02"), context.index("00:00:05")) + self.assertLess(context.index("00:00:05"), context.index("00:00:10")) + self.assertIn("屏幕文字：输入→输出", context) + + def test_multimodal_context_removes_exact_duplicate_visuals(self): + visuals = [ + { + "timestamp_seconds": 5, + "timestamp": "00:00:05", + "description": "显示同一页面", + "visible_text": "设置", + "uncertainty": "", + }, + { + "timestamp_seconds": 10, + "timestamp": "00:00:10", + "description": "显示同一页面", + "visible_text": "设置", + "uncertainty": "", + }, + ] + + context = video_summary.build_multimodal_context("", visuals) + + self.assertIn("00:00:05", context) + self.assertNotIn("00:00:10", context) + + def test_multimodal_context_keeps_only_new_lines_from_incremental_slide(self): + visuals = [ + { + "timestamp_seconds": 5, + "timestamp": "00:00:05", + "description": "显示第一步", + "visible_text": "标题\n选项 A", + "uncertainty": "", + }, + { + "timestamp_seconds": 10, + "timestamp": "00:00:10", + "description": "增加第二步", + "visible_text": "标题\n选项 A\n选项 B", + "uncertainty": "", + }, + ] + + context = video_summary.build_multimodal_context("", visuals) + + self.assertEqual(context.count("选项 A"), 1) + self.assertEqual(context.count("选项 B"), 1) + + def test_compaction_splits_semicolon_text_and_removes_subtitle_overlap(self): + visuals = [ + { + "timestamp_seconds": 5, + "timestamp": "00:00:05", + "description": "显示步骤", + "visible_text": "固定页眉；开始执行；参数 A", + "importance": "medium", + "uncertainty": "", + }, + { + "timestamp_seconds": 10, + "timestamp": "00:00:10", + "description": "显示下一步", + "visible_text": "固定页眉；开始执行；参数 B", + "importance": "medium", + "uncertainty": "", + }, + { + "timestamp_seconds": 15, + "timestamp": "00:00:15", + "description": "显示完成", + "visible_text": "固定页眉；开始执行；参数 C", + "importance": "medium", + "uncertainty": "", + }, + ] + + compacted = video_summary.compact_visual_material( + visuals, + "[00:00:05] 现在开始执行\n", + ) + combined = "\n".join(item["visible_text"] for item in compacted) + + self.assertNotIn("固定页眉", combined) + self.assertNotIn("开始执行", combined) + self.assertIn("参数 A", combined) + + def test_multimodal_budget_preserves_transcript_and_limits_visuals(self): + transcript = "\n".join( + f"[00:00:{index:02d}] 第 {index} 条字幕内容" + for index in range(30) + ) + visuals = [ + { + "timestamp_seconds": index, + "timestamp": video_summary.format_timestamp(index), + "description": "画面新增信息" * 20, + "visible_text": f"参数 {index} " * 20, + "importance": "high", + "uncertainty": "", + } + for index in range(20) + ] + + context = video_summary.build_multimodal_context( + transcript, + visuals, + max_chars=len(transcript) + 500, + ) + + self.assertIn("第 29 条字幕内容", context) + self.assertLessEqual(len(context), len(transcript) + 500) + self.assertLess(context.count("] 画面："), len(visuals)) + + def test_multimodal_budget_keeps_exact_command_evidence(self): + visuals = [ + { + "timestamp_seconds": index, + "timestamp": video_summary.format_timestamp(index), + "description": f"第 {index} 个画面" + "说明" * 20, + "visible_text": ( + "$ npx skills add owner/repo --skill example" + if index == 8 + else f"普通文字 {index}" * 10 + ), + "importance": "high", + "uncertainty": "", + } + for index in range(10) + ] + + context = video_summary.build_multimodal_context("", visuals, max_chars=650) + + self.assertIn("npx skills add owner/repo --skill example", context) + + def test_parses_json_inside_markdown_fence(self): + payload = video_summary.parse_json_object('```json\n{"frames": []}\n```') + + self.assertEqual(payload, {"frames": []}) + + def test_analyzes_frames_and_writes_completed_artifact(self): + response = types.SimpleNamespace( + choices=[ + types.SimpleNamespace( + message=types.SimpleNamespace( + content=( + '{"frames":[{"index":1,"description":"首页",' + '"visible_text":"开始","importance":"high","uncertainty":""}]}' + ) + ) + ) + ] + ) + completions = types.SimpleNamespace(create=lambda **_kwargs: response) + client = types.SimpleNamespace(chat=types.SimpleNamespace(completions=completions)) + config = video_summary.VisionConfig("ollama", "http://localhost/v1", "qwen", 30) + + with tempfile.TemporaryDirectory() as temp_name: + output_dir = Path(temp_name) + frame_path = output_dir / "frame.jpg" + frame_path.write_bytes(b"image") + frames = [video_summary.FrameCandidate(frame_path, 0, "periodic")] + + results = video_summary.analyze_keyframes(client, config, frames, output_dir, 1) + artifact = video_summary.parse_json_object((output_dir / "visual_context.json").read_text()) + + self.assertEqual(results[0]["description"], "首页") + self.assertEqual(artifact["status"], "completed") + + def test_visual_failure_is_strict_and_keeps_failed_artifact(self): + def fail(**_kwargs): + raise RuntimeError("model stopped") + + completions = types.SimpleNamespace(create=fail) + client = types.SimpleNamespace(chat=types.SimpleNamespace(completions=completions)) + config = video_summary.VisionConfig("ollama", "http://localhost/v1", "qwen", 30) + + with tempfile.TemporaryDirectory() as temp_name: + output_dir = Path(temp_name) + frame_path = output_dir / "frame.jpg" + frame_path.write_bytes(b"image") + frames = [video_summary.FrameCandidate(frame_path, 0, "periodic")] + + with patch.object(video_summary.time, "sleep"), self.assertRaises(video_summary.AppError): + video_summary.analyze_keyframes(client, config, frames, output_dir, 1) + artifact = video_summary.parse_json_object((output_dir / "visual_context.json").read_text()) + + self.assertEqual(artifact["status"], "failed") + self.assertIn("model stopped", artifact["error"]) + + def test_visual_resume_only_requests_missing_frames(self): + calls = [] + + def create(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + choices=[ + types.SimpleNamespace( + message=types.SimpleNamespace( + content=( + '{"frames":[{"index":1,"description":"第二页",' + '"visible_text":"","importance":"medium","uncertainty":""}]}' + ) + ) + ) + ] + ) + + client = types.SimpleNamespace( + chat=types.SimpleNamespace(completions=types.SimpleNamespace(create=create)) + ) + config = video_summary.VisionConfig("ollama", "http://localhost/v1", "qwen", 30) + with tempfile.TemporaryDirectory() as temp_name: + output_dir = Path(temp_name) + first_path = output_dir / "first.jpg" + second_path = output_dir / "second.jpg" + first_path.write_bytes(b"first") + second_path.write_bytes(b"second") + frames = [ + video_summary.FrameCandidate(first_path, 0, "periodic"), + video_summary.FrameCandidate(second_path, 5, "scene"), + ] + existing = [ + { + "timestamp_seconds": 0, + "timestamp": "00:00:00", + "image": "first.jpg", + "frame_source": "periodic", + "description": "第一页", + "visible_text": "", + "importance": "medium", + "uncertainty": "", + } + ] + + results = video_summary.analyze_keyframes( + client, + config, + frames, + output_dir, + 1, + existing_results=existing, + ) + + self.assertEqual(len(calls), 1) + self.assertEqual([item["description"] for item in results], ["第一页", "第二页"]) + + def test_visual_batch_falls_back_to_smaller_requests(self): + calls = [] + + def create(**kwargs): + image_count = sum( + item.get("type") == "image_url" + for item in kwargs["messages"][1]["content"] + ) + calls.append(image_count) + if image_count > 1: + raise RuntimeError("context length exceeded") + return types.SimpleNamespace( + choices=[ + types.SimpleNamespace( + message=types.SimpleNamespace( + content=( + '{"frames":[{"index":1,"description":"单帧结果",' + '"visible_text":"","importance":"medium","uncertainty":""}]}' + ) + ) + ) + ] + ) + + client = types.SimpleNamespace( + chat=types.SimpleNamespace(completions=types.SimpleNamespace(create=create)) + ) + config = video_summary.VisionConfig("ollama", "http://localhost/v1", "qwen", 30) + with tempfile.TemporaryDirectory() as temp_name, patch.object(video_summary.time, "sleep"): + output_dir = Path(temp_name) + frames = [] + for index in range(2): + path = output_dir / f"{index}.jpg" + path.write_bytes(b"image") + frames.append(video_summary.FrameCandidate(path, index * 5, "periodic")) + + results = video_summary.analyze_keyframes( + client, + config, + frames, + output_dir, + 2, + ) + artifact = video_summary.parse_json_object( + (output_dir / "visual_context.json").read_text() + ) + + self.assertEqual(len(results), 2) + self.assertEqual(calls, [2, 2, 1, 1]) + self.assertEqual(artifact["metrics"]["fallback_splits"], 1) + + def test_visual_batch_remembers_successful_smaller_size(self): + calls = [] + + def create(**kwargs): + image_count = sum( + item.get("type") == "image_url" + for item in kwargs["messages"][1]["content"] + ) + calls.append(image_count) + if image_count > 2: + raise RuntimeError("context length exceeded") + frames = ",".join( + f'{{"index":{index},"description":"结果 {index}",' + '"visible_text":"","importance":"medium","uncertainty":""}' + for index in range(1, image_count + 1) + ) + return types.SimpleNamespace( + choices=[ + types.SimpleNamespace( + message=types.SimpleNamespace(content=f'{{"frames":[{frames}]}}') + ) + ] + ) + + client = types.SimpleNamespace( + chat=types.SimpleNamespace(completions=types.SimpleNamespace(create=create)) + ) + config = video_summary.VisionConfig("ollama", "http://localhost/v1", "qwen", 30) + with tempfile.TemporaryDirectory() as temp_name, patch.object(video_summary.time, "sleep"): + output_dir = Path(temp_name) + frame_items = [] + for index in range(6): + path = output_dir / f"{index}.jpg" + path.write_bytes(b"image") + frame_items.append(video_summary.FrameCandidate(path, index * 5, "periodic")) + + results = video_summary.analyze_keyframes( + client, + config, + frame_items, + output_dir, + 4, + ) + artifact = video_summary.parse_json_object( + (output_dir / "visual_context.json").read_text() + ) + + self.assertEqual(len(results), 6) + self.assertEqual(calls, [4, 4, 2, 2, 2]) + self.assertEqual(artifact["metrics"]["effective_batch_size"], 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/video_summary.py b/video_summary.py index 908b8c1..3966b71 100644 --- a/video_summary.py +++ b/video_summary.py @@ -1,24 +1,43 @@ from __future__ import annotations import argparse +import base64 +import difflib +import hashlib import json +import math import os import re import shutil import subprocess import sys +import tempfile import time +from collections import Counter +from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Iterable +from urllib import error as urllib_error +from urllib import request as urllib_request from dotenv import load_dotenv SUBTITLE_EXTS = (".srt", ".vtt", ".ass", ".json3", ".srv1", ".srv2", ".srv3", ".ttml") AUDIO_EXTS = (".mp3", ".m4a", ".opus", ".wav", ".webm") +VIDEO_EXTS = (".mp4", ".mkv", ".mov", ".webm", ".flv", ".avi", ".m4v") DEFAULT_COOKIES_PATH = "/app/cookies/cookies.txt" -GENERATED_OUTPUT_FILES = ("transcript.txt", "summary.md", "chatgpt_prompt.md") +GENERATED_OUTPUT_FILES = ( + "transcript.txt", + "summary.md", + "chatgpt_prompt.md", + "visual_context.json", + "visual_context.md", + "multimodal_context.txt", +) +CACHE_FILE_NAME = "pipeline_cache.json" +CACHE_SCHEMA_VERSION = 1 AVAILABLE_MODEL_SIZES = ( "tiny", "tiny.en", @@ -74,13 +93,66 @@ 用一两句话概括这一段内容。 ## 本段时间节点与要点 -按转写文本中的时间节点组织，保留原文已有时间节点；每个节点写出稍详细的事实、观点、步骤或例子。不要自行编造时间。 +先扫描完整分段，再用 6–10 条覆盖开头、中间和结尾。合并相邻时间节点，保留原文已有时间； +每条简洁写出事实、观点、步骤或例子，不要把输出预算都用在前半段，也不要自行编造时间。 ## 本段重要细节 -列出后续合并总结时不能丢失的关键细节、术语、工具名、参数或限制。 +简洁列出后续合并总结时不能丢失的关键细节、术语、工具名、参数或限制。 这是第 {chunk_index}/{chunk_count} 段转写文本： +{transcript} +""" +MULTIMODAL_PROMPT_TEMPLATE = """你是严谨的视频内容总结助手。请只基于给定的视频材料，用中文输出一份视频总结，不要编造材料没有的信息。 + +视频标题：{title} +材料来源：{source} +材料质量提醒：{transcript_quality_note} + +材料同时包含“语音/字幕”和本地视觉模型生成的“画面”记录，两类信息都是理解视频的重要证据。语音/字幕主要承载讲述逻辑、观点和因果关系；画面主要承载屏幕文字、PPT、代码、图表、操作步骤和场景变化。核心观点和分段要点必须综合两类证据，不能把画面只当作可忽略的附录。两者冲突时不要擅自猜测，应保守表述并指出不确定性。命令、URL、包名和版本号必须逐字来自材料；无法确认时不要补全或改写。 + +请按以下格式输出： + +## 视频主题 +用一段话概括视频整体内容。 + +## 核心观点 + +## 分段要点 +按材料中的时间节点组织，每个节点写出稍详细的内容要点。只能使用材料已有的时间节点，不要自行编造时间。 + +## 关键画面信息 +列出画面中对理解视频有关键作用的信息，并避免重复前文。没有则写“无”。 + +## 重要结论 + +## 可执行建议 + +这是第 {chunk_index}/{chunk_count} 段视频材料： + +{transcript} +""" +MULTIMODAL_CHUNK_SUMMARY_TEMPLATE = """你是严谨的视频内容总结助手。请只基于给定的视频材料提取结构化要点，不要编造材料没有的信息。 + +视频标题：{title} +材料来源：{source} +材料质量提醒：{transcript_quality_note} + +材料包含按时间排列的语音/字幕和画面描述，两类信息都是重要证据。语音/字幕主要承载讲述逻辑、观点和因果关系；画面主要承载屏幕文字、PPT、代码、图表、操作步骤和场景变化。时间节点与要点必须综合两类证据。 +先扫描完整分段，再用 6–10 条合并相邻时间点，必须覆盖分段的开头、中间和结尾，不要把输出预算都用在前半段。 + +请输出： + +## 本段主题 + +## 本段时间节点与要点 + +## 本段关键画面信息 + +## 本段重要细节 + +这是第 {chunk_index}/{chunk_count} 段视频材料： + {transcript} """ @@ -97,6 +169,8 @@ class VideoMeta: categories: list[str] = field(default_factory=list) subtitle_langs: list[str] = field(default_factory=list) automatic_caption_langs: list[str] = field(default_factory=list) + selected_subtitle_lang: str | None = None + selected_subtitle_type: str | None = None @dataclass @@ -112,6 +186,51 @@ class TranscriptInfo: warnings: list[str] = field(default_factory=list) +@dataclass +class FrameCandidate: + path: Path + timestamp: float + source: str + hash_value: int | None = None + + +@dataclass +class VisionConfig: + api_key: str + base_url: str + model: str + timeout: float + + +@dataclass +class VisionSamplingPlan: + scan_interval: float + max_gap: float + max_frames: int + + +@dataclass +class SummaryConfig: + provider: str + api_key: str + base_url: str + model: str + timeout: float + chunk_chars: int | None = None + context_length: int | None = None + model_context_length: int | None = None + + +@dataclass +class SummaryResult: + text: str + chunk_chars: int + chunks: int + chunk_lengths: list[int] + context_length: int | None + fallback_used: bool = False + + class AppError(RuntimeError): def __init__(self, code: str, message: str, hints: list[str] | None = None): super().__init__(message) @@ -241,6 +360,58 @@ def load_local_meta(path: Path) -> VideoMeta: title=path.stem or "local-video", id=name, webpage_url=str(path), + duration=probe_media_duration(path), + ) + + +def probe_media_duration(path: Path) -> float | None: + if not shutil.which("ffprobe"): + return None + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(path), + ], + text=True, + encoding="utf-8", + errors="replace", + capture_output=True, + ) + try: + duration = float(result.stdout.strip()) + except ValueError: + return None + return duration if duration > 0 else None + + +def build_vision_sampling_plan( + duration: float | None, + minimum_scan_interval: float, + maximum_gap: float, + maximum_frames: int, +) -> VisionSamplingPlan: + """根据时长控制候选密度和推理预算，CLI 参数仍作为用户边界。""" + if not duration or duration <= 0: + return VisionSamplingPlan( + scan_interval=minimum_scan_interval, + max_gap=maximum_gap, + max_frames=min(maximum_frames, 16), + ) + + desired_frames = max(8, math.floor(duration / 30) + 1) + frame_budget = min(maximum_frames, desired_frames) + scan_interval = max(minimum_scan_interval, duration / max(frame_budget * 4, 1)) + coverage_gap = min(maximum_gap, max(15, duration / max(frame_budget - 1, 1))) + return VisionSamplingPlan( + scan_interval=round(scan_interval, 3), + max_gap=round(coverage_gap, 3), + max_frames=frame_budget, ) @@ -264,6 +435,7 @@ def write_meta( if processing is not None: data["processing"] = processing if options is not None: + data["analysis_mode"] = options.get("analysis_mode") data["options"] = options (output_dir / "meta.json").write_text( json.dumps(data, ensure_ascii=False, indent=2), @@ -271,9 +443,63 @@ def write_meta( ) -def clean_generated_outputs(output_dir: Path) -> None: +def clean_generated_outputs(output_dir: Path, *, clear_cache: bool = False) -> None: for name in GENERATED_OUTPUT_FILES: (output_dir / name).unlink(missing_ok=True) + if clear_cache: + (output_dir / CACHE_FILE_NAME).unlink(missing_ok=True) + + +def load_pipeline_cache(output_dir: Path) -> dict: + path = output_dir / CACHE_FILE_NAME + if not path.is_file(): + return {"schema_version": CACHE_SCHEMA_VERSION, "stages": {}} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {"schema_version": CACHE_SCHEMA_VERSION, "stages": {}} + if payload.get("schema_version") != CACHE_SCHEMA_VERSION or not isinstance(payload.get("stages"), dict): + return {"schema_version": CACHE_SCHEMA_VERSION, "stages": {}} + return payload + + +def save_pipeline_cache(output_dir: Path, cache: dict) -> None: + cache["schema_version"] = CACHE_SCHEMA_VERSION + (output_dir / CACHE_FILE_NAME).write_text( + json.dumps(cache, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + +def stable_signature(payload: dict) -> str: + serialized = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +def source_signature(meta: VideoMeta, local_path: Path | None = None) -> str: + payload: dict = { + "id": meta.id, + "url": meta.webpage_url, + "duration": meta.duration, + } + if local_path and local_path.is_file(): + stat = local_path.stat() + payload["local_size"] = stat.st_size + payload["local_mtime_ns"] = stat.st_mtime_ns + return stable_signature(payload) + + +def cache_stage_matches(cache: dict, stage: str, signature: str) -> bool: + item = cache.get("stages", {}).get(stage) + return isinstance(item, dict) and item.get("signature") == signature + + +def update_cache_stage(cache: dict, stage: str, signature: str, **values) -> None: + cache.setdefault("stages", {})[stage] = { + "signature": signature, + "completed_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), + **values, + } def mark_stage(processing: dict, stage: str, started_at: float) -> None: @@ -290,20 +516,188 @@ def build_processing_info(args: argparse.Namespace, whisper_language: str | None "stages": {}, } options = { + "analysis_mode": "multimodal" if args.with_vision else "quick", "model_size": args.model_size, "language": args.language, "normalized_language": whisper_language, "sub_langs": args.sub_langs, "max_chars": args.max_chars, + "summary_provider": args.summary_provider, "force_transcribe": args.force_transcribe, "keep_audio": args.keep_audio, + "resume": args.resume, "no_llm": args.no_llm, "export_prompt": args.export_prompt or bool(args.summary_from_file), "summary_from_file": bool(args.summary_from_file), + "with_vision": args.with_vision, + "vision_scan_interval": args.vision_scan_interval, + "vision_max_gap": args.vision_max_gap, + "vision_max_frames": args.vision_max_frames, + "vision_batch_size": args.vision_batch_size, + "vision_frame_width": args.vision_frame_width, } return processing, options +def validate_args(args: argparse.Namespace) -> None: + if args.with_vision and (args.no_llm or args.export_prompt or args.summary_from_file): + raise AppError( + "invalid_arguments", + "--with-vision 暂不能与 --no-llm、--export-prompt 或 --summary-from-file 同时使用。", + ) + positive_values = { + "--vision-scan-interval": args.vision_scan_interval, + "--vision-max-gap": args.vision_max_gap, + "--vision-max-frames": args.vision_max_frames, + "--vision-batch-size": args.vision_batch_size, + "--vision-frame-width": args.vision_frame_width, + } + invalid = [name for name, value in positive_values.items() if value <= 0] + if invalid: + raise AppError("invalid_arguments", f"{', '.join(invalid)} 必须大于 0。") + + +def resolve_vision_config() -> VisionConfig: + base_url = (os.getenv("VISION_BASE_URL") or "http://host.docker.internal:11434/v1").rstrip("/") + model = os.getenv("VISION_MODEL") or "qwen3-vl:8b-instruct-q4_K_M" + api_key = os.getenv("VISION_API_KEY") or "ollama" + timeout_text = os.getenv("VISION_TIMEOUT") or "300" + try: + timeout = float(timeout_text) + except ValueError as exc: + raise AppError("invalid_vision_config", "VISION_TIMEOUT 必须是数字。") from exc + if timeout <= 0: + raise AppError("invalid_vision_config", "VISION_TIMEOUT 必须大于 0。") + return VisionConfig(api_key=api_key, base_url=base_url, model=model, timeout=timeout) + + +def resolve_summary_config(provider: str) -> SummaryConfig: + if provider not in {"ollama", "api"}: + raise AppError("invalid_summary_config", "SUMMARY_PROVIDER 只能是 ollama 或 api。") + timeout_text = os.getenv("SUMMARY_TIMEOUT") or "300" + try: + timeout = float(timeout_text) + except ValueError as exc: + raise AppError("invalid_summary_config", "SUMMARY_TIMEOUT 必须是数字。") from exc + if timeout <= 0: + raise AppError("invalid_summary_config", "SUMMARY_TIMEOUT 必须大于 0。") + + if provider == "ollama": + chunk_chars_text = (os.getenv("SUMMARY_LOCAL_CHUNK_CHARS") or "").strip() + chunk_chars = None + if chunk_chars_text and chunk_chars_text.lower() != "auto": + try: + chunk_chars = int(chunk_chars_text) + except ValueError as exc: + raise AppError( + "invalid_summary_config", + "SUMMARY_LOCAL_CHUNK_CHARS 必须是整数或 auto。", + ) from exc + if chunk_chars <= 0: + raise AppError("invalid_summary_config", "SUMMARY_LOCAL_CHUNK_CHARS 必须大于 0。") + + context_length_text = os.getenv("SUMMARY_OLLAMA_CONTEXT_LENGTH") or "8192" + try: + context_length = int(context_length_text) + except ValueError as exc: + raise AppError("invalid_summary_config", "SUMMARY_OLLAMA_CONTEXT_LENGTH 必须是整数。") from exc + if context_length <= 0: + raise AppError("invalid_summary_config", "SUMMARY_OLLAMA_CONTEXT_LENGTH 必须大于 0。") + return SummaryConfig( + provider=provider, + api_key=os.getenv("SUMMARY_API_KEY") or "ollama", + base_url=( + os.getenv("SUMMARY_BASE_URL") + or os.getenv("VISION_BASE_URL") + or "http://host.docker.internal:11434/v1" + ).rstrip("/"), + model=( + os.getenv("SUMMARY_MODEL") + or os.getenv("VISION_MODEL") + or "qwen3-vl:8b-instruct-q4_K_M" + ), + timeout=timeout, + chunk_chars=chunk_chars, + context_length=context_length, + ) + + api_key = os.getenv("OPENAI_API_KEY") or "" + if not api_key: + raise AppError( + "missing_openai_api_key", + "选择 API 总结时必须设置 OPENAI_API_KEY。", + ["也可以使用默认的 --summary-provider ollama 在本机完成总结。"], + ) + return SummaryConfig( + provider=provider, + api_key=api_key, + base_url=(os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1").rstrip("/"), + model=os.getenv("OPENAI_MODEL") or "gpt-4.1-mini", + timeout=timeout, + ) + + +def create_summary_client(config: SummaryConfig): + from openai import OpenAI + + client = OpenAI( + api_key=config.api_key, + base_url=config.base_url, + timeout=config.timeout, + ) + if config.provider != "ollama": + return client + + try: + models = client.models.list() + except Exception as exc: + raise AppError( + "summary_backend_unavailable", + f"无法连接本地总结服务：{exc}", + [f"确认 Ollama 已启动，并能从当前运行环境访问 {config.base_url}。"], + ) from exc + available = {item.id for item in models.data} + if config.model not in available: + raise AppError( + "summary_model_not_found", + f"Ollama 中未找到总结模型：{config.model}", + [f"请先在宿主机执行：ollama pull {config.model}"], + ) + model_info = ollama_request(config, "/api/show", {"model": config.model}) + context_values = [ + value + for key, value in (model_info.get("model_info") or {}).items() + if "context_length" in key.lower() and isinstance(value, int) + ] + if context_values: + config.model_context_length = max(context_values) + if config.context_length: + config.context_length = min(config.context_length, config.model_context_length) + return client + + +def ollama_native_base_url(config: SummaryConfig) -> str: + return config.base_url[:-3] if config.base_url.endswith("/v1") else config.base_url + + +def ollama_request(config: SummaryConfig, path: str, payload: dict) -> dict: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + request = urllib_request.Request( + ollama_native_base_url(config) + path, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib_request.urlopen(request, timeout=config.timeout) as response: + return json.loads(response.read().decode("utf-8")) + except urllib_error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"Ollama HTTP {exc.code}: {detail}") from exc + except (urllib_error.URLError, TimeoutError, json.JSONDecodeError) as exc: + raise RuntimeError(f"Ollama 请求失败：{exc}") from exc + + def maybe_warn_language(meta: VideoMeta, requested_language: str | None, warnings: list[str]) -> None: if requested_language != "zh": return @@ -373,23 +767,64 @@ def build_whisper_initial_prompt(terms: list[str]) -> str | None: def detect_subtitle_source(path: Path, meta: VideoMeta) -> tuple[str, list[str]]: name = path.name warnings: list[str] = [] + language = path.stem.rsplit(".", 1)[-1] if "." in path.stem else None + if language: + meta.selected_subtitle_lang = language + + # B 站 AI 字幕使用 ai-* 语言代码；实际下载文件比元数据接口更可信。 + if language and language.lower().startswith("ai-"): + if language not in meta.automatic_caption_langs: + meta.automatic_caption_langs.append(language) + meta.automatic_caption_langs.sort() + meta.selected_subtitle_type = "automatic" + warnings.append(f"使用的是平台自动字幕（{language}），内容可能存在识别错误。") + return "auto_subtitle", warnings + for lang in meta.subtitle_langs: if f".{lang}." in name or name.endswith(f".{lang}{path.suffix}"): + meta.selected_subtitle_lang = lang + meta.selected_subtitle_type = "manual" return "manual_subtitle", warnings for lang in meta.automatic_caption_langs: if f".{lang}." in name or name.endswith(f".{lang}{path.suffix}"): - warnings.append("使用的是自动字幕，内容可能存在识别错误。") + meta.selected_subtitle_lang = lang + meta.selected_subtitle_type = "automatic" + warnings.append(f"使用的是平台自动字幕（{lang}），内容可能存在识别错误。") return "auto_subtitle", warnings + if language: + if language not in meta.subtitle_langs: + meta.subtitle_langs.append(language) + meta.subtitle_langs.sort() + meta.selected_subtitle_type = "manual" + return "manual_subtitle", warnings warnings.append("已使用字幕文件，但无法确认是人工字幕还是自动字幕。") return "subtitle", warnings +def restore_cached_subtitle_metadata( + transcript_stage: dict, meta: VideoMeta, warnings: list[str] +) -> None: + meta.selected_subtitle_lang = transcript_stage.get("selected_subtitle_lang") + meta.selected_subtitle_type = transcript_stage.get("selected_subtitle_type") + if not meta.selected_subtitle_lang or meta.selected_subtitle_type != "automatic": + return + + language = meta.selected_subtitle_lang + if language not in meta.automatic_caption_langs: + meta.automatic_caption_langs.append(language) + meta.automatic_caption_langs.sort() + warning = f"使用的是平台自动字幕（{language}），内容可能存在识别错误。" + if warning not in warnings: + warnings.append(warning) + + def download_subtitle( url: str, output_dir: Path, cookies: CookieConfig | None, langs: str, meta: VideoMeta, + warnings: list[str] | None = None, ) -> tuple[Path, str, list[str]] | None: require_tool("yt-dlp") before = set(output_dir.glob("*")) @@ -423,7 +858,14 @@ def download_subtitle( source, warnings = detect_subtitle_source(existing[0], meta) return existing[0], source, warnings - if result.returncode != 0: + login_required = "Subtitles are only available when logged in" in result.stderr + if login_required: + message = "B 站字幕需要登录态，当前未获得可用字幕；已回退到 Whisper 转写。" + if warnings is not None and message not in warnings: + warnings.append(message) + print(message, file=sys.stderr) + print("提示：请将有效 cookies.txt 放到 cookies/ 目录后重试。", file=sys.stderr) + elif result.returncode != 0: app_error = classify_command_error(result.stderr, url, "字幕下载") print(f"字幕下载失败，将尝试音频转写：\n{app_error.message}", file=sys.stderr) for hint in app_error.hints: @@ -467,6 +909,61 @@ def download_audio(url: str, output_dir: Path, cookies: CookieConfig | None) -> raise AppError("audio_not_found", "音频下载完成，但没有找到生成的音频文件。") +def download_video(url: str, output_dir: Path, cookies: CookieConfig | None) -> Path: + require_tool("yt-dlp") + require_tool("ffmpeg") + + output_prefix = f"vision-source-{time.time_ns()}" + args = with_cookies( + [ + "yt-dlp", + "--no-playlist", + "--force-overwrites", + "-f", + "bestvideo[height<=1080]+bestaudio/best[height<=1080]/best", + "--merge-output-format", + "mp4", + "-o", + f"{output_prefix}.%(ext)s", + url, + ], + cookies, + ) + result = run_command(args, output_dir) + if result.returncode != 0: + raise classify_command_error(result.stderr, url, "视频下载") + + candidates = sorted( + path for path in output_dir.glob(f"{output_prefix}.*") if path.suffix.lower() in VIDEO_EXTS + ) + if candidates: + return candidates[0] + raise AppError("video_not_found", "视频下载完成，但没有找到生成的视频文件。") + + +def extract_audio_from_video(video_path: Path, output_dir: Path) -> Path: + require_tool("ffmpeg") + audio_path = output_dir / "vision-audio.mp3" + result = run_command( + [ + "ffmpeg", + "-y", + "-i", + str(video_path), + "-vn", + "-codec:a", + "libmp3lame", + "-q:a", + "2", + str(audio_path), + ], + output_dir, + ) + if result.returncode != 0 or not audio_path.is_file(): + raise classify_command_error(result.stderr, action="从视频提取音频") + return audio_path + + def clean_subtitle(path: Path) -> str: suffix = path.suffix.lower() if suffix == ".json3": @@ -501,6 +998,226 @@ def format_timestamp(seconds: float) -> str: return f"{hours:02d}:{minutes:02d}:{secs:02d}" +def image_difference_hash(path: Path) -> int: + from PIL import Image + + with Image.open(path) as image: + grayscale = image.convert("L") + width, height = grayscale.size + # 边缘常见固定页眉、水印和播放器装饰，裁掉后更关注主体内容变化。 + margin_x = max(1, int(width * 0.06)) + margin_y = max(1, int(height * 0.06)) + grayscale = grayscale.crop((margin_x, margin_y, width - margin_x, height - margin_y)).resize((9, 8)) + # Pillow 14 将移除 getdata()，同时保留对旧版 Pillow 的兼容。 + get_pixels = getattr(grayscale, "get_flattened_data", None) + pixels = list(get_pixels() if get_pixels else grayscale.getdata()) + value = 0 + for row in range(8): + offset = row * 9 + for column in range(8): + value = (value << 1) | int(pixels[offset + column] > pixels[offset + column + 1]) + return value + + +def hash_distance(left: int, right: int) -> int: + # 兼容 macOS 可能自带的 Python 3.8；该版本没有 int.bit_count()。 + return bin(left ^ right).count("1") + + +def evenly_sample_frames(frames: list[FrameCandidate], count: int) -> list[FrameCandidate]: + if count <= 0: + return [] + if len(frames) <= count: + return list(frames) + if count == 1: + return [frames[len(frames) // 2]] + indexes = { + round(index * (len(frames) - 1) / (count - 1)) + for index in range(count) + } + return [frames[index] for index in sorted(indexes)] + + +def filter_frame_candidates( + candidates: list[FrameCandidate], + max_gap: float, + max_frames: int, + hash_threshold: int = 10, +) -> list[FrameCandidate]: + if not candidates: + return [] + + # 相近时间的固定帧和转场帧代表同一画面，优先保留转场帧。 + merged: list[FrameCandidate] = [] + for candidate in sorted(candidates, key=lambda item: (item.timestamp, item.source != "scene")): + if merged and abs(candidate.timestamp - merged[-1].timestamp) <= 1: + if candidate.source == "scene": + merged[-1] = candidate + continue + merged.append(candidate) + + filtered: list[FrameCandidate] = [] + for candidate in merged: + candidate.hash_value = image_difference_hash(candidate.path) + if not filtered: + filtered.append(candidate) + continue + previous = filtered[-1] + changed = hash_distance(previous.hash_value or 0, candidate.hash_value) > hash_threshold + if candidate.source == "scene" or changed or candidate.timestamp - previous.timestamp >= max_gap: + filtered.append(candidate) + + if len(filtered) <= max_frames: + return filtered + + anchors: list[FrameCandidate] = [filtered[0], filtered[-1]] + anchors.extend(frame for frame in filtered if frame.source == "scene") + unique_anchors = list({frame.path: frame for frame in anchors}.values()) + unique_anchors.sort(key=lambda item: item.timestamp) + if len(unique_anchors) >= max_frames: + return evenly_sample_frames(unique_anchors, max_frames) + + anchor_paths = {frame.path for frame in unique_anchors} + remaining = [frame for frame in filtered if frame.path not in anchor_paths] + selected = unique_anchors + evenly_sample_frames(remaining, max_frames - len(unique_anchors)) + return sorted(selected, key=lambda item: item.timestamp) + + +def extract_keyframes( + video_path: Path, + output_dir: Path, + scan_interval: float, + max_gap: float, + max_frames: int, + frame_width: int, +) -> list[FrameCandidate]: + require_tool("ffmpeg") + run_name = f"{time.strftime('run-%Y%m%d-%H%M%S')}-{os.getpid()}" + frames_dir = output_dir / "frames" / run_name + frames_dir.mkdir(parents=True, exist_ok=True) + + with tempfile.TemporaryDirectory(prefix="video-summary-frames-") as temp_name: + temp_dir = Path(temp_name) + periodic_pattern = temp_dir / "periodic-%06d.jpg" + periodic_result = run_command( + [ + "ffmpeg", + "-y", + "-i", + str(video_path), + "-vf", + f"fps=1/{scan_interval},scale={frame_width}:-2:force_original_aspect_ratio=decrease,showinfo", + "-q:v", + "3", + "-start_number", + "0", + str(periodic_pattern), + ], + temp_dir, + ) + if periodic_result.returncode != 0: + raise classify_command_error(periodic_result.stderr, action="固定间隔抽帧") + + periodic_paths = sorted(temp_dir.glob("periodic-*.jpg")) + periodic_timestamps = [ + float(value) + for value in re.findall(r"\bn:\s*\d+.*?\bpts_time:\s*([0-9.]+)", periodic_result.stderr) + ] + if len(periodic_timestamps) != len(periodic_paths): + periodic_timestamps = [index * scan_interval for index in range(len(periodic_paths))] + candidates = [ + FrameCandidate(path=path, timestamp=timestamp, source="periodic") + for path, timestamp in zip(periodic_paths, periodic_timestamps) + ] + + scene_pattern = temp_dir / "scene-%06d.jpg" + scene_result = run_command( + [ + "ffmpeg", + "-y", + "-i", + str(video_path), + "-vf", + f"select=gt(scene\\,0.35),scale={frame_width}:-2:force_original_aspect_ratio=decrease,showinfo", + "-vsync", + "vfr", + "-q:v", + "3", + "-start_number", + "0", + str(scene_pattern), + ], + temp_dir, + ) + if scene_result.returncode != 0: + raise classify_command_error(scene_result.stderr, action="镜头变化抽帧") + + scene_paths = sorted(temp_dir.glob("scene-*.jpg")) + scene_timestamps = [ + float(value) + for value in re.findall(r"\bn:\s*\d+.*?\bpts_time:\s*([0-9.]+)", scene_result.stderr) + ] + candidates.extend( + FrameCandidate(path=path, timestamp=timestamp, source="scene") + for path, timestamp in zip(scene_paths, scene_timestamps) + ) + + selected = filter_frame_candidates(candidates, max_gap, max_frames) + persisted: list[FrameCandidate] = [] + for index, candidate in enumerate(selected, start=1): + timestamp = format_timestamp(candidate.timestamp).replace(":", "-") + target = frames_dir / f"{index:04d}-{timestamp}.jpg" + shutil.copy2(candidate.path, target) + persisted.append( + FrameCandidate( + path=target, + timestamp=candidate.timestamp, + source=candidate.source, + hash_value=candidate.hash_value, + ) + ) + + if not persisted: + raise AppError("frames_not_found", "视频抽帧完成，但没有得到可用关键帧。") + return persisted + + +def serialize_keyframes(frames: list[FrameCandidate], output_dir: Path) -> list[dict]: + return [ + { + "path": frame.path.relative_to(output_dir).as_posix(), + "timestamp": frame.timestamp, + "source": frame.source, + "hash_value": frame.hash_value, + } + for frame in frames + ] + + +def load_cached_keyframes(cache: dict, signature: str, output_dir: Path) -> list[FrameCandidate] | None: + if not cache_stage_matches(cache, "keyframes", signature): + return None + items = cache["stages"]["keyframes"].get("frames") + if not isinstance(items, list) or not items: + return None + frames: list[FrameCandidate] = [] + for item in items: + if not isinstance(item, dict): + return None + path = output_dir / str(item.get("path", "")) + if not path.is_file(): + return None + frames.append( + FrameCandidate( + path=path, + timestamp=float(item["timestamp"]), + source=str(item["source"]), + hash_value=item.get("hash_value"), + ) + ) + return frames + + def parse_subtitle_timestamp(value: str) -> str | None: match = re.match(r"(?:(\d+):)?(\d{1,2}):(\d{2})(?:[,.](\d{1,3}))?", value.strip()) if not match: @@ -607,6 +1324,531 @@ def transcribe_audio( return normalize_lines(lines) +def create_vision_client(config: VisionConfig): + from openai import OpenAI + + client = OpenAI( + api_key=config.api_key, + base_url=config.base_url, + timeout=config.timeout, + ) + try: + models = client.models.list() + except Exception as exc: + raise AppError( + "vision_backend_unavailable", + f"无法连接本地视觉服务：{exc}", + [f"确认 Ollama 已启动，并能从当前运行环境访问 {config.base_url}。"], + ) from exc + + available = {item.id for item in models.data} + if config.model not in available: + raise AppError( + "vision_model_not_found", + f"Ollama 中未找到视觉模型：{config.model}", + [f"请先在宿主机执行：ollama pull {config.model}"], + ) + return client + + +def image_data_url(path: Path) -> str: + encoded = base64.b64encode(path.read_bytes()).decode("ascii") + return f"data:image/jpeg;base64,{encoded}" + + +def parse_json_object(text: str) -> dict: + value = text.strip() + if value.startswith("```"): + value = re.sub(r"^```(?:json)?\s*", "", value, flags=re.IGNORECASE) + value = re.sub(r"\s*```$", "", value) + try: + data = json.loads(value) + except json.JSONDecodeError: + start = value.find("{") + end = value.rfind("}") + if start < 0 or end <= start: + raise + data = json.loads(value[start : end + 1]) + if not isinstance(data, dict): + raise ValueError("视觉模型返回值不是 JSON 对象") + return data + + +def normalize_text_field(value) -> str: + if isinstance(value, str): + return value.strip() + if isinstance(value, list): + return "；".join(str(item).strip() for item in value if str(item).strip()) + return "" + + +def normalize_visual_batch(payload: dict, frames: list[FrameCandidate], output_dir: Path) -> list[dict]: + items = payload.get("frames") + if not isinstance(items, list): + raise ValueError("视觉模型返回值缺少 frames 数组") + + indexed: dict[int, dict] = {} + for item in items: + if not isinstance(item, dict): + continue + try: + index = int(item.get("index")) + except (TypeError, ValueError): + continue + indexed[index] = item + + results: list[dict] = [] + for index, frame in enumerate(frames, start=1): + item = indexed.get(index) + if not item: + raise ValueError(f"视觉模型返回值缺少第 {index} 张图片的结果") + description = normalize_text_field(item.get("description")) + if not description: + raise ValueError(f"视觉模型没有描述第 {index} 张图片") + importance = normalize_text_field(item.get("importance")).lower() + if importance not in {"high", "medium", "low"}: + importance = "medium" + visible_text = normalize_text_field(item.get("visible_text")) + results.append( + { + "timestamp_seconds": round(frame.timestamp, 3), + "timestamp": format_timestamp(frame.timestamp), + "image": frame.path.relative_to(output_dir).as_posix(), + "frame_source": frame.source, + "description": description[:160].rstrip(), + "visible_text": visible_text[:320].rstrip(), + "importance": importance, + "uncertainty": normalize_text_field(item.get("uncertainty")), + } + ) + return results + + +def write_visual_outputs( + output_dir: Path, + config: VisionConfig, + frames: list[dict], + status: str, + error: str | None = None, + metrics: dict | None = None, +) -> tuple[Path, Path]: + payload = { + "schema_version": 1, + "status": status, + "model": config.model, + "frame_count": len(frames), + "frames": frames, + } + if error: + payload["error"] = error + if metrics: + payload["metrics"] = metrics + + json_path = output_dir / "visual_context.json" + json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + + lines = ["# 视频画面解析", "", f"- 状态：{status}", f"- 模型：{config.model}", f"- 已解析帧数：{len(frames)}"] + if error: + lines.extend([f"- 错误：{error}"]) + for frame in frames: + lines.extend( + [ + "", + f"## [{frame['timestamp']}]", + "", + f"- 关键帧：`{frame['image']}`", + f"- 画面：{frame['description']}", + f"- 屏幕文字：{frame['visible_text'] or '无'}", + f"- 重要程度：{frame['importance']}", + f"- 不确定性：{frame['uncertainty'] or '无'}", + ] + ) + markdown_path = output_dir / "visual_context.md" + markdown_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8") + return json_path, markdown_path + + +def nearby_transcript_context( + transcript: str, + timestamp: float, + window_seconds: float = 15, + max_chars: int = 500, +) -> str: + timed_lines: list[tuple[float, str]] = [] + for line in transcript.splitlines(): + match = re.match(r"^\[(\d+:\d{2}:\d{2})\]\s*(.+)$", line.strip()) + if match: + timed_lines.append((timestamp_to_seconds(match.group(1)), match.group(2))) + nearby = [text for seconds, text in timed_lines if abs(seconds - timestamp) <= window_seconds] + if not nearby and timed_lines: + nearby = [min(timed_lines, key=lambda item: abs(item[0] - timestamp))[1]] + value = " ".join(nearby) + return value[:max_chars].rstrip() + + +def analyze_keyframes( + client, + config: VisionConfig, + frames: list[FrameCandidate], + output_dir: Path, + batch_size: int, + existing_results: list[dict] | None = None, + transcript: str = "", + context_terms: list[str] | None = None, +) -> list[dict]: + current_images = {frame.path.relative_to(output_dir).as_posix() for frame in frames} + results = [ + item + for item in (existing_results or []) + if isinstance(item, dict) and item.get("image") in current_images + ] + completed_images = {item["image"] for item in results} + pending_frames = [ + frame + for frame in frames + if frame.path.relative_to(output_dir).as_posix() not in completed_images + ] + metrics = { + "batch_size": batch_size, + "effective_batch_size": batch_size, + "requested_frames": len(pending_frames), + "batch_seconds": [], + "fallback_splits": 0, + } + + def request_batch(batch: list[FrameCandidate]) -> list[dict]: + batch_started = time.monotonic() + content: list[dict] = [ + { + "type": "text", + "text": ( + "请逐张分析后只返回 JSON 对象，不要使用 Markdown。格式为：" + '{"frames":[{"index":1,"description":"画面事实",' + '"visible_text":"可辨认的屏幕文字，没有则为空字符串",' + '"importance":"high|medium|low","uncertainty":"不确定信息，没有则为空字符串"}]}。' + "必须为每张图片返回一项。相邻字幕只能用于校正 OCR 和判断信息增量，不能当作图片事实。" + "description 限 80 个中文字符，只写字幕没有表达的布局、操作、图表或画面变化。" + "visible_text 限 200 个中文字符，只保留新增或变化的关键文字、命令、参数和数字；" + "忽略重复页眉、页脚、水印、栏目名及前一张已经出现的文字。" + "importance 按新增价值判断：high=字幕之外的重要信息，medium=补强字幕，" + "low=主要重复字幕或装饰。无法确认的英文、命令或小字必须写入 uncertainty。" + ), + } + ] + if context_terms: + content.append( + { + "type": "text", + "text": "视频已知术语（仅用于 OCR 拼写校正）：" + "、".join(context_terms), + } + ) + for index, frame in enumerate(batch, start=1): + content.extend( + [ + { + "type": "text", + "text": ( + f"第 {index} 张图片，对应视频时间 {format_timestamp(frame.timestamp)}。" + f"相邻字幕：{nearby_transcript_context(transcript, frame.timestamp) or '无'}" + ), + }, + { + "type": "image_url", + "image_url": {"url": image_data_url(frame.path)}, + }, + ] + ) + + last_error: Exception | None = None + for attempt in range(2): + try: + response = client.chat.completions.create( + model=config.model, + messages=[ + { + "role": "system", + "content": "你是严谨的视频关键帧分析器，只报告图片中明确可见的信息。", + }, + {"role": "user", "content": content}, + ], + temperature=0, + response_format={"type": "json_object"}, + max_tokens=max(500, min(900, len(batch) * 180)), + ) + response_text = response.choices[0].message.content or "" + normalized = normalize_visual_batch(parse_json_object(response_text), batch, output_dir) + metrics["batch_seconds"].append( + { + "frames": len(batch), + "seconds": round(time.monotonic() - batch_started, 3), + } + ) + return normalized + except Exception as exc: + last_error = exc + if attempt == 0: + time.sleep(1) + if len(batch) > 1: + metrics["fallback_splits"] += 1 + midpoint = len(batch) // 2 + return request_batch(batch[:midpoint]) + request_batch(batch[midpoint:]) + raise last_error or RuntimeError("视觉批次解析失败") + + try: + offset = 0 + effective_batch_size = batch_size + while offset < len(pending_frames): + batch = pending_frames[offset : offset + effective_batch_size] + fallback_count = metrics["fallback_splits"] + results.extend(request_batch(batch)) + results.sort(key=lambda item: float(item["timestamp_seconds"])) + write_visual_outputs(output_dir, config, results, "processing", metrics=metrics) + if metrics["fallback_splits"] > fallback_count and effective_batch_size > 1: + effective_batch_size = max(1, effective_batch_size // 2) + metrics["effective_batch_size"] = effective_batch_size + offset += len(batch) + except Exception as exc: + write_visual_outputs(output_dir, config, results, "failed", str(exc), metrics) + raise AppError( + "vision_analysis_failed", + f"本地视觉模型解析失败：{exc}", + ["已保留关键帧和已完成的视觉解析结果，便于调试。"], + ) from exc + + write_visual_outputs(output_dir, config, results, "completed", metrics=metrics) + return results + + +def load_visual_results(output_dir: Path, model: str) -> list[dict]: + path = output_dir / "visual_context.json" + if not path.is_file(): + return [] + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return [] + if payload.get("model") != model or not isinstance(payload.get("frames"), list): + return [] + return [item for item in payload["frames"] if isinstance(item, dict)] + + +def load_visual_metrics(output_dir: Path) -> dict: + path = output_dir / "visual_context.json" + if not path.is_file(): + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return payload.get("metrics") if isinstance(payload.get("metrics"), dict) else {} + + +def visual_results_cover_frames( + results: list[dict], + frames: list[FrameCandidate], + output_dir: Path, +) -> bool: + expected = {frame.path.relative_to(output_dir).as_posix() for frame in frames} + completed = {item.get("image") for item in results} + return bool(expected) and expected.issubset(completed) + + +def timestamp_to_seconds(value: str) -> float: + parts = [int(part) for part in value.split(":")] + if len(parts) != 3: + raise ValueError(value) + return float(parts[0] * 3600 + parts[1] * 60 + parts[2]) + + +def normalize_visual_line(value: str) -> str: + return re.sub(r"\s+", " ", value).strip().casefold() + + +def split_visual_lines(value: str) -> list[str]: + return [ + line.strip() + for line in re.split(r"[\r\n；;]+", value) + if line.strip() + ] + + +def compact_visual_material(visual_frames: list[dict], transcript: str = "") -> list[dict]: + ordered = sorted(visual_frames, key=lambda item: float(item["timestamp_seconds"])) + normalized_transcript = normalize_visual_line(transcript) + line_counts: dict[str, int] = {} + for frame in ordered: + for line in {normalize_visual_line(item) for item in split_visual_lines(frame.get("visible_text", ""))}: + line_counts[line] = line_counts.get(line, 0) + 1 + + boilerplate_threshold = max(3, (len(ordered) + 3) // 4) + boilerplate = { + line + for line, count in line_counts.items() + if count >= boilerplate_threshold and len(line) <= 40 + } + + compacted: list[dict] = [] + previous_lines: set[str] = set() + previous_description = "" + for frame in ordered: + description = re.sub(r"\s+", " ", str(frame.get("description", ""))).strip() + raw_lines = split_visual_lines(str(frame.get("visible_text", ""))) + useful_lines = [ + line + for line in raw_lines + if normalize_visual_line(line) not in boilerplate + and ( + len(normalize_visual_line(line)) < 4 + or normalize_visual_line(line) not in normalized_transcript + ) + ] + current_lines = {normalize_visual_line(line) for line in useful_lines} + delta_lines = [ + line + for line in useful_lines + if normalize_visual_line(line) not in previous_lines + ] + description_key = normalize_visual_line(description) + description_similarity = ( + difflib.SequenceMatcher(None, previous_description, description_key).ratio() + if previous_description and description_key + else 0 + ) + if not delta_lines and ( + description_key == previous_description + or description_similarity >= 0.88 + or frame.get("importance") == "low" + ): + previous_lines = current_lines + continue + + visible_text = "\n".join(delta_lines) + if len(visible_text) > 240: + visible_text = visible_text[:240].rstrip() + "……" + description_limit = 80 if visible_text else 120 + if len(description) > description_limit: + description = description[:description_limit].rstrip() + "……" + + compacted.append( + { + **frame, + "description": description, + "visible_text": visible_text, + } + ) + previous_lines = current_lines + previous_description = description_key + return compacted + + +def limit_visual_events( + visual_events: list[tuple[float, int, str]], + char_budget: int, +) -> list[tuple[float, int, str]]: + if char_budget <= 0 or not visual_events: + return [] + total_chars = sum(len(text) + 1 for _seconds, _priority, text in visual_events) + if total_chars <= char_budget: + return visual_events + + target_count = max(1, min(len(visual_events), math.floor(len(visual_events) * char_budget / total_chars))) + if target_count == 1: + indexes = [len(visual_events) // 2] + else: + # 在时间均匀覆盖之外，优先保留可执行命令和链接等高价值字面证据， + # 避免自动字幕中的残缺转写覆盖画面里更可靠的原文。 + evidence_pattern = re.compile( + r"(?:^|[；：\s])(?:\$\s*)?" + r"(?:npx|npm|pnpm|yarn|pipx?|uv|docker|git|cargo|go|brew)\s+\S+" + r"|https?://\S+|github\.com/\S+", + re.IGNORECASE, + ) + evidence_indexes = [ + index + for index, (_seconds, _priority, text) in enumerate(visual_events) + if evidence_pattern.search(text) + ] + reserved_count = min(len(evidence_indexes), max(1, target_count // 5)) + indexes_set = set(evidence_indexes[-reserved_count:]) + coverage_count = target_count - len(indexes_set) + if coverage_count > 0: + coverage_indexes = [ + round(index * (len(visual_events) - 1) / max(coverage_count - 1, 1)) + for index in range(coverage_count) + ] + for index in coverage_indexes: + if len(indexes_set) >= target_count: + break + indexes_set.add(index) + while len(indexes_set) < target_count: + remaining = set(range(len(visual_events))) - indexes_set + next_index = max( + remaining, + key=lambda candidate: min(abs(candidate - selected) for selected in indexes_set), + ) + indexes_set.add(next_index) + indexes = sorted(indexes_set) + selected = [visual_events[index] for index in indexes] + per_event_budget = max(40, char_budget // max(len(selected), 1) - 1) + limited = [ + (seconds, priority, text if len(text) <= per_event_budget else text[:per_event_budget].rstrip() + "……") + for seconds, priority, text in selected + ] + while limited and sum(len(text) + 1 for _seconds, _priority, text in limited) > char_budget: + limited.pop() + return limited + + +def build_multimodal_context( + transcript: str, + visual_frames: list[dict], + max_chars: int | None = None, +) -> str: + transcript_events: list[tuple[float, int, str]] = [] + untimed: list[str] = [] + for line in transcript.splitlines(): + match = re.match(r"^\[(\d+:\d{2}:\d{2})\]\s*(.+)$", line.strip()) + if not match: + if line.strip(): + untimed.append(line.strip()) + continue + transcript_events.append((timestamp_to_seconds(match.group(1)), 0, f"[{match.group(1)}] {match.group(2)}")) + + visual_events: list[tuple[float, int, str]] = [] + for frame in compact_visual_material(visual_frames, transcript): + details = [f"[{frame['timestamp']}] 画面：{frame['description']}"] + if frame.get("visible_text"): + details.append(f"屏幕文字：{frame['visible_text']}") + if frame.get("uncertainty"): + details.append(f"不确定性：{frame['uncertainty']}") + visual_events.append((float(frame["timestamp_seconds"]), 1, "；".join(details))) + + header = [ + "# 多模态视频材料", + "", + "以下内容按时间轴排列；普通行来自语音/字幕，标有“画面”的行是视觉模型提取的画面证据，两类信息都需要参与内容理解。", + ] + if untimed: + header.extend(["", "## 无时间戳的语音/字幕", "", *untimed]) + header.extend(["", "## 时间轴", ""]) + + if max_chars: + base_chars = len("\n".join([*header, *(text for _seconds, _priority, text in transcript_events)])) + 1 + visual_events = limit_visual_events(visual_events, max(0, max_chars - base_chars)) + + lines = [ + *header, + *( + text + for _seconds, _priority, text in sorted( + [*transcript_events, *visual_events], + key=lambda item: (item[0], item[1]), + ) + ), + ] + return "\n".join(lines).rstrip() + "\n" + + def split_text(text: str, max_chars: int) -> list[str]: if max_chars <= 0: raise RuntimeError("--max-chars 必须大于 0") @@ -631,6 +1873,53 @@ def split_text(text: str, max_chars: int) -> list[str]: return chunks +def split_text_balanced(text: str, max_chars: int) -> list[str]: + """在不超过上限的前提下均衡分段，避免最后一段明显过短。""" + if len(text) <= max_chars: + return [text] + chunk_count = math.ceil(len(text) / max_chars) + chunks: list[str] = [] + start = 0 + for index in range(chunk_count - 1): + remaining_chunks = chunk_count - index + ideal_end = start + math.ceil((len(text) - start) / remaining_chunks) + max_end = min(len(text), start + max_chars) + before = text.rfind("\n", start + 1, ideal_end + 1) + after = text.find("\n", ideal_end, max_end + 1) + candidates = [position + 1 for position in (before, after) if position >= start] + end = min(candidates, key=lambda position: abs(position - ideal_end)) if candidates else ideal_end + chunks.append(text[start:end].rstrip("\n")) + start = end + chunks.append(text[start:].lstrip("\n")) + return chunks + + +def determine_summary_chunk_chars(config: SummaryConfig, max_chars: int) -> int: + if config.chunk_chars: + return min(max_chars, config.chunk_chars) + if config.provider != "ollama" or not config.context_length: + return max_chars + + # 为系统提示、分段模板和最终输出预留约 1800 tokens；中文按约 1 字符/token 保守估算。 + context_safe_chars = max(1200, config.context_length - 1800) + return min(max_chars, context_safe_chars) + + +def build_summary_merge_material(partials: list[str]) -> str: + sections = "\n\n".join( + f"## 第 {index}/{len(partials)} 段摘要\n\n{partial}" + for index, partial in enumerate(partials, start=1) + ) + return ( + "# 合并要求\n\n" + "以下是全部分段摘要。最终结果必须逐段覆盖，每一段都至少保留其开头、中间和结尾的重要时间节点；" + "可以合并相邻节点，但不得只详细总结第一段或省略最后一段的后半部分。" + "若材料本身没有注明截断，不得声称“原文截断”。\n\n" + "# 全部分段摘要\n\n" + f"{sections}" + ) + + def load_prompt_template(prompt_text: str | None, prompt_file: str | None) -> str: if prompt_text and prompt_file: raise AppError("invalid_arguments", "--prompt 和 --prompt-file 不能同时使用。") @@ -667,26 +1956,32 @@ def validate_model_size(value: str) -> str: def base_transcript_source(source: str) -> str: suffix = "_partials" - return source[: -len(suffix)] if source.endswith(suffix) else source + normalized = source[: -len(suffix)] if source.endswith(suffix) else source + return normalized.split("+", 1)[0] def transcript_quality_note(source: str) -> str: - source = base_transcript_source(source) - if source == "whisper": - return ( + has_vision = "+vision" in source + normalized_source = base_transcript_source(source) + if normalized_source == "whisper": + note = ( "本文本由 Whisper 音频转写生成，可能存在专有名词、英文术语、产品名、人名、数字、标点和断句错误。" "请结合上下文只修正明显误识别；无法确定时保守表述，不要编造原文没有的信息。" ) - if source == "auto_subtitle": - return ( + elif normalized_source == "auto_subtitle": + note = ( "本文本来自平台自动字幕，可能存在自动识别错误。" "请结合上下文只修正明显误识别；无法确定时保守表述，不要编造原文没有的信息。" ) - if source == "manual_subtitle": - return "本文本来自人工字幕，通常较可靠，但仍可能存在错字、漏字或排版问题；请只基于原文总结。" - if source == "file": - return "本文本来自已有转写稿，来源质量未知；如遇疑似识别错误，请结合上下文保守理解，不要编造。" - return "转写文本可能存在识别、清洗或断句错误；请结合上下文保守总结，不要编造原文没有的信息。" + elif normalized_source == "manual_subtitle": + note = "本文本来自人工字幕，通常较可靠，但仍可能存在错字、漏字或排版问题；请只基于原文总结。" + elif normalized_source == "file": + note = "本文本来自已有转写稿，来源质量未知；如遇疑似识别错误，请结合上下文保守理解，不要编造。" + else: + note = "转写文本可能存在识别、清洗或断句错误；请结合上下文保守总结，不要编造原文没有的信息。" + if has_vision: + note += " 画面描述由本地视觉模型生成，也可能存在漏读或误判；不确定内容不得当作事实。" + return note def render_prompt_template( @@ -728,8 +2023,9 @@ def render_chunk_summary_prompt( chunk_index: int, chunk_count: int, ) -> str: + template = MULTIMODAL_CHUNK_SUMMARY_TEMPLATE if "+vision" in source else CHUNK_SUMMARY_TEMPLATE return render_prompt_template( - CHUNK_SUMMARY_TEMPLATE, + template, transcript, meta, source, @@ -739,62 +2035,93 @@ def render_chunk_summary_prompt( def summarize_with_llm( + client, + config: SummaryConfig, transcript: str, meta: VideoMeta, source: str, max_chars: int, prompt_template: str, -) -> str: - from openai import OpenAI - - api_key = os.getenv("OPENAI_API_KEY") - if not api_key: - raise AppError( - "missing_openai_api_key", - "缺少 OPENAI_API_KEY。可设置 .env，或使用 --no-llm / --export-prompt。", - ["如果没有 API 额度，可以使用 --export-prompt 生成提示词后复制到 ChatGPT。"], - ) - - client = OpenAI( - api_key=api_key, - base_url=os.getenv("OPENAI_BASE_URL") or None, - ) - model = os.getenv("OPENAI_MODEL", "gpt-4.1-mini") - chunks = split_text(transcript, max_chars) - if len(chunks) == 1: - prompt = render_prompt_template(prompt_template, chunks[0], meta, source, 1, 1) +) -> SummaryResult: + model = config.model + effective_max_chars = determine_summary_chunk_chars(config, max_chars) + single_max_tokens = 1300 if config.provider == "ollama" else None + final_max_tokens = 1300 if config.provider == "ollama" else None + # 自动放大分段后需要同步提高中间摘要预算，否则长分段后半部分容易被截断。 + partial_max_tokens = 350 if config.provider == "ollama" else None + fallback_used = False + + while True: + chunks = split_text_balanced(transcript, effective_max_chars) try: - return call_llm(client, model, prompt) - except Exception as exc: - raise classify_llm_error(exc) from exc + if len(chunks) == 1: + prompt = render_prompt_template(prompt_template, chunks[0], meta, source, 1, 1) + text = call_summary_llm(client, config, model, prompt, single_max_tokens) + else: + partials: list[str] = [] + for index, chunk in enumerate(chunks, start=1): + prompt = render_chunk_summary_prompt(chunk, meta, source, index, len(chunks)) + partials.append(call_summary_llm(client, config, model, prompt, partial_max_tokens)) - partials: list[str] = [] - for index, chunk in enumerate(chunks, start=1): - prompt = render_chunk_summary_prompt( - chunk, - meta, - source, - index, - len(chunks), - ) - try: - partials.append(call_llm(client, model, prompt)) + # 多段长文本先提取细节，再由最终模板合并，减少直接二次压缩造成的要点丢失。 + final_prompt = render_prompt_template( + prompt_template, + build_summary_merge_material(partials), + meta, + f"{source}_partials", + 1, + 1, + ) + text = call_summary_llm(client, config, model, final_prompt, final_max_tokens) + return SummaryResult( + text=text, + chunk_chars=effective_max_chars, + chunks=len(chunks), + chunk_lengths=[len(chunk) for chunk in chunks], + context_length=config.context_length, + fallback_used=fallback_used, + ) except Exception as exc: + if config.provider == "ollama" and is_context_length_error(exc) and effective_max_chars > 1200: + effective_max_chars = max(1200, effective_max_chars // 2) + fallback_used = True + continue raise classify_llm_error(exc) from exc - # 多段长文本先提取细节，再由最终模板合并，减少直接二次压缩造成的要点丢失。 - final_prompt = render_prompt_template( - prompt_template, - "\n\n".join(partials), - meta, - f"{source}_partials", - 1, - 1, + +def is_context_length_error(exc: Exception) -> bool: + message = str(exc).lower() + return any( + marker in message + for marker in ("context length", "context window", "too long", "exceeds the context") ) - try: - return call_llm(client, model, final_prompt) - except Exception as exc: - raise classify_llm_error(exc) from exc + + +def call_summary_llm(client, config: SummaryConfig, model: str, prompt: str, max_tokens: int | None) -> str: + if config.provider != "ollama": + return call_llm(client, model, prompt, max_tokens) + options = {"temperature": 0.2} + if config.context_length: + options["num_ctx"] = config.context_length + if max_tokens is not None: + options["num_predict"] = max_tokens + response = ollama_request( + config, + "/api/chat", + { + "model": model, + "messages": [ + { + "role": "system", + "content": "你是严谨的视频内容总结助手，只基于给定文本总结，并使用中文输出。", + }, + {"role": "user", "content": prompt}, + ], + "stream": False, + "options": options, + }, + ) + return str((response.get("message") or {}).get("content") or "") def classify_llm_error(exc: Exception) -> AppError: @@ -849,28 +2176,33 @@ def print_json(payload: dict) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2)) -def call_llm(client, model: str, prompt: str) -> str: - response = client.chat.completions.create( - model=model, - messages=[ +def call_llm(client, model: str, prompt: str, max_tokens: int | None = None) -> str: + request = { + "model": model, + "messages": [ { "role": "system", "content": "你是严谨的视频内容总结助手，只基于给定文本总结，并使用中文输出。", }, {"role": "user", "content": prompt}, ], - temperature=0.2, + "temperature": 0.2, + } + if max_tokens is not None: + request["max_tokens"] = max_tokens + response = client.chat.completions.create( + **request, ) return response.choices[0].message.content or "" -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="字幕优先、Whisper 兜底的视频总结 CLI") +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="支持快速解析和多模态解析的视频总结 CLI") parser.add_argument("url", nargs="?", help="视频链接，或容器/本机可访问的本地音视频文件路径") parser.add_argument("--output", default="outputs", help="输出目录，默认 outputs") parser.add_argument("--model-size", default="small", choices=AVAILABLE_MODEL_SIZES, help="faster-whisper 模型大小，默认 small") parser.add_argument("--language", default="zh", help="Whisper 识别语言；中文视频用 zh，英文视频建议 en，不确定可用 auto；默认 zh") - parser.add_argument("--sub-langs", default="zh-Hans,zh-CN,zh,en", help="字幕语言优先级") + parser.add_argument("--sub-langs", default="zh.*,ai-zh,en.*", help="字幕语言匹配规则，默认覆盖中文、B 站 AI 中文字幕和英文") parser.add_argument("--cookies", default=None, help="cookies.txt 路径；默认自动尝试 /app/cookies/cookies.txt，缺失时忽略") parser.add_argument("--force-transcribe", action="store_true", help="跳过字幕，强制下载音频并转写") parser.add_argument("--keep-audio", action="store_true", help="保留下载的音频文件") @@ -882,7 +2214,25 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--prompt", default=None, help="自定义总结 prompt 模板文本") parser.add_argument("--prompt-file", default=None, help="从文件读取自定义总结 prompt 模板") parser.add_argument("--max-chars", type=int, default=12000, help="LLM 单段最大字符数") - return parser.parse_args() + parser.add_argument( + "--summary-provider", + choices=("ollama", "api"), + default=os.getenv("SUMMARY_PROVIDER") or "ollama", + help="最终总结提供方，默认 ollama；使用 api 时读取 OPENAI_* 配置", + ) + parser.add_argument("--no-resume", dest="resume", action="store_false", help="忽略阶段缓存并强制重新处理") + parser.set_defaults(resume=True) + parser.add_argument( + "--with-vision", + action="store_true", + help="启用多模态解析（语音/字幕 + 关键帧）；默认使用快速解析，仅处理语音/字幕", + ) + parser.add_argument("--vision-scan-interval", type=float, default=5, help="视觉候选帧扫描间隔秒数，默认 5") + parser.add_argument("--vision-max-gap", type=float, default=45, help="相似画面最长保留间隔上限秒数，默认 45") + parser.add_argument("--vision-max-frames", type=int, default=60, help="送入视觉模型的最大关键帧数，默认 60") + parser.add_argument("--vision-batch-size", type=int, default=4, help="每次视觉请求包含的关键帧数，默认 4") + parser.add_argument("--vision-frame-width", type=int, default=768, help="关键帧缩放宽度，默认 768") + return parser.parse_args(argv) def main() -> int: @@ -896,12 +2246,36 @@ def main() -> int: transcript_path: Path | None = None prompt_path: Path | None = None summary_path: Path | None = None + visual_context_path: Path | None = None + multimodal_context_path: Path | None = None meta: VideoMeta | None = None transcript_source: str | None = None whisper_context_terms: list[str] | None = None + vision_config: VisionConfig | None = None + vision_client = None + summary_config: SummaryConfig | None = None + summary_client = None + vision_video_path: Path | None = None + downloaded_vision_video = False + pipeline_cache: dict = {"schema_version": CACHE_SCHEMA_VERSION, "stages": {}} + pipeline_source_signature: str | None = None + transcript_cache_signature: str | None = None + keyframe_cache_signature: str | None = None + vision_cache_signature: str | None = None + vision_sampling_plan: VisionSamplingPlan | None = None + keyframes: list[FrameCandidate] | None = None + summary_input: str | None = None + summary_source: str | None = None + summary_prompt_template = prompt_template warnings: list[str] = [] try: + validate_args(args) + if not (args.no_llm or args.export_prompt or args.summary_from_file): + stage_started = time.monotonic() + summary_config = resolve_summary_config(args.summary_provider) + summary_client = create_summary_client(summary_config) + mark_stage(processing, "summary_preflight", stage_started) output_root = Path(args.output).resolve() output_root.mkdir(parents=True, exist_ok=True) @@ -921,7 +2295,7 @@ def main() -> int: ) output_dir = output_root / safe_name(meta.title) output_dir.mkdir(parents=True, exist_ok=True) - clean_generated_outputs(output_dir) + clean_generated_outputs(output_dir, clear_cache=True) transcript_path = output_dir / "transcript.txt" transcript_path.write_text(transcript, encoding="utf-8") transcript_source = "file" @@ -963,59 +2337,287 @@ def main() -> int: maybe_warn_language(meta, args.language, warnings) output_dir = output_root / (safe_name(meta.title) if is_local_file else safe_name(f"{meta.id}-{meta.title}")) output_dir.mkdir(parents=True, exist_ok=True) - clean_generated_outputs(output_dir) + if not args.resume: + clean_generated_outputs(output_dir, clear_cache=True) + pipeline_cache = load_pipeline_cache(output_dir) + pipeline_source_signature = source_signature(meta, local_input if is_local_file else None) + transcript_cache_signature = stable_signature( + { + "source": pipeline_source_signature, + "model_size": args.model_size, + "language": whisper_language, + "sub_langs": args.sub_langs, + "force_transcribe": args.force_transcribe, + "cookies": bool(cookies and (cookies.file or cookies.browser)), + } + ) write_meta(meta, output_dir, None, warnings, processing, options) - subtitle_path = None - if is_local_file: + if args.with_vision: stage_started = time.monotonic() - whisper_context_terms = build_whisper_context_terms(meta) - transcript = transcribe_audio( - local_input, - args.model_size, - whisper_language, - build_whisper_initial_prompt(whisper_context_terms), + vision_config = resolve_vision_config() + vision_sampling_plan = build_vision_sampling_plan( + meta.duration, + args.vision_scan_interval, + args.vision_max_gap, + args.vision_max_frames, ) - mark_stage(processing, "transcribe", stage_started) - transcript_source = "whisper" - elif not args.force_transcribe: + keyframe_cache_signature = stable_signature( + { + "source": pipeline_source_signature, + "scan_interval": vision_sampling_plan.scan_interval, + "max_gap": vision_sampling_plan.max_gap, + "max_frames": vision_sampling_plan.max_frames, + "frame_width": args.vision_frame_width, + "selection_schema": 2, + } + ) + vision_cache_signature = stable_signature( + { + "keyframes": keyframe_cache_signature, + "model": vision_config.model, + "batch_size": args.vision_batch_size, + "transcript_stage": transcript_cache_signature, + "context_terms": build_whisper_context_terms(meta), + "prompt_schema": 3, + } + ) + processing["vision"] = { + "model": vision_config.model, + "status": "preparing", + "frames": 0, + "sampling": asdict(vision_sampling_plan), + } + write_meta(meta, output_dir, None, warnings, processing, options) + vision_client = create_vision_client(vision_config) + mark_stage(processing, "vision_preflight", stage_started) + keyframes = load_cached_keyframes(pipeline_cache, keyframe_cache_signature, output_dir) + if keyframes: + processing.setdefault("cache_hits", []).append("keyframes") + elif is_local_file: + vision_video_path = local_input + else: + stage_started = time.monotonic() + vision_video_path = download_video(args.url, output_dir, cookies) + downloaded_vision_video = True + mark_stage(processing, "download_video", stage_started) + write_meta(meta, output_dir, None, warnings, processing, options) + + transcript_path = output_dir / "transcript.txt" + transcript: str | None = None + if ( + args.resume + and transcript_cache_signature + and cache_stage_matches(pipeline_cache, "transcript", transcript_cache_signature) + and transcript_path.is_file() + ): + transcript = transcript_path.read_text(encoding="utf-8") + transcript_stage = pipeline_cache["stages"]["transcript"] + transcript_source = transcript_stage.get("source") or "unknown" + whisper_context_terms = transcript_stage.get("whisper_context_terms") + restore_cached_subtitle_metadata(transcript_stage, meta, warnings) + processing.setdefault("cache_hits", []).append("transcript") + + subtitle_path = None + if transcript is None and not is_local_file and not args.force_transcribe: stage_started = time.monotonic() - subtitle_info = download_subtitle(args.url, output_dir, cookies, args.sub_langs, meta) + subtitle_info = download_subtitle( + args.url, + output_dir, + cookies, + args.sub_langs, + meta, + warnings, + ) mark_stage(processing, "download_subtitle", stage_started) if subtitle_info: subtitle_path, transcript_source, subtitle_warnings = subtitle_info warnings.extend(subtitle_warnings) - - if not is_local_file and subtitle_path: - stage_started = time.monotonic() - transcript = clean_subtitle(subtitle_path) - mark_stage(processing, "clean_subtitle", stage_started) - if not transcript.strip(): - print("字幕文件为空，将尝试音频转写。", file=sys.stderr) - warnings.append("字幕文件为空，已回退到 Whisper 转写。") - subtitle_path = None - - if not is_local_file and not subtitle_path: - stage_started = time.monotonic() - audio_path = download_audio(args.url, output_dir, cookies) - mark_stage(processing, "download_audio", stage_started) - stage_started = time.monotonic() + stage_started = time.monotonic() + transcript = clean_subtitle(subtitle_path) + mark_stage(processing, "clean_subtitle", stage_started) + if not transcript.strip(): + print("字幕文件为空，将尝试音频转写。", file=sys.stderr) + warnings.append("字幕文件为空，已回退到 Whisper 转写。") + transcript = None + subtitle_path = None + + audio_path: Path | None = None + if transcript is None: whisper_context_terms = build_whisper_context_terms(meta) - transcript = transcribe_audio( - audio_path, - args.model_size, - whisper_language, - build_whisper_initial_prompt(whisper_context_terms), - ) - mark_stage(processing, "transcribe", stage_started) + transcription_input = local_input + if not is_local_file: + stage_started = time.monotonic() + if vision_video_path: + audio_path = extract_audio_from_video(vision_video_path, output_dir) + mark_stage(processing, "extract_audio", stage_started) + else: + audio_path = download_audio(args.url, output_dir, cookies) + mark_stage(processing, "download_audio", stage_started) + transcription_input = audio_path + + def transcribe_job() -> tuple[str, float]: + started = time.monotonic() + text = transcribe_audio( + transcription_input, + args.model_size, + whisper_language, + build_whisper_initial_prompt(whisper_context_terms or []), + ) + return text, time.monotonic() - started + + def keyframe_job() -> tuple[list[FrameCandidate], float]: + if not vision_video_path or not vision_sampling_plan: + raise AppError("vision_not_ready", "视觉视频尚未准备完成。") + started = time.monotonic() + frames = extract_keyframes( + vision_video_path, + output_dir, + vision_sampling_plan.scan_interval, + vision_sampling_plan.max_gap, + vision_sampling_plan.max_frames, + args.vision_frame_width, + ) + return frames, time.monotonic() - started + + if args.with_vision and keyframes is None and vision_video_path: + processing["parallel_stages"] = ["transcribe", "extract_keyframes"] + with ThreadPoolExecutor(max_workers=2, thread_name_prefix="video-summary") as executor: + transcript_future = executor.submit(transcribe_job) + keyframe_future = executor.submit(keyframe_job) + transcript, transcribe_seconds = transcript_future.result() + keyframes, keyframe_seconds = keyframe_future.result() + processing["stages"]["transcribe"] = round(transcribe_seconds, 3) + processing["stages"]["extract_keyframes"] = round(keyframe_seconds, 3) + else: + transcript, transcribe_seconds = transcribe_job() + processing["stages"]["transcribe"] = round(transcribe_seconds, 3) transcript_source = "whisper" - if not args.keep_audio: + if audio_path and not args.keep_audio: audio_path.unlink(missing_ok=True) - transcript_path = output_dir / "transcript.txt" - transcript_path.write_text(transcript, encoding="utf-8") + if transcript is None: + raise AppError("transcript_empty", "字幕和 Whisper 均未生成可用文本。") + if not transcript_path.is_file() or "transcript" not in processing.get("cache_hits", []): + transcript_path.write_text(transcript, encoding="utf-8") + if transcript_cache_signature: + update_cache_stage( + pipeline_cache, + "transcript", + transcript_cache_signature, + path="transcript.txt", + source=transcript_source, + whisper_context_terms=whisper_context_terms, + selected_subtitle_lang=meta.selected_subtitle_lang, + selected_subtitle_type=meta.selected_subtitle_type, + ) + + if args.with_vision and keyframes is None: + if not vision_video_path or not vision_sampling_plan: + raise AppError("vision_not_ready", "视觉视频尚未准备完成。") + stage_started = time.monotonic() + keyframes = extract_keyframes( + vision_video_path, + output_dir, + vision_sampling_plan.scan_interval, + vision_sampling_plan.max_gap, + vision_sampling_plan.max_frames, + args.vision_frame_width, + ) + mark_stage(processing, "extract_keyframes", stage_started) + + if args.with_vision and keyframes and keyframe_cache_signature: + update_cache_stage( + pipeline_cache, + "keyframes", + keyframe_cache_signature, + frames=serialize_keyframes(keyframes, output_dir), + ) + save_pipeline_cache(output_dir, pipeline_cache) write_meta(meta, output_dir, transcript_source, warnings, processing, options, whisper_context_terms) + summary_input = transcript + summary_source = transcript_source + summary_prompt_template = prompt_template + if args.with_vision: + if not vision_config or not vision_client or not keyframes: + raise AppError("vision_not_ready", "视觉解析环境没有正确初始化。") + processing["vision"]["frames"] = len(keyframes) + processing["vision"]["status"] = "analyzing" + + if downloaded_vision_video: + try: + vision_video_path.unlink(missing_ok=True) + downloaded_vision_video = False + except OSError as exc: + warnings.append(f"临时视频清理失败：{exc}") + + visual_context_path = output_dir / "visual_context.json" + existing_visual_results = load_visual_results(output_dir, vision_config.model) + visual_cache_complete = bool( + args.resume + and vision_cache_signature + and cache_stage_matches(pipeline_cache, "vision", vision_cache_signature) + and visual_results_cover_frames(existing_visual_results, keyframes, output_dir) + ) + if visual_cache_complete: + visual_frames = existing_visual_results + processing.setdefault("cache_hits", []).append("vision") + else: + stage_started = time.monotonic() + visual_frames = analyze_keyframes( + vision_client, + vision_config, + keyframes, + output_dir, + args.vision_batch_size, + existing_results=existing_visual_results if args.resume else None, + transcript=transcript, + context_terms=build_whisper_context_terms(meta), + ) + mark_stage(processing, "analyze_keyframes", stage_started) + if vision_cache_signature: + update_cache_stage( + pipeline_cache, + "vision", + vision_cache_signature, + path="visual_context.json", + frame_count=len(visual_frames), + ) + save_pipeline_cache(output_dir, pipeline_cache) + + visual_summary_limit = ( + determine_summary_chunk_chars(summary_config, args.max_chars) + if summary_config + else args.max_chars + ) + summary_input = build_multimodal_context( + transcript, + visual_frames, + max_chars=visual_summary_limit, + ) + multimodal_context_path = output_dir / "multimodal_context.txt" + multimodal_context_path.write_text(summary_input, encoding="utf-8") + summary_source = f"{transcript_source}+vision" + if not args.prompt and not args.prompt_file: + summary_prompt_template = MULTIMODAL_PROMPT_TEMPLATE + processing["vision"].update( + { + "status": "completed", + "frames": len(visual_frames), + "frame_sources": dict(Counter(frame.source for frame in keyframes)), + "context_frames": summary_input.count("] 画面："), + "compacted_frames": len(compact_visual_material(visual_frames, transcript)), + "visual_char_budget": max(0, visual_summary_limit - len(transcript)), + "multimodal_chars": len(summary_input), + "analysis_metrics": load_visual_metrics(output_dir), + "visual_context": str(visual_context_path), + "multimodal_context": str(multimodal_context_path), + } + ) + processing["summary_source"] = summary_source + write_meta(meta, output_dir, transcript_source, warnings, processing, options, whisper_context_terms) + if args.export_prompt: stage_started = time.monotonic() prompt_path = write_chatgpt_prompt(transcript, meta, transcript_source, output_dir, args.max_chars, prompt_template) @@ -1054,38 +2656,132 @@ def main() -> int: print(f"已生成转写稿：{transcript_path}") return 0 - stage_started = time.monotonic() - summary = summarize_with_llm(transcript, meta, transcript_source, args.max_chars, prompt_template) - mark_stage(processing, "summarize", stage_started) summary_path = output_dir / "summary.md" - summary_path.write_text(summary.strip() + "\n", encoding="utf-8") + if not summary_config or not summary_client or summary_input is None or summary_source is None: + raise AppError("summary_not_ready", "最终总结模型没有正确初始化。") + summary_cache_signature = stable_signature( + { + "input": hashlib.sha256(summary_input.encode("utf-8")).hexdigest(), + "provider": summary_config.provider, + "base_url": summary_config.base_url, + "model": summary_config.model, + "max_chars": args.max_chars, + "provider_chunk_chars": summary_config.chunk_chars, + "context_length": summary_config.context_length, + "summary_schema": 10, + "prompt": hashlib.sha256(summary_prompt_template.encode("utf-8")).hexdigest(), + } + ) + summary_result: SummaryResult | None = None + if ( + args.resume + and cache_stage_matches(pipeline_cache, "summary", summary_cache_signature) + and summary_path.is_file() + ): + processing.setdefault("cache_hits", []).append("summary") + else: + stage_started = time.monotonic() + summary_result = summarize_with_llm( + summary_client, + summary_config, + summary_input, + meta, + summary_source, + args.max_chars, + summary_prompt_template, + ) + mark_stage(processing, "summarize", stage_started) + summary_path.write_text(summary_result.text.strip() + "\n", encoding="utf-8") + update_cache_stage( + pipeline_cache, + "summary", + summary_cache_signature, + path="summary.md", + provider=summary_config.provider, + model=summary_config.model, + chunk_chars=summary_result.chunk_chars, + chunks=summary_result.chunks, + chunk_lengths=summary_result.chunk_lengths, + context_length=summary_result.context_length, + fallback_used=summary_result.fallback_used, + ) + save_pipeline_cache(output_dir, pipeline_cache) + processing["status"] = "completed" + summary_stage = pipeline_cache.get("stages", {}).get("summary", {}) + effective_summary_chars = ( + summary_result.chunk_chars + if summary_result + else summary_stage.get("chunk_chars") or determine_summary_chunk_chars(summary_config, args.max_chars) + ) + summary_chunks = ( + summary_result.chunks + if summary_result + else summary_stage.get("chunks") or len(split_text_balanced(summary_input, effective_summary_chars)) + ) + processing["summary"] = { + "provider": summary_config.provider, + "model": summary_config.model, + "status": "completed", + "strategy": "configured" if summary_config.chunk_chars else "ollama_context_auto", + "chunk_chars": effective_summary_chars, + "chunks": summary_chunks, + "chunk_lengths": ( + summary_result.chunk_lengths + if summary_result + else summary_stage.get("chunk_lengths") + or [len(chunk) for chunk in split_text_balanced(summary_input, effective_summary_chars)] + ), + "context_length": summary_config.context_length, + "model_context_length": summary_config.model_context_length, + "fallback_used": ( + summary_result.fallback_used if summary_result else bool(summary_stage.get("fallback_used")) + ), + } mark_total(processing, run_started) write_meta(meta, output_dir, transcript_source, warnings, processing, options, whisper_context_terms) if args.json: - print_json( - { - "ok": True, - "output_dir": str(output_dir), - "transcript": str(transcript_path), - "summary": str(summary_path), - "transcript_source": transcript_source, - "warnings": warnings, - } - ) + payload = { + "ok": True, + "output_dir": str(output_dir), + "transcript": str(transcript_path), + "summary": str(summary_path), + "transcript_source": transcript_source, + "warnings": warnings, + } + if visual_context_path and multimodal_context_path: + payload["visual_context"] = str(visual_context_path) + payload["multimodal_context"] = str(multimodal_context_path) + print_json(payload) else: print(f"已生成总结：{summary_path}") return 0 except AppError as exc: - if output_dir and transcript_path and transcript_path.exists() and not prompt_path: + processing["status"] = "failed" + processing["error"] = {"code": exc.code, "message": exc.message} + if ( + args.with_vision + and processing.get("vision") + and processing["vision"].get("status") != "completed" + ): + processing["vision"]["status"] = "failed" + if summary_config: + processing["summary"] = { + "provider": summary_config.provider, + "model": summary_config.model, + "status": "failed", + "error": exc.message, + } + can_export_fallback = not args.with_vision or multimodal_context_path is not None + if output_dir and transcript_path and transcript_path.exists() and not prompt_path and can_export_fallback: try: stage_started = time.monotonic() prompt_path = write_chatgpt_prompt( - transcript_path.read_text(encoding="utf-8"), + summary_input or transcript_path.read_text(encoding="utf-8"), meta or VideoMeta(title="video", id="video", webpage_url=""), - transcript_source or "unknown", + summary_source or transcript_source or "unknown", output_dir, args.max_chars, - prompt_template, + summary_prompt_template, ) mark_stage(processing, "export_prompt", stage_started) exc.hints.append(f"已自动生成 ChatGPT 提示词，可复制到 ChatGPT：{prompt_path}") @@ -1098,36 +2794,54 @@ def main() -> int: except Exception: pass if args.json: - print_json( - { - "ok": False, - "error": { - "code": exc.code, - "message": exc.message, - "hints": exc.hints, - }, - "output_dir": str(output_dir) if output_dir else None, - "transcript": str(transcript_path) if transcript_path else None, - "prompt": str(prompt_path) if prompt_path else None, - "warnings": warnings, - } - ) + payload = { + "ok": False, + "error": { + "code": exc.code, + "message": exc.message, + "hints": exc.hints, + }, + "output_dir": str(output_dir) if output_dir else None, + "transcript": str(transcript_path) if transcript_path else None, + "prompt": str(prompt_path) if prompt_path else None, + "warnings": warnings, + } + if args.with_vision: + payload["visual_context"] = str(visual_context_path) if visual_context_path else None + payload["multimodal_context"] = str(multimodal_context_path) if multimodal_context_path else None + print_json(payload) else: print(f"错误：{exc.message}", file=sys.stderr) for hint in exc.hints: print(f"提示：{hint}", file=sys.stderr) return 1 except Exception as exc: - if output_dir and transcript_path and transcript_path.exists() and not prompt_path: + processing["status"] = "failed" + processing["error"] = {"code": "unexpected_error", "message": str(exc)} + if ( + args.with_vision + and processing.get("vision") + and processing["vision"].get("status") != "completed" + ): + processing["vision"]["status"] = "failed" + if summary_config: + processing["summary"] = { + "provider": summary_config.provider, + "model": summary_config.model, + "status": "failed", + "error": str(exc), + } + can_export_fallback = not args.with_vision or multimodal_context_path is not None + if output_dir and transcript_path and transcript_path.exists() and not prompt_path and can_export_fallback: try: stage_started = time.monotonic() prompt_path = write_chatgpt_prompt( - transcript_path.read_text(encoding="utf-8"), + summary_input or transcript_path.read_text(encoding="utf-8"), meta or VideoMeta(title="video", id="video", webpage_url=""), - transcript_source or "unknown", + summary_source or transcript_source or "unknown", output_dir, args.max_chars, - prompt_template, + summary_prompt_template, ) mark_stage(processing, "export_prompt", stage_started) except Exception: @@ -1139,20 +2853,22 @@ def main() -> int: except Exception: pass if args.json: - print_json( - { - "ok": False, - "error": { - "code": "unexpected_error", - "message": str(exc), - "hints": [], - }, - "output_dir": str(output_dir) if output_dir else None, - "transcript": str(transcript_path) if transcript_path else None, - "prompt": str(prompt_path) if prompt_path else None, - "warnings": warnings, - } - ) + payload = { + "ok": False, + "error": { + "code": "unexpected_error", + "message": str(exc), + "hints": [], + }, + "output_dir": str(output_dir) if output_dir else None, + "transcript": str(transcript_path) if transcript_path else None, + "prompt": str(prompt_path) if prompt_path else None, + "warnings": warnings, + } + if args.with_vision: + payload["visual_context"] = str(visual_context_path) if visual_context_path else None + payload["multimodal_context"] = str(multimodal_context_path) if multimodal_context_path else None + print_json(payload) else: print(f"错误：{exc}", file=sys.stderr) if prompt_path: