Funasr Transcribe

{
  "models": [
    {
      "id": "iic/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
      "name": "ASR 主模型 (Paraformer)",
      "description": "语音识别核心模型，支持中文",
      "required": true
    },
    {
      "id": "iic/speech_fsmn_vad_zh-cn-16k-common-pytorch",
      "name": "VAD 模型",
      "description": "语音活动检测，用于分段",
      "required": true
    },
    {
      "id": "iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch",
      "name": "标点模型",
      "description": "自动恢复标点符号",
      "required": true
    },
    {
      "id": "damo/speech_campplus_speaker-diarization_common",
      "name": "说话人分离模型 (CAM++)",
      "description": "识别不同说话人",
      "required": false
    },
    {
      "id": "iic/SenseVoiceSmall",
      "name": "SenseVoice-Small（单人快速模式）",
      "description": "单人讲课/语音笔记的快速模型，首次使用 ONNX 模式时按需下载",
      "required": false
    }
  ],
  "default_model": "iic/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch"
}

# Core dependencies
fastapi>=0.100.0
uvicorn[standard]>=0.20.0
funasr>=1.3.0
funasr-onnx>=0.4.1
modelscope>=1.33.0
torch>=2.0.0
torchaudio>=2.0.0
httpx>=0.20.0

# Speaker diarization requires scikit-learn (indirect dependency)
# Must pin numpy<2 to avoid binary incompatibility with older pandas/sklearn builds
numpy>=1.20,<2
pandas>=1.3,<3
scikit-learn>=1.0,<2

# Video slide extraction
scenedetect[opencv]>=0.6.4
imagehash>=4.3.1

{
  "env": {
    "PATH": "/Users/maoking/.nvm/versions/node/v24.11.0/bin:/Users/maoking/.nvm/versions/node/v24.11.0/bin:/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/System/Cryptexes/App/usr/bin:/usr/bin:/bin:/usr/sbin:/sbin:/var/run/com.apple.security.cryptexd/codex.system/bootstrap/usr/local/bin:/var/run/com.apple.security.cryptexd/codex.system/bootstrap/usr/bin:/var/run/com.apple.security.cryptexd/codex.system/bootstrap/usr/appleinternal/bin://Applications/Topaz Photo AI.app/Contents/Resources/bin:/Library/Apple/usr/bin:/Applications/Little Snitch.app/Contents/Components:/usr/local/share/dotnet:~/.dotnet/tools:/Users/maoking/.local/bin:/opt/homebrew/opt/python@3.14/Frameworks/Python.framework/Versions/3.14/bin:/Users/maoking/.bun/bin:/Users/maoking/.pixi/bin:/Users/maoking/.opencode/bin:/Users/maoking/.antigravity/antigravity/bin:/Users/maoking/Library/Python/3.11/bin:/Users/maoking/.nvm/versions/node/v24.11.0/bin:/Users/maoking/.cargo/bin:/Users/maoking/.orbstack/bin:/Users/maoking/.cache/lm-studio/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/agent-sdk-dev/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/thedotmack/claude-mem/9.1.1/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/code-simplifier/1.0.0/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/context7/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/frontend-design/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/playground/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/playwright/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/ralph-loop/1.0.0/bin:/Users/maoking/.claude/plugins/cache/claude-plugins-official/skill-creator/8e9bf4929af5/bin:/Users/maoking/.claude/plugins/cache/superpowers-marketplace/superpowers/4.0.3/bin:/Users/maoking/.orbstack/bin",
    "FUNASR_PYTHON": "/opt/homebrew/bin/python3"
  },
  "detected": {
    "python3": "/opt/homebrew/bin/python3",
    "curl": "/usr/bin/curl",
    "ffmpeg": "/opt/homebrew/bin/ffmpeg",
    "ffprobe": "/opt/homebrew/bin/ffprobe",
    "python_version": "3.14.3",
    "python_deps": {
      "funasr": "1.3.1",
      "torch": "2.11.0",
      "fastapi": "0.135.3",
      "uvicorn": "0.44.0"
    },
    "platform": "Darwin",
    "machine": "arm64",
    "detected_at": "2026-04-11T11:25:20.858937"
  }
}

MIT License

Copyright (c) 2025 FunASR Transcribe Skill Contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

---

## Third-Party Dependencies

This Skill utilizes the following third-party services:

- **FunASR Service** (https://github.com/modelscope/FunASR) - Speech recognition service
  - Separate terms of service may apply
  - Requires local FunASR server for operation

This license applies only to the Skill code and does not extend to the
underlying FunASR service or its usage terms.

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
FunASR 自动转录 + 总结脚本

此脚本用于 OpenClaw / Claude Code 等 Agent 环境中，自动完成：
1. 转录音频/视频文件
2. 生成 AI 总结提示词
3. 输出总结内容（供 Agent 调用 LLM 生成总结）
4. 将总结注入 Markdown 文件

用法:
    python auto_transcribe.py <音频文件路径> [选项]

选项:
    --output PATH       输出 Markdown 文件路径（默认与音频同目录）
    --diarize           启用说话人分离
    --model MODEL       指定模型（paraformer / paraformer-onnx / sensevoice / sensevoice-onnx）
    --fast             单人快速模式（自动关闭 diarization，保留当前模型路径）
    --no-summary       跳过总结步骤
    --prompt-only      只返回总结提示词，不生成总结
    --api URL          API 地址（默认 http://127.0.0.1:8765）

示例:
    python auto_transcribe.py /path/to/audio.aac
    python auto_transcribe.py /path/to/audio.mp4 --diarize
    python auto_transcribe.py /path/to/course.m4a --model paraformer-onnx --no-diarize
    python auto_transcribe.py /path/to/audio.m4a --prompt-only
"""

import argparse
import json
import os
import sys
import requests
from pathlib import Path
from urllib.parse import urlparse


def check_server(api_url: str) -> bool:
    """检查服务是否运行"""
    try:
        resp = requests.get(f"{api_url}/health", timeout=5)
        if resp.status_code == 200:
            return True
    except Exception:
        pass
    return False


def start_server(api_url: str):
    """尝试启动服务"""
    print("FunASR 服务未运行，尝试启动...")
    import subprocess
    script_dir = Path(__file__).parent.absolute()
    parsed = urlparse(api_url)
    host = parsed.hostname or "127.0.0.1"
    port = str(parsed.port or 8765)
    subprocess.Popen(
        [sys.executable, str(script_dir / "server.py"), "--host", host, "--port", port],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True
    )
    # 等待服务启动
    import time
    for _ in range(30):
        time.sleep(1)
        if check_server(api_url):
            print("✅ FunASR 服务已启动")
            return True
    print("❌ 无法启动 FunASR 服务")
    return False


def transcribe(file_path: str, output_path: str = None, diarize: bool = False,
               model: str = None, fast: bool = False,
               api_url: str = "http://127.0.0.1:8765",
               extract_slides: bool = False, slide_threshold: float = 27.0) -> dict:
    """转录音频文件"""
    print(f"📝 转录中: {file_path}")

    payload = {
        "file_path": file_path,
        "diarize": diarize,
        "fast": fast,
        "extract_slides": extract_slides,
        "slide_threshold": slide_threshold
    }
    if output_path:
        payload["output_path"] = output_path
    if model:
        payload["model"] = model
    
    resp = requests.post(f"{api_url}/transcribe", json=payload, timeout=600)
    resp.raise_for_status()
    result = resp.json()
    
    if result.get("success"):
        print(f"✅ 转录完成: {result.get('output_path')}")
    else:
        print(f"❌ 转录失败: {result.get('error')}")
        sys.exit(1)
    
    return result


def get_summary_prompt(md_path: str, api_url: str = "http://127.0.0.1:8765") -> dict:
    """获取总结提示词"""
    print(f"📋 生成总结提示词...")
    
    resp = requests.post(
        f"{api_url}/summary",
        json={"md_path": md_path},
        timeout=30
    )
    resp.raise_for_status()
    result = resp.json()
    
    if result.get("success"):
        print(f"✅ 提示词已生成")
    else:
        print(f"❌ 获取提示词失败: {result.get('error')}")
        sys.exit(1)
    
    return result


def inject_summary(md_path: str, summary_content: str, api_url: str = "http://127.0.0.1:8765") -> dict:
    """注入总结到 Markdown 文件"""
    print(f"📝 注入总结到文件...")
    
    resp = requests.post(
        f"{api_url}/inject_summary",
        json={
            "md_path": md_path,
            "summary_content": summary_content
        },
        timeout=30
    )
    resp.raise_for_status()
    result = resp.json()
    
    if result.get("success"):
        print(f"✅ 总结已注入: {result.get('output_path')}")
    else:
        print(f"❌ 注入失败: {result.get('error')}")
        sys.exit(1)
    
    return result


def main():
    parser = argparse.ArgumentParser(description="FunASR 自动转录 + 总结")
    parser.add_argument("file", help="音频/视频文件路径")
    parser.add_argument("--output", "-o", help="输出 Markdown 文件路径")
    parser.add_argument("--diarize", action="store_true", default=True, help="启用说话人分离（默认启用）")
    parser.add_argument("--no-diarize", action="store_false", dest="diarize", help="禁用说话人分离")
    parser.add_argument("--model", choices=["paraformer", "paraformer-onnx", "sensevoice", "sensevoice-onnx"], help="指定模型")
    parser.add_argument("--fast", action="store_true", help="单人快速模式：关闭 diarization，保留当前模型路径")
    parser.add_argument("--no-summary", action="store_true", help="跳过总结步骤")
    parser.add_argument("--prompt-only", action="store_true", help="只返回总结提示词，不生成总结")
    parser.add_argument("--api", default="http://127.0.0.1:8765", help="API 地址")
    parser.add_argument("--slides", action="store_true", help="提取视频关键帧截图（PPT幻灯片）")
    parser.add_argument("--slide-threshold", type=float, default=27.0, help="场景检测阈值（默认27.0，值越低越灵敏）")
    
    args = parser.parse_args()
    
    api_url = args.api

    if args.fast and args.diarize:
        args.diarize = False
        print("⚡ fast 模式已自动关闭说话人分离")
    
    # 检查服务
    if not check_server(api_url):
        if not start_server(api_url):
            print("错误: FunASR 服务未运行且无法启动")
            sys.exit(1)
    
    # 确定输出路径
    file_path = Path(args.file).absolute()
    if args.output:
        output_path = str(Path(args.output).absolute())
    else:
        output_path = str(file_path.with_suffix(".md"))
    
    # 步骤 1: 转录
    transcribe_result = transcribe(
        str(file_path),
        output_path=output_path,
        diarize=args.diarize,
        model=args.model,
        fast=args.fast,
        api_url=api_url,
        extract_slides=args.slides,
        slide_threshold=args.slide_threshold,
    )
    
    md_path = transcribe_result.get("output_path", output_path)
    
    # 步骤 2: 获取总结提示词
    if not args.no_summary:
        summary_result = get_summary_prompt(md_path, api_url)
        
        if args.prompt_only:
            # 只输出提示词（供 Agent 使用）
            print("\n" + "="*60)
            print("总结提示词（供 Agent 调用 LLM 生成总结）:")
            print("="*60)
            print(summary_result.get("summary_prompt", ""))
            print("="*60)
            print(f"\n📄 Markdown 文件: {md_path}")
            print("\n下一步: 使用上面的提示词调用 LLM 生成总结，")
            print("然后使用 inject_summary 端点将总结注入文件。")
        else:
            # 输出完整信息
            print("\n" + "="*60)
            print("📋 总结提示词:")
            print("="*60)
            print(summary_result.get("summary_prompt", ""))
            print("="*60)
            print(f"\n📄 Markdown 文件: {md_path}")
            print("\n💡 提示: 使用上述提示词调用 LLM 生成总结，")
            print("   然后调用 /inject_summary 将总结注入文件。")
            print("   或使用本脚本的完整流程自动完成。")
    else:
        print(f"\n✅ 转录完成: {md_path}")
        print("   (已跳过总结步骤)")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
FunASR Skill 环境检测脚本

检测当前环境是否满足 funasr-transcribe skill 的运行要求。
首次使用前必须运行此脚本进行环境检测。

用法:
    python3 scripts/check_env.py

退出码:
    0 - 所有检测通过
    1 - 检测失败（环境不满足要求）
"""

import sys
import shutil
import subprocess
from pathlib import Path

# 检测结果
issues = []
warnings = []


def check_python():
    """检测 Python 环境"""
    print("=" * 60)
    print("检测 Python3 环境...")
    print("-" * 60)

    # 1. 检查 python3 命令是否可用
    python3_path = shutil.which("python3")
    if python3_path:
        print(f"  ✅ python3 命令可用: {python3_path}")
    else:
        print("  ❌ python3 命令不可用")
        print("  💡 建议: 使用 homebrew 安装 python@3.14")
        print("     brew install python@3.14")
        issues.append("python3 命令不可用")

    # 2. 检查 Python 版本
    try:
        result = subprocess.run(
            ["python3", "--version"],
            capture_output=True,
            text=True,
            timeout=5
        )
        version_output = result.stdout.strip() or result.stderr.strip()
        print(f"  ℹ️  {version_output}")

        # 解析版本号
        version_str = version_output.replace("Python ", "")
        major, minor, _ = version_str.split(".")[:3]
        if int(major) >= 3 and int(minor) >= 8:
            print("  ✅ Python 版本满足要求 (>= 3.8)")
        else:
            issues.append(f"Python 版本过低: {version_str} (需要 >= 3.8)")
            print(f"  ❌ Python 版本过低 (需要 >= 3.8)")
    except FileNotFoundError:
        print("  ❌ 无法执行 python3 命令")
    except subprocess.TimeoutExpired:
        print("  ❌ python3 命令执行超时")
    except Exception as e:
        print(f"  ❌ 检查 Python 版本时出错: {e}")

    print()


def check_curl():
    """检测 curl"""
    print("=" * 60)
    print("检测 curl...")
    print("-" * 60)

    curl_path = shutil.which("curl")
    if curl_path:
        print(f"  ✅ curl 命令可用: {curl_path}")
        try:
            result = subprocess.run(
                ["curl", "--version"],
                capture_output=True,
                text=True,
                timeout=5
            )
            version_line = result.stdout.split("\n")[0]
            print(f"  ℹ️  {version_line}")
        except:
            pass
    else:
        print("  ❌ curl 命令不可用")
        print("  💡 建议: macOS 通常自带 curl，如果不可用请检查 PATH")
        issues.append("curl 命令不可用")

    print()


def check_basic_commands():
    """检测基本命令"""
    print("=" * 60)
    print("检测基本命令 (ls, ps, grep)...")
    print("-" * 60)

    basic_commands = ["ls", "ps", "grep"]
    all_ok = True

    for cmd in basic_commands:
        path = shutil.which(cmd)
        if path:
            print(f"  ✅ {cmd}: {path}")
        else:
            print(f"  ❌ {cmd}: 不可用")
            all_ok = False

    if not all_ok:
        issues.append("部分基本命令不可用")

    print()


def check_skill_dirs():
    """检测 skill 目录结构"""
    print("=" * 60)
    print("检测 Skill 目录结构...")
    print("-" * 60)

    script_dir = Path(__file__).parent.absolute()
    skill_dir = script_dir.parent

    required_files = [
        "SKILL.md",
        "scripts/server.py",
        "scripts/server-onnx.py",
        "scripts/transcribe.py",
        "scripts/auto_transcribe.py",
        "scripts/setup.py",
    ]

    all_ok = True
    for file_path in required_files:
        full_path = skill_dir / file_path
        if full_path.exists():
            print(f"  ✅ {file_path}")
        else:
            print(f"  ❌ {file_path} (缺失)")
            all_ok = False

    if not all_ok:
        issues.append("Skill 文件缺失")

    print()

    return all_ok


def main():
    print()
    print("=" * 60)
    print("  FunASR Skill - 环境检测")
    print("=" * 60)
    print()
    print("首次使用 funasr-transcribe skill 前，请先检测环境是否满足要求。")
    print()

    check_python()
    check_curl()
    check_basic_commands()
    skill_ok = check_skill_dirs()

    # 汇总结果
    print("=" * 60)
    print("  检测结果汇总")
    print("=" * 60)

    if not issues:
        print()
        print("  ✅ 所有检测通过！环境满足要求。")
        print()
        print("  下一步：")
        print("    1. 运行安装脚本: python3 scripts/setup.py")
        print("    2. 启动服务: python3 scripts/server.py")
        print("       或 ONNX 加速服务: python3 scripts/server-onnx.py --preload")
        print("    3. 开始转录！")
        print()
        return 0
    else:
        print()
        print(f"  ❌ 检测到 {len(issues)} 个问题，环境不满足要求：")
        print()
        for i, issue in enumerate(issues, 1):
            print(f"    {i}. {issue}")
        print()
        print("  请修复上述问题后重新运行检测：")
        print("    python3 scripts/check_env.py")
        print()
        return 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
FunASR Skill 环境检测与配置生成脚本

检测当前环境的工具路径和依赖，生成 skill-env.json 供执行器读取。
适用于 Claude Code CLI（诊断用）和 Raycast agent-executor（环境注入用）。

用法:
    python3 scripts/init_env.py          # 检测环境并生成 skill-env.json
    python3 scripts/init_env.py --check  # 只检测，不写文件
    python3 scripts/init_env.py --force  # 强制重新检测（覆盖已有文件）

退出码:
    0 - 检测通过，skill-env.json 已生成
    1 - 检测失败（缺少必要工具）
"""

import os
import sys
import json
import shutil
import platform
import subprocess
import argparse
from pathlib import Path
from datetime import datetime

# skill 根目录
SKILL_DIR = Path(__file__).resolve().parent.parent
ENV_FILE = SKILL_DIR / "assets" / "skill-env.json"

# 必需工具
REQUIRED_TOOLS = ["python3", "curl"]
# 可选但建议的工具
OPTIONAL_TOOLS = ["ffmpeg", "ffprobe"]
# Python 依赖（检测是否已安装）
PYTHON_DEPS = ["funasr", "funasr_onnx", "torch", "fastapi", "uvicorn"]


def get_login_shell_path():
    """通过 login shell 获取完整的 PATH 环境变量"""
    system = platform.system()

    if system == "Darwin" or system == "Linux":
        # 尝试通过 login shell 获取 PATH
        shell = os.environ.get("SHELL", "/bin/zsh")
        try:
            result = subprocess.run(
                [shell, "-l", "-c", "echo $PATH"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0 and result.stdout.strip():
                return result.stdout.strip()
        except (subprocess.TimeoutExpired, FileNotFoundError):
            pass

        # 备选：尝试 path_helper（macOS）
        if system == "Darwin":
            try:
                result = subprocess.run(
                    ["/usr/libexec/path_helper", "-s"],
                    capture_output=True, text=True, timeout=5
                )
                if result.returncode == 0:
                    # path_helper 输出格式: PATH="..."; export PATH;
                    path_line = result.stdout.strip()
                    if 'PATH="' in path_line:
                        path_val = path_line.split('PATH="')[1].split('"')[0]
                        return path_val
            except (subprocess.TimeoutExpired, FileNotFoundError):
                pass

    # 最终回退：使用当前 PATH
    return os.environ.get("PATH", "")


def detect_tool(name, search_path=None):
    """检测工具的实际路径"""
    return shutil.which(name, path=search_path)


def detect_python_version(python_path):
    """检测 Python 版本"""
    if not python_path:
        return None
    try:
        result = subprocess.run(
            [python_path, "--version"],
            capture_output=True, text=True, timeout=5
        )
        output = result.stdout.strip() or result.stderr.strip()
        return output.replace("Python ", "")
    except Exception:
        return None


def check_python_deps():
    """检查 Python 依赖是否已安装"""
    installed = {}
    for dep in PYTHON_DEPS:
        try:
            result = subprocess.run(
                [sys.executable, "-c", f"import {dep}; print({dep}.__version__)"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode == 0:
                installed[dep] = result.stdout.strip()
            else:
                installed[dep] = None
        except Exception:
            installed[dep] = None
    return installed


def run_detection():
    """执行完整的环境检测"""
    issues = []
    warnings = []

    # 1. 获取完整 PATH
    full_path = get_login_shell_path()
    if not full_path:
        issues.append("无法获取 PATH 环境变量")

    # 2. 检测必需工具
    detected_tools = {}
    for tool in REQUIRED_TOOLS:
        path = detect_tool(tool, full_path)
        if path:
            detected_tools[tool] = path
        else:
            # 也尝试在默认 PATH 中查找
            path = detect_tool(tool)
            if path:
                detected_tools[tool] = path
            else:
                issues.append(f"必需工具 {tool} 未找到")

    # 3. 检测可选工具
    for tool in OPTIONAL_TOOLS:
        path = detect_tool(tool, full_path) or detect_tool(tool)
        if path:
            detected_tools[tool] = path
        else:
            if tool == "ffmpeg":
                warnings.append(f"可选工具 {tool} 未安装（视频关键帧提取需要）")

    # 4. Python 版本
    python_version = None
    if "python3" in detected_tools:
        python_version = detect_python_version(detected_tools["python3"])
        if python_version:
            parts = python_version.split(".")
            if len(parts) >= 2 and (int(parts[0]) < 3 or int(parts[1]) < 8):
                issues.append(f"Python 版本过低: {python_version}（需要 >= 3.8）")

    # 5. Python 依赖
    python_deps = check_python_deps()

    return {
        "full_path": full_path,
        "tools": detected_tools,
        "python_version": python_version,
        "python_deps": python_deps,
        "issues": issues,
        "warnings": warnings,
    }


def generate_env_json(detection):
    """生成 skill-env.json 内容"""
    env_data = {
        "env": {
            "PATH": detection["full_path"],
        },
        "detected": {
            **detection["tools"],
            "python_version": detection["python_version"],
            "python_deps": detection["python_deps"],
            "platform": platform.system(),
            "machine": platform.machine(),
            "detected_at": datetime.now().isoformat(),
        },
    }

    # 添加自定义环境变量
    if "python3" in detection["tools"]:
        env_data["env"]["FUNASR_PYTHON"] = detection["tools"]["python3"]

    return env_data


def print_results(detection):
    """打印检测结果"""
    print()
    print("=" * 60)
    print("  FunASR Skill - 环境检测")
    print("=" * 60)
    print()

    # PATH
    print(f"  PATH: {detection['full_path'][:80]}...")
    print()

    # 工具
    print("  工具检测:")
    for tool in REQUIRED_TOOLS:
        path = detection["tools"].get(tool)
        status = f"✅ {path}" if path else "❌ 未找到"
        print(f"    {tool}: {status}")
    for tool in OPTIONAL_TOOLS:
        path = detection["tools"].get(tool)
        status = f"✅ {path}" if path else "⚠️  未安装（可选）"
        print(f"    {tool}: {status}")
    print()

    # Python
    if detection["python_version"]:
        print(f"  Python 版本: {detection['python_version']}")
    print()

    # 依赖
    print("  Python 依赖:")
    for dep, ver in detection["python_deps"].items():
        status = f"✅ {ver}" if ver else "❌ 未安装"
        print(f"    {dep}: {status}")
    print()

    # 问题
    if detection["issues"]:
        print(f"  ❌ 发现 {len(detection['issues'])} 个问题:")
        for issue in detection["issues"]:
            print(f"    - {issue}")
        print()

    if detection["warnings"]:
        for w in detection["warnings"]:
            print(f"  ⚠️  {w}")
        print()


def main():
    parser = argparse.ArgumentParser(description="FunASR Skill 环境检测与配置生成")
    parser.add_argument("--check", action="store_true", help="只检测，不写文件")
    parser.add_argument("--force", action="store_true", help="强制重新检测")
    args = parser.parse_args()

    # 如果已有 skill-env.json 且不是强制模式，跳过
    if ENV_FILE.exists() and not args.force and not args.check:
        print(f"✅ skill-env.json 已存在: {ENV_FILE}")
        print("   使用 --force 强制重新检测")
        return 0

    # 执行检测
    detection = run_detection()
    print_results(detection)

    # 如果有严重问题，报错退出
    if detection["issues"]:
        print("=" * 60)
        print("  ❌ 环境检测未通过，请修复上述问题后重试")
        print("  运行 python3 scripts/setup.py 安装依赖")
        print("=" * 60)
        return 1

    # 生成 skill-env.json
    if not args.check:
        env_data = generate_env_json(detection)
        ENV_FILE.write_text(
            json.dumps(env_data, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        print(f"  ✅ 已生成: {ENV_FILE}")
        print()

    print("=" * 60)
    print("  ✅ 环境检测通过")
    print("=" * 60)
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
FunASR ONNX 加速服务入口。

默认行为：
1. 将默认模型切到 `paraformer-onnx`
2. 打开 ONNX INT8 量化
3. 其余参数复用 server.py
"""

import os

os.environ.setdefault("FUNASR_SERVER_DEFAULT_MODEL", "paraformer-onnx")
os.environ.setdefault("FUNASR_SERVER_DEFAULT_QUANTIZE", "1")

from server import main


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
FunASR 语音转文字 - 一键安装脚本
自动安装依赖和下载模型，支持 Windows/macOS/Linux
"""

import os
import sys
import json
import platform
import subprocess
import argparse
import shutil
from pathlib import Path


# 获取脚本所在目录和 skill 根目录
SCRIPT_DIR = Path(__file__).parent.absolute()
SKILL_DIR = SCRIPT_DIR.parent
REQUIREMENTS_FILE = SKILL_DIR / "assets" / "requirements.txt"
MODELS_CONFIG = SKILL_DIR / "assets" / "models.json"

# 最低系统要求
MIN_MEMORY_GB = 4  # 最低内存要求
MIN_DISK_GB = 5    # 最低磁盘空间要求（模型约 1.2GB + 依赖）
MIN_PYTHON_VERSION = (3, 8)


def print_step(msg: str):
    """打印步骤信息"""
    print(f"\n{'='*60}")
    print(f"  {msg}")
    print(f"{'='*60}\n")


def print_success(msg: str):
    print(f"✅ {msg}")


def print_error(msg: str):
    print(f"❌ {msg}")


def print_warning(msg: str):
    print(f"⚠️  {msg}")


def print_info(msg: str):
    print(f"ℹ️  {msg}")


def get_system_info():
    """获取系统信息"""
    info = {
        'os': platform.system(),
        'os_version': platform.version(),
        'machine': platform.machine(),
        'python_version': sys.version_info,
        'memory_gb': None,
        'disk_free_gb': None,
        'gpu': None,
    }

    # 获取内存信息
    try:
        if info['os'] == 'Darwin':  # macOS
            import subprocess
            result = subprocess.run(['sysctl', '-n', 'hw.memsize'], capture_output=True, text=True)
            info['memory_gb'] = int(result.stdout.strip()) / (1024**3)
        elif info['os'] == 'Windows':
            import ctypes
            kernel32 = ctypes.windll.kernel32
            c_ulong = ctypes.c_ulong
            class MEMORYSTATUS(ctypes.Structure):
                _fields_ = [
                    ('dwLength', c_ulong),
                    ('dwMemoryLoad', c_ulong),
                    ('dwTotalPhys', c_ulong),
                    ('dwAvailPhys', c_ulong),
                    ('dwTotalPageFile', c_ulong),
                    ('dwAvailPageFile', c_ulong),
                    ('dwTotalVirtual', c_ulong),
                    ('dwAvailVirtual', c_ulong),
                ]
            memstatus = MEMORYSTATUS()
            memstatus.dwLength = ctypes.sizeof(MEMORYSTATUS)
            kernel32.GlobalMemoryStatus(ctypes.byref(memstatus))
            info['memory_gb'] = memstatus.dwTotalPhys / (1024**3)
        else:  # Linux
            with open('/proc/meminfo', 'r') as f:
                for line in f:
                    if line.startswith('MemTotal:'):
                        info['memory_gb'] = int(line.split()[1]) / (1024**2)
                        break
    except:
        pass

    # 获取磁盘空间
    try:
        cache_dir = get_model_cache_dir().parent
        cache_dir.mkdir(parents=True, exist_ok=True)
        _, _, free = shutil.disk_usage(cache_dir)
        info['disk_free_gb'] = free / (1024**3)
    except:
        pass

    return info


def check_system_requirements():
    """检查系统环境是否满足要求"""
    print_step("检查系统环境")

    info = get_system_info()
    warnings = []
    errors = []

    # 操作系统
    os_name = info['os']
    if os_name == 'Darwin':
        print_success(f"操作系统: macOS ({info['os_version']})")
        print_info(f"架构: {info['machine']}")
        if info['machine'] == 'arm64':
            print_info("Apple Silicon 检测到，将使用 MPS 加速")
    elif os_name == 'Windows':
        print_success(f"操作系统: Windows ({info['os_version']})")
        print_info(f"架构: {info['machine']}")
    elif os_name == 'Linux':
        print_success(f"操作系统: Linux ({info['os_version']})")
    else:
        warnings.append(f"未测试的操作系统: {os_name}")

    # Python 版本
    py_version = info['python_version']
    print(f"Python 版本: {py_version.major}.{py_version.minor}.{py_version.micro}")
    if py_version < MIN_PYTHON_VERSION:
        errors.append(f"需要 Python {MIN_PYTHON_VERSION[0]}.{MIN_PYTHON_VERSION[1]} 或更高版本")
    else:
        print_success("Python 版本符合要求")

    # 内存检查
    if info['memory_gb']:
        print(f"系统内存: {info['memory_gb']:.1f} GB")
        if info['memory_gb'] < MIN_MEMORY_GB:
            errors.append(f"内存不足，需要至少 {MIN_MEMORY_GB} GB（当前 {info['memory_gb']:.1f} GB）")
        elif info['memory_gb'] < 8:
            warnings.append(f"内存较低（{info['memory_gb']:.1f} GB），大文件转录可能较慢")
            print_warning(f"建议 8GB 以上内存以获得更好性能")
        else:
            print_success("内存充足")
    else:
        warnings.append("无法检测内存大小")

    # 磁盘空间检查
    if info['disk_free_gb']:
        print(f"可用磁盘空间: {info['disk_free_gb']:.1f} GB")
        if info['disk_free_gb'] < MIN_DISK_GB:
            errors.append(f"磁盘空间不足，需要至少 {MIN_DISK_GB} GB（当前 {info['disk_free_gb']:.1f} GB）")
        else:
            print_success("磁盘空间充足")
    else:
        warnings.append("无法检测磁盘空间")

    # GPU 检测（预检测，不需要 PyTorch）
    gpu_info = detect_gpu_without_torch()
    if gpu_info:
        print_info(f"检测到 GPU: {gpu_info}")
    else:
        print_info("未检测到 GPU，将使用 CPU 推理（速度较慢）")

    # 输出警告
    if warnings:
        print("\n注意事项:")
        for w in warnings:
            print_warning(w)

    # 输出错误
    if errors:
        print("\n发现以下问题:")
        for e in errors:
            print_error(e)
        return False

    print_success("\n系统环境检查通过")
    return True


def detect_gpu_without_torch():
    """在不依赖 PyTorch 的情况下检测 GPU"""
    system = platform.system()

    try:
        if system == 'Darwin':
            # macOS: 检查是否有 Apple Silicon
            if platform.machine() == 'arm64':
                return "Apple Silicon (MPS)"
            return None

        elif system == 'Windows':
            # Windows: 尝试使用 nvidia-smi
            result = subprocess.run(
                ['nvidia-smi', '--query-gpu=name', '--format=csv,noheader'],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0 and result.stdout.strip():
                return result.stdout.strip().split('\n')[0]
            return None

        elif system == 'Linux':
            # Linux: 尝试使用 nvidia-smi
            result = subprocess.run(
                ['nvidia-smi', '--query-gpu=name', '--format=csv,noheader'],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0 and result.stdout.strip():
                return result.stdout.strip().split('\n')[0]
            return None

    except (subprocess.TimeoutExpired, FileNotFoundError, Exception):
        pass

    return None


def check_python_version():
    """检查 Python 版本"""
    version = sys.version_info
    if version < MIN_PYTHON_VERSION:
        print_error(f"需要 Python {MIN_PYTHON_VERSION[0]}.{MIN_PYTHON_VERSION[1]} 或更高版本")
        return False
    return True


def is_externally_managed_env():
    """检测是否为外部管理的 Python 环境（如 Homebrew）"""
    try:
        import sysconfig
        stdlib_path = Path(sysconfig.get_path('stdlib'))
        if (stdlib_path / 'EXTERNALLY-MANAGED').exists():
            return True
        # 尝试 dry-run 检测
        result = subprocess.run(
            [sys.executable, "-m", "pip", "install", "--dry-run", "pip"],
            capture_output=True, text=True, timeout=10
        )
        return "externally-managed-environment" in result.stderr
    except:
        return False


def install_dependencies():
    """安装 pip 依赖"""
    print_step("安装依赖包")

    if not REQUIREMENTS_FILE.exists():
        print_error(f"找不到 requirements.txt: {REQUIREMENTS_FILE}")
        return False

    print(f"从 {REQUIREMENTS_FILE} 安装依赖...")
    print_info("这可能需要几分钟，请耐心等待...")

    # 构建安装命令
    cmd = [sys.executable, "-m", "pip", "install", "-r", str(REQUIREMENTS_FILE)]

    # 检测是否为外部管理环境（如 Homebrew Python）
    if is_externally_managed_env():
        print_info("检测到外部管理的 Python 环境（如 Homebrew）")
        print_info("使用 --break-system-packages 标志安装")
        cmd.append("--break-system-packages")

    try:
        subprocess.check_call(cmd)
        print_success("依赖安装完成")
        return True
    except subprocess.CalledProcessError as e:
        print_error(f"依赖安装失败: {e}")
        return False


def load_models_config():
    """加载模型配置"""
    if not MODELS_CONFIG.exists():
        print_error(f"找不到模型配置文件: {MODELS_CONFIG}")
        return None

    with open(MODELS_CONFIG, 'r', encoding='utf-8') as f:
        return json.load(f)


def get_model_cache_dir():
    """获取 ModelScope 模型缓存目录"""
    cache_dir = os.environ.get('MODELSCOPE_CACHE', os.path.expanduser('~/.cache/modelscope/hub'))
    return Path(cache_dir) / "models"


def check_model_exists(model_id: str) -> bool:
    """检查模型是否已下载"""
    cache_dir = get_model_cache_dir()
    model_path = cache_dir / model_id.replace('/', os.sep)
    return model_path.exists() and any(model_path.iterdir())


def download_models():
    """下载所有需要的模型"""
    print_step("下载 ASR 模型")

    config = load_models_config()
    if not config:
        return False

    models = config.get('models', [])
    if not models:
        print_info("没有需要下载的模型")
        return True

    # 检查 modelscope 是否可用
    try:
        from modelscope.hub.snapshot_download import snapshot_download
    except ImportError:
        print_error("modelscope 未安装，请先运行依赖安装")
        return False

    cache_dir = get_model_cache_dir()
    print(f"模型缓存目录: {cache_dir}")

    success_count = 0
    total_count = len([m for m in models if m.get('required', True)])

    for model in models:
        model_id = model['id']
        model_name = model.get('name', model_id)
        required = model.get('required', True)

        if not required:
            continue

        print(f"\n[{success_count + 1}/{total_count}] 处理: {model_name}")
        print(f"    模型 ID: {model_id}")

        # 检查是否已存在
        if check_model_exists(model_id):
            print_success(f"已存在，跳过下载")
            success_count += 1
            continue

        # 下载模型
        print(f"    正在下载...")
        try:
            snapshot_download(model_id)
            print_success(f"下载完成")
            success_count += 1
        except Exception as e:
            print_error(f"下载失败: {e}")
            if required:
                return False

    print(f"\n模型下载完成: {success_count}/{total_count}")
    return success_count == total_count


def verify_installation():
    """验证安装结果"""
    print_step("验证安装")

    errors = []

    # 检查依赖
    try:
        import fastapi
        print_success(f"FastAPI {fastapi.__version__}")
    except ImportError:
        errors.append("FastAPI 未安装")

    try:
        import uvicorn
        print_success(f"Uvicorn {uvicorn.__version__}")
    except ImportError:
        errors.append("Uvicorn 未安装")

    try:
        import funasr
        print_success(f"FunASR 已安装")
    except ImportError:
        errors.append("FunASR 未安装")

    try:
        import funasr_onnx
        print_success(f"funasr-onnx {getattr(funasr_onnx, '__version__', '已安装')}")
    except ImportError:
        errors.append("funasr-onnx 未安装，ONNX 加速功能不可用")

    try:
        import torch
        print_success(f"PyTorch {torch.__version__}")

        # 检查设备
        if torch.cuda.is_available():
            print_success(f"CUDA 可用: {torch.cuda.get_device_name(0)}")
        elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
            print_success("Apple MPS 可用")
        else:
            print_info("将使用 CPU 推理")
    except ImportError:
        errors.append("PyTorch 未安装")

    # 检查说话人分离依赖（scikit-learn）
    try:
        import sklearn
        print_success(f"scikit-learn {sklearn.__version__} (说话人分离需要)")
    except ImportError:
        errors.append("scikit-learn 未安装，说话人分离功能将不可用")
    except Exception as e:
        errors.append(f"scikit-learn 导入失败: {e}，说话人分离功能将不可用")

    # 检查 numpy 版本（避免不兼容问题）
    try:
        import numpy
        print_success(f"numpy {numpy.__version__}")
        if numpy.__version__.startswith('2.'):
            print_warning("检测到 numpy 2.x，可能与部分依赖不兼容，建议使用 numpy<2")
    except ImportError:
        errors.append("numpy 未安装")

    # 检查模型
    config = load_models_config()
    if config:
        models = config.get('models', [])
        for model in models:
            if model.get('required', True):
                if check_model_exists(model['id']):
                    print_success(f"模型: {model.get('name', model['id'])}")
                else:
                    errors.append(f"模型未下载: {model['id']}")

    if errors:
        print("\n发现以下问题:")
        for err in errors:
            print_error(err)
        return False

    print_success("\n安装验证通过!")
    return True


def main():
    parser = argparse.ArgumentParser(
        description='FunASR 语音转文字 - 一键安装脚本（支持 Windows/macOS/Linux）',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python setup.py              # 完整安装（环境检查 + 依赖 + 模型）
  python setup.py --skip-deps  # 只下载模型
  python setup.py --skip-models # 只安装依赖
  python setup.py --verify     # 只验证安装
  python setup.py --check      # 只检查系统环境
"""
    )
    parser.add_argument('--skip-deps', action='store_true', help='跳过依赖安装')
    parser.add_argument('--skip-models', action='store_true', help='跳过模型下载')
    parser.add_argument('--verify', action='store_true', help='只验证安装，不安装任何内容')
    parser.add_argument('--check', action='store_true', help='只检查系统环境')

    args = parser.parse_args()

    print("\n" + "="*60)
    print("  FunASR 语音转文字 - 一键安装")
    print("  支持: Windows / macOS / Linux")
    print("="*60)

    # 只验证安装
    if args.verify:
        success = verify_installation()
        if success:
            print_info("正在刷新环境配置 skill-env.json ...")
            subprocess.run(
                [sys.executable, str(SCRIPT_DIR / "init_env.py"), "--force"],
                check=False,
            )
        sys.exit(0 if success else 1)

    # 只检查环境
    if args.check:
        success = check_system_requirements()
        sys.exit(0 if success else 1)

    # 检查系统环境
    if not check_system_requirements():
        print_error("\n系统环境不满足要求，请解决上述问题后重试")
        sys.exit(1)

    # 安装依赖
    if not args.skip_deps:
        if not install_dependencies():
            sys.exit(1)
    else:
        print_info("跳过依赖安装")

    # 下载模型
    if not args.skip_models:
        if not download_models():
            sys.exit(1)
    else:
        print_info("跳过模型下载")

    # 验证
    if not verify_installation():
        print_error("\n安装可能不完整，请检查上述错误")
        sys.exit(1)

    print_step("安装完成!")
    print("现在可以启动服务:")
    print(f"  python {SCRIPT_DIR / 'server.py'}")
    print()

    # 生成环境配置
    print_info("正在生成环境配置 skill-env.json ...")
    try:
        subprocess.run(
            [sys.executable, str(SCRIPT_DIR / "init_env.py"), "--force"],
            check=False,
        )
    except Exception:
        print_warning("生成 skill-env.json 失败，不影响正常使用")


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
"""
视频关键帧（PPT 幻灯片）提取模块

四层过滤流水线：
  视频 → 第1层：场景检测 + 定时兜底采样
      → 第2层：pHash 去重
      → 第3层：空白回查补帧
      → 第4层：最小间隔过滤 + 兜底帧清理

仅对视频文件生效，音频文件会被跳过。

依赖:
    pip install scenedetect[opencv] imagehash Pillow
"""

import os
from dataclasses import dataclass
from pathlib import Path

# 视频文件扩展名
VIDEO_EXTENSIONS = {'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.webm'}


@dataclass
class SlideFrame:
    """提取的关键帧"""
    timestamp_ms: int       # 毫秒时间戳，用于与转录文本对齐
    image_path: str         # 保存的图片绝对路径
    time_label: str         # 可读标签如 "02m15s"
    relative_path: str = ""  # 相对于 Markdown 文件的路径
    description: str = ""      # 截图描述（用于 alt 文本）
    is_fallback: bool = False  # 是否为兜底帧


class SlideExtractor:
    """视频关键帧提取器

    四层过滤流水线:
        视频 → 场景检测+兜底采样 → pHash 去重 → 空白回查补帧 → 最终过滤

    Args:
        threshold: ContentDetector 灵敏度阈值（默认 20.0，值越低越灵敏）
        min_scene_len: 最小场景时长（秒），短于此间隔的切换会被过滤
        hash_threshold: pHash 汉明距离阈值，低于此值视为相同画面（默认16）
        fallback_interval: 兜底采样间隔（秒），保证每 N 秒至少有一帧（默认60）
        gap_threshold: 空白回查阈值（秒），间隔超过此值时触发均匀补帧（默认180）
    """

    def __init__(
        self,
        threshold: float = 20.0,
        min_scene_len: float = 3.0,
        hash_threshold: int = 16,
        fallback_interval: int = 60,
        gap_threshold: int = 180,
    ):
        self.threshold = threshold
        self.min_scene_len = min_scene_len
        self.hash_threshold = hash_threshold
        self.fallback_interval = fallback_interval
        self.gap_threshold = gap_threshold

    def extract(self, video_path: str, output_dir: str) -> list:
        """主入口：提取关键帧，返回按时间排序的 SlideFrame 列表"""
        ext = Path(video_path).suffix.lower()
        if ext not in VIDEO_EXTENSIONS:
            print(f"[slide_extractor] 跳过非视频文件: {video_path}")
            return []

        os.makedirs(output_dir, exist_ok=True)

        import cv2
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            print(f"[slide_extractor] 无法打开视频: {video_path}")
            return []
        fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        duration_ms = int(total_frames / fps * 1000)
        cap.release()

        # 第 1 层：场景检测 + 定时兜底采样
        raw_frames = self._detect_with_fallback(video_path, output_dir, fps, total_frames)
        if not raw_frames:
            print("[slide_extractor] 未检测到任何帧")
            return []
        print(f"[slide_extractor] 第1层（场景+兜底）: {len(raw_frames)} 个候选帧")

        # 第 2 层：pHash 去重
        deduped = self._deduplicate(raw_frames)
        print(f"[slide_extractor] 第2层（pHash去重）: {len(deduped)} 个帧")

        # 第 3 层：空白回查补帧
        backfilled = self._backfill_gaps(video_path, output_dir, deduped, fps, duration_ms)
        print(f"[slide_extractor] 第3层（空白回查）: {len(backfilled)} 个帧")

        # 第 4 层：最小间隔过滤 + 兜底帧清理
        filtered = self._final_filter(backfilled)
        print(f"[slide_extractor] 第4层（最终过滤）: {len(filtered)} 个关键帧")

        # 清理被过滤掉的图片文件
        kept_paths = {f.image_path for f in filtered}
        for f in raw_frames + backfilled:
            if f.image_path not in kept_paths and os.path.exists(f.image_path):
                os.remove(f.image_path)

        return filtered

    def _detect_with_fallback(self, video_path: str, output_dir: str,
                               fps: float, total_frames: int) -> list:
        """第 1 层：双通道合并（场景检测 + 定时兜底采样）"""
        import cv2

        frames = {}  # key: timestamp_ms, value: SlideFrame

        # --- 通道 A：PySceneDetect ---
        scene_frames = self._detect_scenes(video_path, output_dir)
        for f in scene_frames:
            if f.timestamp_ms not in frames:
                frames[f.timestamp_ms] = f

        # --- 通道 B：定时兜底采样 ---
        duration_ms = int(total_frames / fps * 1000)
        fallback_ms = self.fallback_interval * 1000
        cap = cv2.VideoCapture(video_path)
        idx = 1

        ts_ms = 0
        while ts_ms <= duration_ms:
            if ts_ms not in frames:
                frame_pos = int(ts_ms / 1000.0 * fps)
                cap.set(cv2.CAP_PROP_POS_FRAMES, frame_pos)
                ret, frame = cap.read()
                if ret:
                    ts_s = ts_ms / 1000.0
                    time_label = self._format_time_label(ts_s)
                    img_filename = f"slide_fb_{idx:03d}_{time_label}.jpg"
                    img_path = os.path.join(output_dir, img_filename)
                    cv2.imwrite(img_path, frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
                    frames[ts_ms] = SlideFrame(
                        timestamp_ms=ts_ms,
                        image_path=img_path,
                        time_label=time_label,
                        is_fallback=True,
                    )
                    idx += 1
            ts_ms += fallback_ms

        cap.release()

        # 按时间排序
        return sorted(frames.values(), key=lambda f: f.timestamp_ms)

    def _detect_scenes(self, video_path: str, output_dir: str) -> list:
        """场景检测子函数"""
        try:
            from scenedetect import SceneManager, open_video
            from scenedetect.detectors import ContentDetector
        except ImportError:
            print("[slide_extractor] 缺少依赖，请安装: pip install scenedetect[opencv]")
            return []

        try:
            video = open_video(video_path)
        except Exception as e:
            print(f"[slide_extractor] 无法打开视频: {e}")
            return []

        scene_manager = SceneManager()
        scene_manager.add_detector(
            ContentDetector(
                threshold=self.threshold,
                min_scene_len=int(self.min_scene_len * video.frame_rate) if video.frame_rate else 90,
            )
        )

        scene_manager.detect_scenes(video)
        scene_list = scene_manager.get_scene_list()

        if not scene_list:
            return []

        import cv2
        frames = []
        cap = cv2.VideoCapture(video_path)
        fps = cap.get(cv2.CAP_PROP_FPS) or 25.0

        for i, (start, _end) in enumerate(scene_list):
            frame_num = start.get_frames()
            timestamp_sec = frame_num / fps
            timestamp_ms = int(timestamp_sec * 1000)

            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_num)
            ret, frame = cap.read()
            if not ret:
                continue

            time_label = self._format_time_label(timestamp_sec)
            img_filename = f"slide_{i + 1:03d}_{time_label}.jpg"
            img_path = os.path.join(output_dir, img_filename)

            cv2.imwrite(img_path, frame, [cv2.IMWRITE_JPEG_QUALITY, 85])

            frames.append(SlideFrame(
                timestamp_ms=timestamp_ms,
                image_path=img_path,
                time_label=time_label,
            ))

        cap.release()
        return frames

    def _deduplicate(self, frames: list) -> list:
        """第 2 层：使用感知哈希去除视觉上相似的帧"""
        try:
            from PIL import Image
            import imagehash
        except ImportError:
            print("[slide_extractor] 缺少依赖，请安装: pip install imagehash Pillow")
            return frames

        if not frames:
            return frames

        result = []
        prev_hash = None

        for frame in frames:
            try:
                img = Image.open(frame.image_path)
                current_hash = imagehash.phash(img)

                if prev_hash is not None:
                    distance = current_hash - prev_hash
                    if distance < self.hash_threshold:
                        # 视觉上相似，跳过
                        continue

                prev_hash = current_hash
                result.append(frame)
            except Exception:
                result.append(frame)

        return result

    def _backfill_gaps(self, video_path: str, output_dir: str,
                       frames: list, fps: float, duration_ms: int) -> list:
        """第 3 层：扫描空白区域，按间隔均匀补帧"""
        if len(frames) < 2:
            return frames

        result = list(frames)
        gaps_to_fill = []

        # 找出所有大间隔
        for i in range(len(result) - 1):
            gap_ms = result[i + 1].timestamp_ms - result[i].timestamp_ms
            if gap_ms > self.gap_threshold * 1000:
                gaps_to_fill.append((result[i].timestamp_ms, result[i + 1].timestamp_ms))

        # 也检查开头和结尾的空白
        if result[0].timestamp_ms > self.gap_threshold * 1000:
            gaps_to_fill.append((0, result[0].timestamp_ms))
        if duration_ms - result[-1].timestamp_ms > self.gap_threshold * 1000:
            gaps_to_fill.append((result[-1].timestamp_ms, duration_ms))

        if not gaps_to_fill:
            return result

        print(f"[slide_extractor] 发现 {len(gaps_to_fill)} 个空白区域，开始回查...")

        import cv2
        cap = cv2.VideoCapture(video_path)
        backfill_frames = []
        idx = 1

        for start_ms, end_ms in gaps_to_fill:
            # 按 fallback_interval 均匀采样，而非只补中间1帧
            gap_ms = end_ms - start_ms
            interval_ms = self.fallback_interval * 1000
            ts_ms = start_ms + interval_ms  # 跳过起始位置（已有帧）

            while ts_ms < end_ms:
                frame_pos = int(ts_ms / 1000.0 * fps)
                cap.set(cv2.CAP_PROP_POS_FRAMES, frame_pos)
                ret, frame = cap.read()
                if not ret:
                    ts_ms += interval_ms
                    continue

                ts_s = ts_ms / 1000.0
                time_label = self._format_time_label(ts_s)
                img_filename = f"slide_bf_{idx:03d}_{time_label}.jpg"
                img_path = os.path.join(output_dir, img_filename)
                cv2.imwrite(img_path, frame, [cv2.IMWRITE_JPEG_QUALITY, 85])

                backfill_frames.append(SlideFrame(
                    timestamp_ms=ts_ms,
                    image_path=img_path,
                    time_label=time_label,
                    is_fallback=True,
                ))
                idx += 1
                ts_ms += interval_ms

        cap.release()

        # 合并并重新排序
        all_frames = result + backfill_frames
        all_frames.sort(key=lambda f: f.timestamp_ms)

        # 对补帧做 pHash 去重（只检查补帧与邻近帧）
        try:
            from PIL import Image
            import imagehash
        except ImportError:
            return all_frames

        final = []
        for frame in all_frames:
            if not frame.is_fallback:
                final.append(frame)
                continue

            # 补帧：检查与前一帧的 pHash 距离
            if final:
                try:
                    prev_img = Image.open(final[-1].image_path)
                    cur_img = Image.open(frame.image_path)
                    prev_h = imagehash.phash(prev_img)
                    cur_h = imagehash.phash(cur_img)
                    if cur_h - prev_h < self.hash_threshold:
                        # 视觉上确实没变化，删除补帧图片
                        if os.path.exists(frame.image_path):
                            os.remove(frame.image_path)
                        continue
                except Exception:
                    pass

            final.append(frame)

        return final

    def _final_filter(self, frames: list) -> list:
        """第 4 层：最小间隔过滤 + 兜底帧智能清理"""
        if not frames:
            return frames

        min_ms = int(self.min_scene_len * 1000)
        result = [frames[0]]

        for frame in frames[1:]:
            gap = frame.timestamp_ms - result[-1].timestamp_ms
            if gap >= min_ms:
                result.append(frame)
            elif frame.is_fallback and gap < min_ms:
                # 兜底帧距离前一帧太近，说明场景检测已覆盖，删除
                if os.path.exists(frame.image_path):
                    os.remove(frame.image_path)

        return result

    @staticmethod
    def _format_time_label(seconds: float) -> str:
        """将秒数格式化为可读标签，如 '02m15s'"""
        m = int(seconds) // 60
        s = int(seconds) % 60
        if m > 0:
            return f"{m:02d}m{s:02d}s"
        return f"{s:02d}s"

    @staticmethod
    def is_video_file(file_path: str) -> bool:
        """判断是否为视频文件"""
        return Path(file_path).suffix.lower() in VIDEO_EXTENSIONS