diff --git a/.gitignore b/.gitignore index fef83f4..db2e4bc 100644 --- a/.gitignore +++ b/.gitignore @@ -13,39 +13,51 @@ dist/ build/ .vite/ -# 环境配置(永远不上传) +# 环境配置(永远不上传) .env .env.local -# 编辑器 +# 编辑器 .vscode/ .idea/ *.swp -# 操作系统 +# 操作系统 .DS_Store Thumbs.db -# 日志运行产物 +# 日志运行产物 *.log cookie.txt cookies*.txt -# 临时文件 +# 临时文件 *.tmp *.bak homePC/ -# Syncthing (极空间 NAS 同步工具产物) +# Syncthing (本机 NAS 同步) .stignore -# ĘýžÝżâą¸ˇÝ / SQL dump(˛ť˝ř°ćąžżâ) +# 备份 / SQL dump backups/ -# 敏感脚本(永不上传) +# 敏感脚本(永不提交) backend/scripts/reset_password.py - -# ǰ˲/ռλͼƬ(ͼϴԲ) -frontend/src/assets/*.jpg -frontend/src/assets/*.jpeg -frontend/src/assets/*.png + +# 前端素材 +frontend/src/assets/*.jpg +frontend/src/assets/*.jpeg +frontend/src/assets/*.png + +# Doco 素材目录(视频不进 git) +programs/*/source/video.* +programs/*/source/*.mp4 +programs/*/source/*.mov +programs/*/work/ + +# Doco .env +doco/.env + +# Python build artifacts +*.egg-info/ diff --git a/doco/.env.example b/doco/.env.example new file mode 100644 index 0000000..44cfec2 --- /dev/null +++ b/doco/.env.example @@ -0,0 +1,26 @@ +# Doco 子模块凭证配置 +# 复制此文件为 .env 并填入真实凭证 +# 格式: VARIABLE_NAME=your_value + +# ======================================================================== +# 讯飞 ASR (录音文件转写标准版) +# 申请地址: https://console.xfyun.cn/ +# 注意: 不要用"大模型版",用"录音文件转写标准版" +# 凭证类型: APP_ID + SECRET_KEY +# ======================================================================== +XFYUN_APP_ID=your_xfyun_app_id +XFYUN_SECRET_KEY=your_xfyun_secret_key + +# ======================================================================== +# DeepSeek Vision (OCR 用) +# 申请地址: https://platform.deepseek.com/ +# 凭证类型: API_KEY +# ======================================================================== +DEEPSEEK_API_KEY=your_deepseek_api_key + +# ======================================================================== +# Anthropic Claude API (AI 融合层,P3 才用) +# 申请地址: https://console.anthropic.com/ +# 凭证类型: API_KEY +# ======================================================================== +ANTHROPIC_API_KEY=your_anthropic_api_key diff --git a/doco/README.md b/doco/README.md new file mode 100644 index 0000000..51348a5 --- /dev/null +++ b/doco/README.md @@ -0,0 +1,116 @@ +# Doco - TPS 中台终版文稿生成子模块 + +> 央视《军事科技》栏目 - 终版文稿自动生成流水线 + +## 项目状态 + +**当前 Phase: P1** - 视频双路拆分预处理 + +## 功能概述 + +Doco 将一期《军事科技》节目视频拆分为两路输入,供下游三方融合(P3)使用: + +| 输出 | 规格 | 存放位置 | +|---|---|---| +| B 稿 | 带时间戳的 txt,`[Nm Ns] 句子`格式 | `work/b_manuscript.txt` | +| 音频 | 16kHz / 单声道 / 16bit WAV | `work/audio_16k.wav` | +| 关键帧索引 | JSON | `work/keyframes.json` | + +## 系统依赖 + +- **Python >= 3.12** +- **ffmpeg >= 4.x** (必须安装并加入 PATH) + - Windows 下载: https://ffmpeg.org/download.html + - macOS: `brew install ffmpeg` + - Linux: `apt install ffmpeg` + +## 安装 + +```bash +# 1. 克隆仓库后进入 doco 目录 +cd doco + +# 2. 安装依赖 +pip install -e . + +# 3. 配置凭证 +cp .env.example .env +# 编辑 .env,填入三组 API 凭证 +``` + +## 凭证配置 + +Doco 使用三组独立凭证,互不混用: + +| 服务 | 用途 | 申请地址 | +|---|---|---| +| 讯飞开放平台 - 录音文件转写(标准版) | 音频转文字 | https://console.xfyun.cn/ | +| DeepSeek Vision | OCR 识别 | https://platform.deepseek.com/ | +| Anthropic Claude API | AI 融合层(P3) | https://console.anthropic.com/ | + +> 注意: 讯飞要用"录音文件转写标准版",不要用"大模型版" + +## P1 使用方法 + +### 准备素材目录 + +``` +programs/ +└── ep001_20260612_fangkong_fandao/ + └── source/ + └── video.mp4 ← 放入节目视频 +``` + +> video.mp4 由制片人放入,不放进 git + +### 运行拆分 + +```bash +doco split \ + --episode-id ep001_20260612_fangkong_fandao \ + --input-video programs/ep001_20260612_fangkong_fandao/source/video.mp4 \ + --output-dir programs/ep001_20260612_fangkong_fandao/work/ +``` + +### 输出产物 + +``` +programs/ep001_20260612_fangkong_fandao/work/ +├── frames/ # 抽出的所有帧(临时) +├── audio_16k.wav # 音频(16kHz/单声道/16bit) +├── b_manuscript.txt # B 稿([Nm Ns] 句子格式) +└── keyframes.json # 关键帧索引 +``` + +## P1 验收标准 + +1. `work/b_manuscript.txt` 格式为 `[Nm Ns] 句子`,每行一句 +2. `work/audio_16k.wav` 规格为 16kHz/单声道/16bit,能被讯飞 ASR 接收 +3. `work/keyframes.json` 字段符合定义 + +## 目录结构 + +``` +doco/ +├── src/ +│ ├── __init__.py +│ ├── cli.py # CLI 入口 +│ ├── video_split.py # P1 核心:视频双路拆分 +│ ├── asr_adapter.py # 讯飞 ASR 适配层 +│ └── ocr_adapter.py # P2:DeepSeek Vision OCR +├── tests/ +│ ├── test_video_split.py # 单元测试 +│ └── fixtures/ +│ └── mini_test.mp4 # 迷你测试视频 +├── docs/ +├── .env.example # 凭证模板 +├── README.md +└── pyproject.toml +``` + +## 相关文档 + +- Brief: `docs/doco/Doco子项目_Brief.md` +- 设计文档: `docs/doco/doco_project_design.md` +- 讯飞接入笔记: `docs/doco/doco_xfyun_integration_notes.md` +- 主项目回复: `docs/doco/主project对Doco_PRDv2的回复.md` diff --git a/doco/pyproject.toml b/doco/pyproject.toml new file mode 100644 index 0000000..7fe92d7 --- /dev/null +++ b/doco/pyproject.toml @@ -0,0 +1,36 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "doco" +version = "0.1.0" +description = "TPS 中台 - 终版文稿生成子模块(视频双路拆分 + 三方融合)" +readme = "README.md" +requires-python = ">=3.12" +license = { text = "Proprietary" } +authors = [ + { name = "刘统制片组" } +] + +dependencies = [ + "Pillow>=10.0.0", + "imagehash>=4.3.1", + "requests>=2.31.0", + "python-dotenv>=1.0.0", + "click>=8.1.0", + "python-docx>=1.1.0", + "anthropic>=0.18.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.4.0", +] + +[project.scripts] +doco = "doco.src.cli:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["doco.src*"] diff --git a/doco/src/__init__.py b/doco/src/__init__.py new file mode 100644 index 0000000..40fe39a --- /dev/null +++ b/doco/src/__init__.py @@ -0,0 +1,6 @@ +# -*- coding: utf-8 -*- +""" +doco - TPS 中台终版文稿生成子模块 +""" + +__version__ = "0.1.0" diff --git a/doco/src/asr_adapter.py b/doco/src/asr_adapter.py new file mode 100644 index 0000000..ef9b6e3 --- /dev/null +++ b/doco/src/asr_adapter.py @@ -0,0 +1,298 @@ +# -*- coding: utf-8 -*- +""" +讯飞 ASR 适配层 +================================================= +来源: demo 跑通的 xfyun_asr_standard.py +改动: 凭证从环境变量读取,不再硬编码 + +接口: https://raasr.xfyun.cn/v2/api/upload / getResult +签名: signa = base64(HmacSHA1(MD5(appid + ts), secretKey)) + +特性: +- 支持热词列表(hotWord),提升专业术语识别率 +- 支持军事领域参数(pd=mil) +- 支持顺滑+口语规整(输出更接近书面语) +- 默认语种 cn(中文普通话),免费包标配 + +凭证来源: 环境变量 +- XFYUN_APP_ID +- XFYUN_SECRET_KEY +""" + +import base64 +import hashlib +import hmac +import json +import os +import re +import time +import wave +from urllib.parse import quote +from typing import List, Tuple, Optional + +import requests + +# ======================================================================== +# 凭证(从环境变量读取) +# ======================================================================== + +APP_ID = os.environ.get("XFYUN_APP_ID", "").strip() +SECRET_KEY = os.environ.get("XFYUN_SECRET_KEY", "").strip() + + +# ======================================================================== +# 接口配置 +# ======================================================================== + +HOST = "https://raasr.xfyun.cn/v2/api" +UPLOAD_URL = HOST + "/upload" +RESULT_URL = HOST + "/getResult" + +# 业务参数 +LANGUAGE = "cn" # 中文普通话 +PD = "mil" # 军事领域优化 +ENG_SMOOTHPROC = "true" # 顺滑(去掉"嗯/那个") +ENG_COLLOQPROC = "true" # 口语规整 + +# 轮询配置 +POLL_INTERVAL_SECONDS = 30 +MAX_WAIT_MINUTES = 30 + + +# ======================================================================== +# 热词列表(每期节目调用前从 A 稿提取) +# ======================================================================== + +def get_hot_words() -> List[str]: + """获取热词列表,P2 实现时从 A 稿提取""" + return [] + + +# ======================================================================== +# 签名+工具 +# ======================================================================== + + +def make_signa(app_id: str, secret_key: str, ts: str) -> str: + """ + 讯飞老版签名:signa = base64(HmacSHA1(MD5(appid + ts), secretKey)) + """ + base_string = (app_id + ts).encode("utf-8") + md5_str = hashlib.md5(base_string).hexdigest() # 32位小写hex + mac = hmac.new( + secret_key.encode("utf-8"), + md5_str.encode("utf-8"), + digestmod=hashlib.sha1, + ) + signa = base64.b64encode(mac.digest()).decode("utf-8") + return signa + + +def get_audio_duration_ms(filepath: str) -> int: + """获取音频时长(毫秒)。WAV用内置,MP3用mutagen。""" + ext = os.path.splitext(filepath)[1].lower() + + if ext == ".wav": + with wave.open(filepath, "rb") as wf: + n_frames = wf.getnframes() + sample_rate = wf.getframerate() + duration_ms = int(round(n_frames / sample_rate * 1000)) + return duration_ms + + if ext == ".mp3": + try: + from mutagen.mp3 import MP3 + return int(MP3(filepath).info.length * 1000) + except ImportError: + return 0 + + raise ValueError(f"不支持的音频格式: {ext}") + + +# ======================================================================== +# 上传 +# ======================================================================== + + +def upload_audio( + filepath: str, + hot_words: Optional[List[str]] = None, +) -> str: + """上传音频,返回 orderId""" + if not os.path.exists(filepath): + raise FileNotFoundError(f"音频文件不存在: {filepath}") + + if not APP_ID or not SECRET_KEY: + raise ValueError("请先设置 XFYUN_APP_ID 和 XFYUN_SECRET_KEY 环境变量") + + file_size = os.path.getsize(filepath) + file_name = os.path.basename(filepath) + duration_ms = get_audio_duration_ms(filepath) + ts = str(int(time.time())) + + signa = make_signa(APP_ID, SECRET_KEY, ts) + + # 构建URL参数 + params = { + "appId": APP_ID, + "signa": signa, + "ts": ts, + "fileSize": str(file_size), + "fileName": file_name, + "duration": str(duration_ms), + "language": LANGUAGE, + "pd": PD, + "eng_smoothproc": ENG_SMOOTHPROC, + "eng_colloqproc": ENG_COLLOQPROC, + } + + # 热词,用 | 分隔 + if hot_words: + hot_word_str = "|".join(hot_words) + params["hotWord"] = hot_word_str + + url_parts = [f"{quote(k, safe='')}={quote(str(v), safe='')}" for k, v in params.items()] + url = f"{UPLOAD_URL}?{'&'.join(url_parts)}" + + headers = { + "Content-Type": "application/json", + } + + with open(filepath, "rb") as f: + audio_bytes = f.read() + + resp = requests.post(url, headers=headers, data=audio_bytes, timeout=300) + + data = resp.json() + if data.get("code") != "000000": + raise RuntimeError(f"上传失败: code={data.get('code')}, desc={data.get('descInfo')}") + + order_id = data["content"]["orderId"] + return order_id + + +# ======================================================================== +# 查询结果 +# ======================================================================== + + +def query_result(order_id: str) -> dict: + """单次查询""" + ts = str(int(time.time())) + signa = make_signa(APP_ID, SECRET_KEY, ts) + + params = { + "appId": APP_ID, + "signa": signa, + "ts": ts, + "orderId": order_id, + "resultType": "transfer", + } + url_parts = [f"{quote(k, safe='')}={quote(str(v), safe='')}" for k, v in params.items()] + url = f"{RESULT_URL}?{'&'.join(url_parts)}" + + resp = requests.post(url, timeout=30) + return resp.json() + + +def poll_until_done(order_id: str) -> dict: + """轮询直到完成""" + start_time = time.time() + while True: + elapsed = time.time() - start_time + if elapsed > MAX_WAIT_MINUTES * 60: + raise TimeoutError(f"超过 {MAX_WAIT_MINUTES} 分钟未完成") + + data = query_result(order_id) + order_info = data.get("content", {}).get("orderInfo", {}) + status = order_info.get("status") + fail_type = order_info.get("failType", 0) + + if status == 4: + return data + if status == -1: + raise RuntimeError(f"转写失败: failType={fail_type}, 数据: {data}") + + time.sleep(POLL_INTERVAL_SECONDS) + + +# ======================================================================== +# 结果解析 +# ======================================================================== + + +def parse_order_result(order_result_str: str) -> List[Tuple[int, int, str]]: + """ + 解析嵌套JSON,返回 [(sentence_start_ms, sentence_end_ms, text), ...] + """ + if not order_result_str: + return [] + + cleaned = re.sub(r"\\\\", r"\\", order_result_str) + outer = json.loads(cleaned) + + sentences = [] + for item in outer.get("lattice", []): + inner_str = item.get("json_1best", "") + if not inner_str: + continue + inner = json.loads(inner_str) + st = inner.get("st", {}) + bg = int(st.get("bg", 0)) + ed = int(st.get("ed", 0)) + + words = [] + for rt in st.get("rt", []): + for ws in rt.get("ws", []): + for cw in ws.get("cw", []): + w = cw.get("w", "").strip() + wp = cw.get("wp", "n") + if w and wp != "g": + words.append(w) + sentence = "".join(words).strip() + if sentence: + sentences.append((bg, ed, sentence)) + + return sentences + + +def format_timestamp(ms: int) -> str: + """毫秒转 [Nm Ns] 格式""" + total_sec = ms // 1000 + return f"{total_sec // 60}m{total_sec % 60}s" + + +def transcribe( + audio_path: str, + hot_words: Optional[List[str]] = None, +) -> List[Tuple[int, int, str]]: + """ + 完整转写流程:上传 → 轮询 → 解析 + 返回 [(start_ms, end_ms, text), ...] + """ + order_id = upload_audio(audio_path, hot_words=hot_words) + result_data = poll_until_done(order_id) + order_result_str = result_data["content"]["orderResult"] + return parse_order_result(order_result_str) + + +def write_asr_result( + sentences: List[Tuple[int, int, str]], + output_dir: str, +) -> Tuple[str, str]: + """ + 将 ASR 结果写入文件 + 返回 (timed_txt_path, raw_json_path) + """ + os.makedirs(output_dir, exist_ok=True) + + timed_lines = [f"[{format_timestamp(bg)}] {text}" for bg, _, text in sentences] + timed_path = os.path.join(output_dir, "asr_result_timed.txt") + with open(timed_path, "w", encoding="utf-8") as f: + f.write("\n".join(timed_lines)) + + raw_path = os.path.join(output_dir, "asr_result_raw.json") + with open(raw_path, "w", encoding="utf-8") as f: + f.write("{}") # 占位,P2 实现时填入原始返回 + + return timed_path, raw_path diff --git a/doco/src/cli.py b/doco/src/cli.py new file mode 100644 index 0000000..0a1057c --- /dev/null +++ b/doco/src/cli.py @@ -0,0 +1,104 @@ +# -*- coding: utf-8 -*- +""" +doco CLI 入口 +P1: doco split 子命令 +P3: doco process 子命令(带 --input-a-draft 和 --cleanup-level) +""" + +import click +import sys +from pathlib import Path + +# P1 相关 +from .video_split import split_video + + +@click.group() +@click.version_option(version="0.1.0") +def main(): + """TPS 中台 - 终版文稿生成工具""" + pass + + +@main.command("split") +@click.option( + "--episode-id", + required=True, + help="节目 ID,如 ep001_20260612_fangkong_fandao", +) +@click.option( + "--input-video", + required=True, + type=click.Path(exists=True), + help="输入视频文件路径", +) +@click.option( + "--output-dir", + required=True, + type=click.Path(), + help="输出目录(work/ 路径)", +) +@click.option( + "--phash-threshold", + default=8, + type=int, + help="pHash 海明距离阈值,用于检测字幕变化(默认 8)", +) +def split(episode_id: str, input_video: str, output_dir: str, phash_threshold: int): + """ + P1: 视频双路拆分 + - A 路:抽帧 + pHash 变化检测 + OCR → B 稿 txt + - B 路:提取音频(16kHz/单声道/16bit WAV) + """ + video_path = Path(input_video) + out_dir = Path(output_dir) + + click.echo(f"[doco split] episode_id={episode_id}") + click.echo(f"[doco split] input_video={video_path}") + click.echo(f"[doco split] output_dir={out_dir}") + click.echo(f"[doco split] phash_threshold={phash_threshold}") + + try: + result = split_video( + video_path=video_path, + output_dir=out_dir, + episode_id=episode_id, + phash_threshold=phash_threshold, + ) + click.echo(f"[ok] B 稿: {result['b_manuscript_path']}") + click.echo(f"[ok] 音频: {result['audio_path']}") + click.echo(f"[ok] 关键帧索引: {result['keyframes_path']}") + click.echo(f"[ok] 关键帧数量: {result['keyframe_count']}") + except Exception as e: + click.echo(f"[error] {e}", err=True) + sys.exit(1) + + +@main.command("process") +@click.option("--episode-id", required=True, help="节目 ID") +@click.option("--input-video", required=True, type=click.Path(exists=True), help="输入视频") +@click.option("--input-a-draft", required=True, type=click.Path(exists=True), help="A 稿 docx") +@click.option("--output-dir", required=True, type=click.Path(), help="输出目录") +@click.option( + "--cleanup-level", + default="medium", + type=click.Choice(["keep_all", "medium", "clean"]), + help="口语清理档位(默认 medium)", +) +def process( + episode_id: str, + input_video: str, + input_a_draft: str, + output_dir: str, + cleanup_level: str, +): + """ + P3: 三方融合全流程 + 需要 A 稿 + B 稿(本命令调用 split) + ASR 结果,融合输出终版 docx + 差异报告 + """ + click.echo("[doco process] P3 全流程暂未实现,请先使用 split 命令") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/doco/src/video_split.py b/doco/src/video_split.py new file mode 100644 index 0000000..bdd2adb --- /dev/null +++ b/doco/src/video_split.py @@ -0,0 +1,353 @@ +# -*- coding: utf-8 -*- +""" +视频双路拆分 - P1 核心模块 +================================================= +功能: + A 路:视频帧 → pHash 变化检测 → OCR → B 稿 txt + B 路:视频 → 16kHz/单声道/16bit WAV + +不引入 ffmpeg-python 等 wrapper,只用 subprocess 调系统 ffmpeg。 +""" + +import hashlib +import json +import os +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Dict, List, Tuple, Optional + +from PIL import Image +import imagehash + + +# ======================================================================== +# 凭证(从环境变量读取,供 OCR 调用 DeepSeek Vision) +# ======================================================================== + +DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "").strip() + + +# ======================================================================== +# FFmpeg 封装 +# ======================================================================== + + +def check_ffmpeg(): + """检查 ffmpeg 是否在 PATH 中""" + result = shutil.which("ffmpeg") + if result is None: + raise RuntimeError( + "ffmpeg 未找到,请先安装 ffmpeg 并加入 PATH。" + "下载地址: https://ffmpeg.org/download.html" + ) + return result + + +def extract_frames( + video_path: Path, + output_dir: Path, + fps: int = 1, +) -> List[Tuple[int, int, Path]]: + """ + 按固定 fps 抽帧 + 返回: [(frame_index, timestamp_ms, image_path), ...] + """ + check_ffmpeg() + + frames_dir = output_dir / "frames" + frames_dir.mkdir(parents=True, exist_ok=True) + + # ffmpeg 抽帧,格式 frame_%04d.png + frame_pattern = str(frames_dir / "frame_%04d.png") + cmd = [ + "ffmpeg", + "-i", str(video_path), + "-vf", f"fps={fps}", + "-q:v", "2", # JPEG 质量 + frame_pattern, + "-y", # 覆盖 + ] + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"ffmpeg 抽帧失败: {result.stderr}") + + # 收集抽出的帧 + frames = [] + for i, f in enumerate(sorted(frames_dir.glob("frame_*.png"))): + # 时间戳:第 i 帧就是 i 秒 + timestamp_ms = i * 1000 + frames.append((i, timestamp_ms, f)) + + return frames + + +def extract_audio( + video_path: Path, + output_path: Path, +) -> Path: + """ + 用 ffmpeg 提取音频,转为 16kHz/单声道/16bit WAV + """ + check_ffmpeg() + + cmd = [ + "ffmpeg", + "-i", str(video_path), + "-ac", "1", # 单声道 + "-ar", "16000", # 16kHz + "-sample_fmt", "s16", # 16bit + str(output_path), + "-y", + ] + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"ffmpeg 音频提取失败: {result.stderr}") + + return output_path + + +# ======================================================================== +# pHash 变化检测 +# ======================================================================== + + +def compute_phash(image_path: Path) -> str: + """计算图片的 pHash,返回 hex 字符串""" + img = Image.open(image_path) + ph = imagehash.phash(img) + return str(ph) + + +def find_keyframes( + frames: List[Tuple[int, int, Path]], + threshold: int = 8, +) -> List[Dict]: + """ + 基于 pHash 海明距离找出字幕变化的关键帧 + + 算法: + - 第一帧总是关键帧 + - 后续帧:如果与上一个关键帧的 pHash 海明距离 > threshold,则是新关键帧 + + threshold: 海明距离阈值,默认 8 + """ + if not frames: + return [] + + keyframes = [] + last_keyframe_phash = None + + for frame_index, timestamp_ms, image_path in frames: + phash = compute_phash(image_path) + + is_keyframe = False + if last_keyframe_phash is None: + # 第一帧总是关键帧 + is_keyframe = True + else: + # 计算海明距离 + hamming = hamming_distance(last_keyframe_phash, phash) + if hamming > threshold: + is_keyframe = True + + if is_keyframe: + keyframes.append({ + "frame_index": frame_index, + "timestamp_ms": timestamp_ms, + "frame_image_path": str(image_path), + "phash": phash, + "ocr_text": "", # P2 调用 DeepSeek Vision 填充 + }) + last_keyframe_phash = phash + + return keyframes + + +def hamming_distance(s1: str, s2: str) -> int: + """计算两个 hex pHash 字符串的海明距离""" + if len(s1) != len(s2): + # pHash 长度不一致,取较长字符串的长度作为海明距离上限 + return max(len(s1), len(s2)) + return sum(c1 != c2 for c1, c2 in zip(s1, s2)) + + +# ======================================================================== +# OCR 接口(P2 实现,目前返回占位) +# ======================================================================== + + +def ocr_frame(image_path: Path) -> str: + """ + 识别帧内文字,返回纯文本 + + P1: 返回占位文本 + P2: 调用 DeepSeek Vision API + """ + if not DEEPSEEK_API_KEY: + # 无 API Key,返回占位 + return f"[OCR待填充 frame={image_path.name}]" + + # P2 实现:调用 DeepSeek Vision + # TODO: P2 实现 + return f"[OCR待填充 frame={image_path.name}]" + + +def ocr_keyframes(keyframes: List[Dict]) -> List[Dict]: + """对关键帧列表逐一调用 OCR""" + result = [] + for kf in keyframes: + image_path = Path(kf["frame_image_path"]) + ocr_text = ocr_frame(image_path) + kf_copy = kf.copy() + kf_copy["ocr_text"] = ocr_text + result.append(kf_copy) + return result + + +# ======================================================================== +# B 稿格式化 +# ======================================================================== + + +def format_timestamp(ms: int) -> str: + """毫秒转 [Nm Ns] 格式""" + total_sec = ms // 1000 + return f"{total_sec // 60}m{total_sec % 60}s" + + +def build_b_manuscript(keyframes: List[Dict]) -> List[str]: + """ + 将关键帧 OCR 结果合并为 B 稿 + 合并相邻同文本的关键帧 + """ + lines = [] + last_text = None + + for kf in keyframes: + text = kf["ocr_text"].strip() + if not text: + continue + + # 跳过占位文本 + if text.startswith("[OCR待填充"): + text = "" + + if text and text != last_text: + ts = format_timestamp(kf["timestamp_ms"]) + lines.append(f"[{ts}] {text}") + last_text = text + + return lines + + +def write_b_manuscript(lines: List[str], output_path: Path) -> Path: + """写入 B 稿 txt""" + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w", encoding="utf-8") as f: + f.write("\n".join(lines)) + return output_path + + +# ======================================================================== +# 主流程 +# ======================================================================== + + +def split_video( + video_path: Path, + output_dir: Path, + episode_id: str, + phash_threshold: int = 8, + fps: int = 1, +) -> Dict[str, any]: + """ + 视频双路拆分主流程 + + 参数: + video_path: 输入视频路径 + output_dir: 输出目录(work/ 路径) + episode_id: 节目 ID + phash_threshold: pHash 海明距离阈值,默认 8 + fps: 抽帧帧率,默认 1(每秒一帧) + + 返回: + { + "b_manuscript_path": Path, + "audio_path": Path, + "keyframes_path": Path, + "keyframe_count": int, + } + """ + video_path = Path(video_path) + output_dir = Path(output_dir) + + if not video_path.exists(): + raise FileNotFoundError(f"视频文件不存在: {video_path}") + + # 创建输出目录 + output_dir.mkdir(parents=True, exist_ok=True) + frames_dir = output_dir / "frames" + frames_dir.mkdir(parents=True, exist_ok=True) + + print(f"[video_split] 开始处理: {video_path.name}") + print(f"[video_split] 抽帧 fps={fps}, pHash threshold={phash_threshold}") + + # ---- A 路:抽帧 + pHash 检测 + OCR ---- + print("[video_split] A路:抽帧...") + frames = extract_frames(video_path, output_dir, fps=fps) + print(f"[video_split] 抽帧完成,共 {len(frames)} 帧") + + print("[video_split] pHash 变化检测...") + keyframes = find_keyframes(frames, threshold=phash_threshold) + print(f"[video_split] 检测到 {len(keyframes)} 个关键帧") + + print("[video_split] OCR 关键帧...") + keyframes = ocr_keyframes(keyframes) + print(f"[video_split] OCR 完成") + + # ---- B 路:音频提取 ---- + print("[video_split] B路:提取音频...") + audio_path = output_dir / "audio_16k.wav" + extract_audio(video_path, audio_path) + print(f"[video_split] 音频提取完成: {audio_path}") + + # ---- 输出产物 ---- + # B 稿 + b_lines = build_b_manuscript(keyframes) + b_manuscript_path = output_dir / "b_manuscript.txt" + write_b_manuscript(b_lines, b_manuscript_path) + print(f"[video_split] B稿写入: {b_manuscript_path} ({len(b_lines)} 行)") + + # 关键帧索引 JSON + keyframes_data = { + "video_path": str(video_path), + "fps_sampled": fps, + "phash_threshold": phash_threshold, + "keyframes": keyframes, + } + keyframes_path = output_dir / "keyframes.json" + with open(keyframes_path, "w", encoding="utf-8") as f: + json.dump(keyframes_data, f, ensure_ascii=False, indent=2) + print(f"[video_split] 关键帧索引写入: {keyframes_path}") + + # 清理临时帧文件(可选,保留供调试) + # shutil.rmtree(frames_dir) + + return { + "b_manuscript_path": str(b_manuscript_path), + "audio_path": str(audio_path), + "keyframes_path": str(keyframes_path), + "keyframe_count": len(keyframes), + } diff --git a/doco/tests/test_video_split.py b/doco/tests/test_video_split.py new file mode 100644 index 0000000..872ee4f --- /dev/null +++ b/doco/tests/test_video_split.py @@ -0,0 +1,141 @@ +# -*- coding: utf-8 -*- +""" +video_split 单元测试 +""" + +import os +import json +from pathlib import Path + +import pytest + +# 确保 src 在 path 中 +import sys +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from doco.src.video_split import ( + hamming_distance, + format_timestamp, + build_b_manuscript, + compute_phash, +) + + +class TestHammingDistance: + def test_identical_strings(self): + assert hamming_distance("abc", "abc") == 0 + + def test_different_strings(self): + assert hamming_distance("abc", "abd") == 1 + + def test_different_length(self): + # 长度不同时,返回较长字符串的长度 + assert hamming_distance("abc", "ab") == 3 + + +class TestFormatTimestamp: + def test_zero(self): + assert format_timestamp(0) == "0m0s" + + def test_seconds_only(self): + assert format_timestamp(30000) == "0m30s" # 30秒 + + def test_minutes_and_seconds(self): + assert format_timestamp(90000) == "1m30s" # 1分30秒 + + def test_longer(self): + assert format_timestamp(3723000) == "62m3s" # 62分3秒 + + +class TestBuildBManuscript: + def test_empty_keyframes(self): + lines = build_b_manuscript([]) + assert lines == [] + + def test_single_frame(self): + keyframes = [ + {"timestamp_ms": 0, "ocr_text": "测试字幕"} + ] + lines = build_b_manuscript(keyframes) + assert len(lines) == 1 + assert "[0m0s]" in lines[0] + assert "测试字幕" in lines[0] + + def test_duplicate_text_merged(self): + """相邻同文本应合并""" + keyframes = [ + {"timestamp_ms": 0, "ocr_text": "相同"}, + {"timestamp_ms": 1000, "ocr_text": "相同"}, + {"timestamp_ms": 2000, "ocr_text": "不同"}, + ] + lines = build_b_manuscript(keyframes) + assert len(lines) == 2 + + def test_placeholder_skipped(self): + """OCR占位文本应跳过""" + keyframes = [ + {"timestamp_ms": 0, "ocr_text": "[OCR待填充 frame=001.png]"}, + {"timestamp_ms": 1000, "ocr_text": "真实字幕"}, + ] + lines = build_b_manuscript(keyframes) + assert len(lines) == 1 + assert "真实字幕" in lines[0] + + +class TestKeyframesJson: + """验证 keyframes.json 输出格式""" + + def test_keyframe_structure(self, tmp_path): + """验证单个关键帧的 JSON 结构""" + # 模拟关键帧数据 + kf = { + "frame_index": 1, + "timestamp_ms": 1000, + "frame_image_path": str(tmp_path / "frame_0001.png"), + "phash": "ff00aabb12345678", + "ocr_text": "测试", + } + + # 验证字段存在 + assert "frame_index" in kf + assert "timestamp_ms" in kf + assert "frame_image_path" in kf + assert "phash" in kf + assert "ocr_text" in kf + + def test_keyframes_json_output(self, tmp_path): + """验证完整 keyframes.json 输出""" + frames_dir = tmp_path / "frames" + frames_dir.mkdir() + + # 创建一个测试图片 + test_img = frames_dir / "frame_0001.png" + test_img.write_bytes(b"fake_png_data") + + keyframes_data = { + "video_path": str(tmp_path / "video.mp4"), + "fps_sampled": 1, + "phash_threshold": 8, + "keyframes": [ + { + "frame_index": 0, + "timestamp_ms": 0, + "frame_image_path": str(test_img), + "phash": "abcd1234", + "ocr_text": "首帧", + } + ] + } + + json_path = tmp_path / "keyframes.json" + with open(json_path, "w", encoding="utf-8") as f: + json.dump(keyframes_data, f, ensure_ascii=False, indent=2) + + # 验证可读 + with open(json_path, "r", encoding="utf-8") as f: + loaded = json.load(f) + + assert loaded["fps_sampled"] == 1 + assert loaded["phash_threshold"] == 8 + assert len(loaded["keyframes"]) == 1 + assert loaded["keyframes"][0]["frame_index"] == 0 diff --git a/docs/api_credentials_inventory.md b/docs/api_credentials_inventory.md new file mode 100644 index 0000000..b9e6f0f --- /dev/null +++ b/docs/api_credentials_inventory.md @@ -0,0 +1,44 @@ +# API 凭证清单 + +> TPS 中台所有外部 API 凭证的元信息登记 +> **不存储真实凭证**,真实凭证在各自子模块的 `.env` 中 +> 由制片人维护,key 到期或更换时更新此文档 + +--- + +## 字段说明 + +| 字段 | 含义 | +|---|---| +| 子模块 | 凭证所属的子模块 | +| API 服务 | 具体的 API 服务名称 | +| Key 类型 | APP_ID+SECRET_KEY / API_KEY / 其他 | +| 开通日 | 凭证申请日期 | +| 激活状态 | 是否已激活,额度信息 | +| 到期日 | 有效期,永久有效则填 — | +| 责任人 | 谁负责管理/续费 | +| 备注 | 其他说明 | + +--- + +## Doco 子模块 + +| 子模块 | API 服务 | Key 类型 | 开通日 | 激活状态 | 到期日 | 责任人 | 备注 | +|---|---|---|---|---|---|---|---| +| doco | 讯飞开放平台 - 录音文件转写(标准版) | APP_ID + SECRET_KEY | 2026-06-12 | 待激活(需走 0 元购买) | 2027-06-12 | 制片人 | demo 凭证已过期,需新申请 | +| doco | DeepSeek Vision | API_KEY | 2026-06-12 | 已激活 | — | 制片人 | doco OCR 用 | +| doco | Anthropic Claude API | API_KEY | 2026-06-12 | 已激活 | — | 制片人 | AI 融合层(P3) | + +--- + +## 主项目 + +| 子模块 | API 服务 | Key 类型 | 开通日 | 激活状态 | 到期日 | 责任人 | 备注 | +|---|---|---|---|---|---|---|---| +| (主) | Anthropic Claude API | API_KEY | 2026-05-14 | 已激活 | — | 制片人 | Cline Plan/Act 模型用 | +| (主) | MiniMax M2.7 | API_KEY | 2026-05-14 | 已激活 | — | 制片人 | Cline Plan/Act 模型用 | +| (主) | DeepSeek API | API_KEY | 2026-05-14 | 已激活 | — | 制片人 | embedding 服务 | + +--- + +*最后更新: 2026-06-12*