skill-video-lapian/scripts/transcribe.py

#!/usr/bin/env python3
"""transcribe.py — 提取音频并用 fmode-listen 转写(讯飞 LFASR, 分段时间戳).

用法:
  python3 transcribe.py --workdir /tmp/lapian
前置: workdir/video.mp4 (fetch_video.py 产物)
产物: workdir/audio.wav + workdir/transcript.json
  transcript.json = {code, data:{text, segments:[{bg,ed,speaker,text}]}}, 毫秒
环境: FEME_NEWAPI_TOKEN(或 FMODE_API_KEY) — 传给 fmode-listen 的 FMODE_API_TOKEN
"""
import argparse
import json
import os
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import get_api_token  # noqa: E402


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--workdir", required=True)
    ap.add_argument("--language", default="autodialect",
                    help="autodialect(中英+方言自动)/zh/en, 默认 autodialect")
    args = ap.parse_args()

    video = os.path.join(args.workdir, "video.mp4")
    wav = os.path.join(args.workdir, "audio.wav")
    out_json = os.path.join(args.workdir, "transcript.json")
    if not os.path.exists(video):
        raise SystemExit(f"[transcribe] 缺少 {video}, 先跑 fetch_video.py")

    if not (os.path.exists(wav) and os.path.getsize(wav) > 1000):
        print("[transcribe] 提取音频: 16kHz 单声道 wav")
        r = subprocess.run(["ffmpeg", "-y", "-i", video, "-vn", "-ar", "16000",
                            "-ac", "1", "-f", "wav", wav],
                           capture_output=True, text=True)
        if r.returncode != 0:
            raise SystemExit(f"[transcribe] ffmpeg 失败: {r.stderr[-500:]}")
    else:
        print("[transcribe] 复用已提取的 audio.wav")

    # 短路: 已有转写直接复用(转写计费, 别重复花钱)
    if os.path.exists(out_json):
        try:
            old = json.load(open(out_json, encoding="utf-8"))
            segs = (old.get("data") or {}).get("segments") or []
            if segs:
                print(f"[transcribe] 复用已有转写({len(segs)}段), 跳过计费")
                return
        except Exception:
            pass

    env = dict(os.environ, FMODE_API_TOKEN=get_api_token())
    cmd = ["npx", "--yes", "fmode-listen@latest", "transcribe", "--", wav,
           "--language", args.language, "--out", out_json]
    print("[transcribe]", " ".join(cmd))
    r = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=1800)
    print(r.stdout[-800:])
    if r.returncode != 0:
        raise SystemExit(f"[transcribe] fmode-listen 失败: {r.stderr[-800:]}")
    if not os.path.exists(out_json):
        raise SystemExit("[transcribe] 转写产物缺失")

    data = json.load(open(out_json, encoding="utf-8"))
    segs = (data.get("data") or {}).get("segments") or []
    print(f"[transcribe] 完成: {len(segs)} 段, 全文 {len((data.get('data') or {}).get('text',''))} 字")


if __name__ == "__main__":
    main()