11 KiB
11 KiB
PySceneDetect 视频拆分服务 JSON-RPC 接口
🎯 概述
PySceneDetect视频拆分服务现在支持JSON-RPC协议，提供标准化的API接口，便于与其他系统集成。
📡 JSON-RPC 协议
输出格式
所有命令的输出都遵循JSON-RPC 2.0规范：
成功响应
{
"jsonrpc": "2.0",
"id": null,
"result": {
// 具体的结果数据
}
}
错误响应
{
"jsonrpc": "2.0",
"id": null,
"error": {
"code": "ERROR_CODE",
"message": "错误描述"
}
}
🔧 可用命令
1. analyze - 视频分析
命令格式
python python_core/services/video_splitter.py analyze <video_path> [--threshold <value>]
参数
video_path: 视频文件路径--threshold: 检测阈值 (默认: 30.0)
成功响应示例
{
"jsonrpc": "2.0",
"id": null,
"result": {
"success": true,
"video_path": "/path/to/video.mp4",
"total_scenes": 3,
"total_duration": 10.04,
"average_scene_duration": 3.35,
"scenes": [
{
"scene_number": 1,
"start_time": 0.0,
"end_time": 4.04,
"duration": 4.04,
"start_frame": 0,
"end_frame": 97
},
{
"scene_number": 2,
"start_time": 4.04,
"end_time": 8.04,
"duration": 4.0,
"start_frame": 97,
"end_frame": 193
},
{
"scene_number": 3,
"start_time": 8.04,
"end_time": 10.04,
"duration": 2.0,
"start_frame": 193,
"end_frame": 241
}
]
}
}
错误响应示例
{
"jsonrpc": "2.0",
"id": null,
"error": {
"code": "ANALYSIS_FAILED",
"message": "Video file not found: /path/to/video.mp4"
}
}
2. detect_scenes - 场景检测
命令格式
python python_core/services/video_splitter.py detect_scenes <video_path> [--threshold <value>] [--detector <type>]
参数
video_path: 视频文件路径--threshold: 检测阈值 (默认: 30.0)--detector: 检测器类型 ("content" 或 "threshold", 默认: "content")
成功响应示例
{
"jsonrpc": "2.0",
"id": null,
"result": {
"success": true,
"video_path": "/path/to/video.mp4",
"total_scenes": 3,
"scenes": [
{
"scene_number": 1,
"start_time": 0.0,
"end_time": 4.04,
"duration": 4.04,
"start_frame": 0,
"end_frame": 97
}
],
"detection_settings": {
"threshold": 30.0,
"detector_type": "content"
}
}
}
3. split - 视频拆分
命令格式
python python_core/services/video_splitter.py split <video_path> [options...]
参数
video_path: 视频文件路径--threshold: 检测阈值 (默认: 30.0)--detector: 检测器类型 (默认: "content")--output-dir: 输出目录--output-base: 输出基础目录
成功响应示例
{
"jsonrpc": "2.0",
"id": null,
"result": {
"success": true,
"message": "Successfully split video into 3 scenes",
"input_video": "/path/to/video.mp4",
"output_directory": "/tmp/video_splits/video_20250711_201530",
"scenes": [
{
"scene_number": 1,
"start_time": 0.0,
"end_time": 4.04,
"duration": 4.04,
"start_frame": 0,
"end_frame": 97
}
],
"output_files": [
"/tmp/video_splits/video_20250711_201530/video-Scene-001.mp4",
"/tmp/video_splits/video_20250711_201530/video-Scene-002.mp4",
"/tmp/video_splits/video_20250711_201530/video-Scene-003.mp4"
],
"total_scenes": 3,
"total_duration": 10.04,
"processing_time": 3.02
}
}
错误响应示例
{
"jsonrpc": "2.0",
"id": null,
"error": {
"code": "SPLIT_FAILED",
"message": "FFmpeg failed with return code: 1"
}
}
🔍 错误代码
| 错误代码 | 描述 | 可能原因 |
|---|---|---|
ANALYSIS_FAILED |
视频分析失败 | 文件不存在、格式不支持 |
SPLIT_FAILED |
视频拆分失败 | FFmpeg错误、磁盘空间不足 |
INVALID_COMMAND |
无效命令 | 命令名称错误 |
INTERNAL_ERROR |
内部错误 | 程序异常、依赖缺失 |
💻 编程接口使用
Python 示例
import subprocess
import json
def call_video_splitter(command, video_path, **kwargs):
"""调用视频拆分服务"""
cmd = [
"python", "python_core/services/video_splitter.py",
command, video_path
]
# 添加参数
for key, value in kwargs.items():
cmd.extend([f"--{key.replace('_', '-')}", str(value)])
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
# 解析JSON-RPC响应
if result.stdout.startswith("JSONRPC:"):
json_str = result.stdout[8:]
return json.loads(json_str)
else:
return json.loads(result.stdout)
else:
raise Exception(f"Command failed: {result.stderr}")
# 使用示例
try:
# 分析视频
response = call_video_splitter("analyze", "video.mp4", threshold=30.0)
if "result" in response:
result = response["result"]
print(f"检测到 {result['total_scenes']} 个场景")
# 拆分视频
response = call_video_splitter("split", "video.mp4", threshold=30.0)
if "result" in response:
result = response["result"]
if result["success"]:
print(f"拆分成功: {len(result['output_files'])} 个文件")
else:
print(f"拆分失败: {result['message']}")
except Exception as e:
print(f"调用失败: {e}")
Node.js 示例
const { spawn } = require('child_process');
function callVideoSplitter(command, videoPath, options = {}) {
return new Promise((resolve, reject) => {
const args = [
'python_core/services/video_splitter.py',
command,
videoPath
];
// 添加参数
for (const [key, value] of Object.entries(options)) {
args.push(`--${key.replace(/_/g, '-')}`, String(value));
}
const process = spawn('python3', args);
let stdout = '';
let stderr = '';
process.stdout.on('data', (data) => {
stdout += data.toString();
});
process.stderr.on('data', (data) => {
stderr += data.toString();
});
process.on('close', (code) => {
if (code === 0) {
try {
// 解析JSON-RPC响应
let jsonStr = stdout.trim();
if (jsonStr.startsWith('JSONRPC:')) {
jsonStr = jsonStr.substring(8);
}
const response = JSON.parse(jsonStr);
resolve(response);
} catch (error) {
reject(new Error(`JSON parse error: ${error.message}`));
}
} else {
reject(new Error(`Command failed: ${stderr}`));
}
});
});
}
// 使用示例
async function example() {
try {
// 分析视频
const analyzeResponse = await callVideoSplitter('analyze', 'video.mp4', {
threshold: 30.0
});
if (analyzeResponse.result) {
console.log(`检测到 ${analyzeResponse.result.total_scenes} 个场景`);
}
// 拆分视频
const splitResponse = await callVideoSplitter('split', 'video.mp4', {
threshold: 30.0,
'output-dir': './output'
});
if (splitResponse.result && splitResponse.result.success) {
console.log(`拆分成功: ${splitResponse.result.output_files.length} 个文件`);
}
} catch (error) {
console.error('调用失败:', error.message);
}
}
🔧 集成建议
1. 错误处理
def safe_call_video_splitter(command, video_path, **kwargs):
try:
response = call_video_splitter(command, video_path, **kwargs)
if "error" in response:
# JSON-RPC错误
error = response["error"]
raise Exception(f"[{error['code']}] {error['message']}")
if "result" in response:
result = response["result"]
if isinstance(result, dict) and not result.get("success", True):
# 业务逻辑错误
raise Exception(f"Operation failed: {result.get('error', 'Unknown error')}")
return result
return response
except json.JSONDecodeError as e:
raise Exception(f"Invalid JSON response: {e}")
except subprocess.CalledProcessError as e:
raise Exception(f"Process error: {e}")
2. 异步处理
import asyncio
import concurrent.futures
async def async_video_splitter(command, video_path, **kwargs):
"""异步调用视频拆分服务"""
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(call_video_splitter, command, video_path, **kwargs)
return await loop.run_in_executor(None, lambda: future.result())
# 使用示例
async def process_videos(video_list):
tasks = []
for video_path in video_list:
task = async_video_splitter("analyze", video_path)
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
3. 批量处理
def batch_process_videos(video_list, command="analyze", **kwargs):
"""批量处理视频"""
results = []
for video_path in video_list:
try:
result = call_video_splitter(command, video_path, **kwargs)
results.append({
"video_path": video_path,
"success": True,
"result": result
})
except Exception as e:
results.append({
"video_path": video_path,
"success": False,
"error": str(e)
})
return results
🎉 总结
PySceneDetect视频拆分服务的JSON-RPC接口提供了：
- ✅ 标准化协议: 遵循JSON-RPC 2.0规范
- ✅ 完整功能: 支持分析、检测、拆分三种操作
- ✅ 详细响应: 包含完整的场景信息和处理结果
- ✅ 错误处理: 标准化的错误代码和消息
- ✅ 易于集成: 支持多种编程语言调用
现在可以轻松地将视频拆分功能集成到任何系统中！
JSON-RPC接口 - 让视频拆分服务更易集成！