comi-logic/mxivideo
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: 修复错误

Browse Source
This commit is contained in:
root
2025-07-12 13:46:00 +08:00
parent 31de1e5a4d
commit a63a58253e
10 changed files with 0 additions and 3158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
examples/jsonrpc_client_demo.py
View File

@@ -1,279 +0,0 @@
#!/usr/bin/env python3
"""
JSON-RPC Client Demo
JSON-RPC 客户端演示
演示如何使用Python客户端调用场景检测JSON-RPC API
"""
import json
import requests
import time
from typing import Dict, Any, Optional
class SceneDetectionClient:
"""场景检测JSON-RPC客户端"""
def __init__(self, server_url: str = "http://localhost:8080"):
self.server_url = server_url
self.request_id = 0
def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""调用JSON-RPC方法"""
self.request_id += 1
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": self.request_id
}
try:
response = requests.post(
self.server_url,
json=payload,
headers={'Content-Type': 'application/json'},
timeout=60
)
if response.status_code == 200:
return response.json()
else:
return {
"error": {
"code": response.status_code,
"message": f"HTTP Error: {response.text}"
}
}
except requests.exceptions.RequestException as e:
return {
"error": {
"code": -1,
"message": f"Request failed: {str(e)}"
}
}
def detect_scenes(self, video_path: str, detector_type: str = "content",
threshold: float = 30.0, min_scene_length: float = 1.0) -> Dict[str, Any]:
"""基础场景检测"""
params = {
"video_path": video_path,
"detector_type": detector_type,
"threshold": threshold,
"min_scene_length": min_scene_length
}
return self._call_method("scene.detect", params)
def detect_scenes_workflow(self, video_path: str, detector_type: str = "content",
threshold: float = 30.0, min_scene_length: float = 1.0,
output_path: Optional[str] = None, output_format: str = "json",
enable_ai_analysis: bool = True) -> Dict[str, Any]:
"""LangGraph工作流场景检测"""
params = {
"video_path": video_path,
"detector_type": detector_type,
"threshold": threshold,
"min_scene_length": min_scene_length,
"enable_ai_analysis": enable_ai_analysis
}
if output_path:
params["output_path"] = output_path
params["output_format"] = output_format
return self._call_method("scene.detect_workflow", params)
def get_video_info(self, video_path: str) -> Dict[str, Any]:
"""获取视频信息"""
params = {"video_path": video_path}
return self._call_method("scene.get_video_info", params)
def batch_detect(self, directory: str, detector_type: str = "content",
threshold: float = 30.0, min_scene_length: float = 1.0,
output_dir: Optional[str] = None, output_format: str = "json") -> Dict[str, Any]:
"""批量场景检测"""
params = {
"directory": directory,
"detector_type": detector_type,
"threshold": threshold,
"min_scene_length": min_scene_length,
"output_format": output_format
}
if output_dir:
params["output_dir"] = output_dir
return self._call_method("scene.batch_detect", params)
def demo_basic_detection():
"""演示基础场景检测"""
print("🎯 基础场景检测演示")
print("=" * 50)
client = SceneDetectionClient("http://localhost:8081")
# 测试视频路径
video_path = "assets/1/1752032011698.mp4"
print(f"📹 检测视频: {video_path}")
# 调用基础检测
start_time = time.time()
result = client.detect_scenes(video_path, threshold=15.0)
end_time = time.time()
if "error" in result:
print(f"❌ 检测失败: {result['error']}")
return
detection_result = result.get("result", {})
if detection_result.get("success"):
print(f"✅ 检测成功!")
print(f" 场景数量: {detection_result['total_scenes']}")
print(f" 视频时长: {detection_result['total_duration']:.2f}秒")
print(f" 检测时间: {detection_result['detection_time']:.2f}秒")
print(f" API调用时间: {end_time - start_time:.2f}秒")
# 显示场景详情
scenes = detection_result.get("scenes", [])
print(f"\n🎬 场景详情:")
for scene in scenes[:5]: # 只显示前5个
print(f" 场景 {scene['index']}: {scene['start_time']:.2f}s - {scene['end_time']:.2f}s")
if len(scenes) > 5:
print(f" ... 还有 {len(scenes) - 5} 个场景")
else:
print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}")
def demo_workflow_detection():
"""演示工作流场景检测"""
print("\n🔄 LangGraph工作流检测演示")
print("=" * 50)
client = SceneDetectionClient("http://localhost:8081")
# 测试视频路径
video_path = "assets/1/1752032011698.mp4"
print(f"📹 检测视频: {video_path}")
# 调用工作流检测
start_time = time.time()
result = client.detect_scenes_workflow(
video_path,
threshold=15.0,
enable_ai_analysis=False # 禁用AI分析以避免API密钥问题
)
end_time = time.time()
if "error" in result:
print(f"❌ 工作流检测失败: {result['error']}")
return
workflow_result = result.get("result", {})
detection_result = workflow_result.get("detection_result", {})
video_info = workflow_result.get("video_info", {})
ai_analysis = workflow_result.get("ai_analysis")
workflow_state = workflow_result.get("workflow_state")
if detection_result.get("success"):
print(f"✅ 工作流检测成功!")
print(f" 工作流状态: {workflow_state}")
print(f" 场景数量: {detection_result['total_scenes']}")
print(f" 检测时间: {detection_result['detection_time']:.2f}秒")
print(f" API调用时间: {end_time - start_time:.2f}秒")
# 显示视频信息
if video_info:
print(f"\n📹 视频信息:")
print(f" 分辨率: {video_info.get('resolution')}")
print(f" 帧率: {video_info.get('fps'):.2f} fps")
print(f" 时长: {video_info.get('duration'):.2f}秒")
# 显示AI分析结果
if ai_analysis:
print(f"\n🧠 AI分析: {ai_analysis}")
else:
print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}")
def demo_video_info():
"""演示获取视频信息"""
print("\n📊 视频信息获取演示")
print("=" * 50)
client = SceneDetectionClient("http://localhost:8081")
# 测试视频路径
video_path = "assets/1/1752032011698.mp4"
print(f"📹 获取视频信息: {video_path}")
# 获取视频信息
start_time = time.time()
result = client.get_video_info(video_path)
end_time = time.time()
if "error" in result:
print(f"❌ 获取失败: {result['error']}")
return
info_result = result.get("result", {})
if info_result.get("success"):
info = info_result.get("info", {})
print(f"✅ 获取成功!")
print(f" 文件名: {info.get('filename')}")
print(f" 分辨率: {info.get('resolution')}")
print(f" 帧率: {info.get('fps'):.2f} fps")
print(f" 总帧数: {info.get('frame_count'):,}")
print(f" 时长: {info.get('duration'):.2f}秒")
print(f" 文件大小: {info.get('file_size'):,} 字节")
print(f" API调用时间: {end_time - start_time:.2f}秒")
else:
print(f"❌ 获取失败: {info_result.get('error', '未知错误')}")
def main():
"""主演示函数"""
print("🚀 JSON-RPC 场景检测客户端演示")
print("=" * 60)
# 检查服务器连接
client = SceneDetectionClient("http://localhost:8081")
try:
# 简单的连接测试
test_result = client.get_video_info("nonexistent.mp4")
if "error" in test_result and "Request failed" in str(test_result["error"]):
print("❌ 无法连接到JSON-RPC服务器")
print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081")
return
except Exception as e:
print(f"❌ 连接测试失败: {e}")
return
print("✅ 服务器连接正常")
# 运行演示
demo_video_info()
demo_basic_detection()
demo_workflow_detection()
print("\n🎉 演示完成!")
print("\n💡 更多用法:")
print(" • 调整检测阈值以获得不同的场景分割效果")
print(" • 使用不同的检测器类型 (content/threshold/adaptive)")
print(" • 启用AI分析获得智能建议 (需要配置API密钥)")
print(" • 批量处理多个视频文件")
if __name__ == "__main__":
main()

86
examples/test_jsonrpc_registration.py
View File

@@ -1,86 +0,0 @@
#!/usr/bin/env python3
"""
Test JSON-RPC Registration
测试JSON-RPC方法注册
"""
import sys
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from python_core.utils.jsonrpc_enhanced import method_registry
from python_core.scene_detection import SceneDetector
def test_method_registration():
"""测试方法注册"""
print("🧪 测试JSON-RPC方法注册...")
# 创建检测器并注册方法
detector = SceneDetector()
detector.register_jsonrpc_methods()
# 检查注册的方法
print(f"📋 已注册的方法: {list(method_registry.methods.keys())}")
# 测试方法调用
test_request = '''
{
"jsonrpc": "2.0",
"method": "scene.get_video_info",
"params": {
"video_path": "assets/1/1752032011698.mp4"
},
"id": 1
}
'''
print("📤 测试方法调用...")
response = method_registry.handle_request(test_request)
if response:
print("📥 收到响应:")
print(response)
else:
print("📥 收到空响应（异步模式）")
return True
def test_direct_method_call():
"""测试直接方法调用"""
print("\n🧪 测试直接方法调用...")
detector = SceneDetector()
# 直接调用方法
result = detector.jsonrpc_get_video_info("assets/1/1752032011698.mp4")
print(f"📋 直接调用结果: {result}")
return True
def main():
"""主函数"""
print("🚀 开始测试JSON-RPC方法注册")
print("=" * 50)
try:
test_method_registration()
test_direct_method_call()
print("\n✅ 所有测试通过")
return True
except Exception as e:
print(f"\n❌ 测试失败: {e}")
import traceback
print(f"详细错误: {traceback.format_exc()}")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)

271
examples/test_jsonrpc_workflow.py
View File

@@ -1,271 +0,0 @@
#!/usr/bin/env python3
"""
测试JSON-RPC工作流进度反馈
Test JSON-RPC Workflow Progress Feedback
"""
import json
import requests
import time
import threading
import websocket
from typing import Dict, Any
class JSONRPCWorkflowTester:
"""JSON-RPC工作流测试器"""
def __init__(self, server_url: str = "http://localhost:8081"):
self.server_url = server_url
self.request_id = 0
self.progress_updates = []
def _call_method(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""调用JSON-RPC方法"""
self.request_id += 1
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": self.request_id
}
print(f"📤 发送请求 (ID: {self.request_id}): {method}")
print(f" 参数: {json.dumps(params, indent=2, ensure_ascii=False)}")
try:
response = requests.post(
self.server_url,
json=payload,
headers={'Content-Type': 'application/json'},
timeout=120 # 增加超时时间
)
if response.status_code == 200:
result = response.json()
print(f"📥 收到响应 (ID: {self.request_id}): {response.status_code}")
return result
else:
print(f"❌ HTTP错误: {response.status_code}")
return {
"error": {
"code": response.status_code,
"message": f"HTTP Error: {response.text}"
}
}
except requests.exceptions.RequestException as e:
print(f"❌ 请求失败: {str(e)}")
return {
"error": {
"code": -1,
"message": f"Request failed: {str(e)}"
}
}
def test_basic_detection(self):
"""测试基础场景检测"""
print("\n" + "="*60)
print("🎯 测试基础场景检测")
print("="*60)
params = {
"video_path": "assets/1/1752032011698.mp4",
"detector_type": "content",
"threshold": 15.0,
"min_scene_length": 1.0
}
start_time = time.time()
result = self._call_method("scene.detect", params)
end_time = time.time()
if "error" in result:
print(f"❌ 检测失败: {result['error']}")
return False
detection_result = result.get("result", {})
if detection_result.get("success"):
print(f"✅ 基础检测成功!")
print(f" 场景数量: {detection_result['total_scenes']}")
print(f" 检测时间: {detection_result['detection_time']:.2f}秒")
print(f" API调用时间: {end_time - start_time:.2f}秒")
return True
else:
print(f"❌ 检测失败: {detection_result.get('error', '未知错误')}")
return False
def test_workflow_detection(self):
"""测试工作流场景检测"""
print("\n" + "="*60)
print("🔄 测试LangGraph工作流检测")
print("="*60)
params = {
"video_path": "assets/1/1752032011698.mp4",
"detector_type": "content",
"threshold": 15.0,
"min_scene_length": 1.0,
"enable_ai_analysis": False # 禁用AI分析避免API密钥问题
}
start_time = time.time()
result = self._call_method("scene.detect_workflow", params)
end_time = time.time()
if "error" in result:
print(f"❌ 工作流检测失败: {result['error']}")
return False
workflow_result = result.get("result", {})
detection_result = workflow_result.get("detection_result", {})
if detection_result.get("success"):
print(f"✅ 工作流检测成功!")
print(f" 工作流状态: {workflow_result.get('workflow_state')}")
print(f" 场景数量: {detection_result['total_scenes']}")
print(f" 检测时间: {detection_result['detection_time']:.2f}秒")
print(f" API调用时间: {end_time - start_time:.2f}秒")
# 显示视频信息
video_info = workflow_result.get("video_info", {})
if video_info:
print(f" 视频信息: {video_info.get('resolution')}, {video_info.get('fps'):.2f}fps")
# 显示AI分析结果
ai_analysis = workflow_result.get("ai_analysis")
if ai_analysis:
print(f" AI分析: {ai_analysis[:100]}...")
return True
else:
print(f"❌ 工作流检测失败: {detection_result.get('error', '未知错误')}")
return False
def test_video_info(self):
"""测试视频信息获取"""
print("\n" + "="*60)
print("📊 测试视频信息获取")
print("="*60)
params = {
"video_path": "assets/1/1752032011698.mp4"
}
start_time = time.time()
result = self._call_method("scene.get_video_info", params)
end_time = time.time()
if "error" in result:
print(f"❌ 获取失败: {result['error']}")
return False
info_result = result.get("result", {})
if info_result.get("success"):
info = info_result.get("info", {})
print(f"✅ 信息获取成功!")
print(f" 文件名: {info.get('filename')}")
print(f" 分辨率: {info.get('resolution')}")
print(f" 帧率: {info.get('fps'):.2f} fps")
print(f" 时长: {info.get('duration'):.2f}秒")
print(f" 文件大小: {info.get('file_size'):,} 字节")
print(f" API调用时间: {end_time - start_time:.2f}秒")
return True
else:
print(f"❌ 获取失败: {info_result.get('error', '未知错误')}")
return False
def test_error_handling(self):
"""测试错误处理"""
print("\n" + "="*60)
print("🚨 测试错误处理")
print("="*60)
# 测试不存在的文件
params = {
"video_path": "nonexistent_video.mp4",
"threshold": 15.0
}
result = self._call_method("scene.detect", params)
if "error" in result:
print(f"✅ 正确处理了请求错误: {result['error']['message']}")
else:
detection_result = result.get("result", {})
if not detection_result.get("success"):
print(f"✅ 正确处理了应用错误: {detection_result.get('error')}")
else:
print(f"❌ 应该返回错误，但返回了成功结果")
return False
# 测试无效的方法
result = self._call_method("scene.invalid_method", {})
if "error" in result and result["error"]["code"] == -32601:
print(f"✅ 正确处理了方法不存在错误")
else:
print(f"❌ 没有正确处理方法不存在错误")
return False
return True
def run_all_tests(self):
"""运行所有测试"""
print("🚀 JSON-RPC工作流测试开始")
print("="*60)
# 检查服务器连接
try:
test_result = self._call_method("scene.get_video_info", {"video_path": "test.mp4"})
if "error" in test_result and "Request failed" in str(test_result["error"]):
print("❌ 无法连接到JSON-RPC服务器")
print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081")
return False
except Exception as e:
print(f"❌ 连接测试失败: {e}")
return False
print("✅ 服务器连接正常")
# 运行测试
tests = [
("视频信息获取", self.test_video_info),
("基础场景检测", self.test_basic_detection),
("工作流场景检测", self.test_workflow_detection),
("错误处理", self.test_error_handling),
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
if test_func():
passed += 1
print(f"✅ {test_name} - 通过")
else:
print(f"❌ {test_name} - 失败")
except Exception as e:
print(f"❌ {test_name} - 异常: {e}")
print("\n" + "="*60)
print(f"🎉 测试完成: {passed}/{total} 通过")
print("="*60)
if passed == total:
print("🎊 所有测试都通过了！JSON-RPC工作流功能正常")
else:
print("⚠️ 部分测试失败，请检查服务器和配置")
return passed == total
def main():
"""主函数"""
tester = JSONRPCWorkflowTester("http://localhost:8081")
tester.run_all_tests()
if __name__ == "__main__":
main()

248
examples/test_refactored_modules.py
View File

@@ -1,248 +0,0 @@
#!/usr/bin/env python3
"""
Test Refactored Modules
测试重构后的模块
验证重构后的场景检测模块是否正常工作
"""
import sys
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from python_core.scene_detection import (
SceneDetector,
DetectorType,
OutputFormat,
SceneInfo,
DetectionResult,
SceneDetectionWorkflowState
)
def test_imports():
"""测试模块导入"""
print("🧪 测试模块导入...")
try:
# 测试类型导入
assert DetectorType.CONTENT == "content"
assert OutputFormat.JSON == "json"
print("✅ 类型导入成功")
# 测试数据模型
scene = SceneInfo(0, 0.0, 5.0, 5.0)
assert scene.index == 0
assert scene.duration == 5.0
print("✅ 数据模型导入成功")
# 测试主接口类
detector = SceneDetector()
assert detector is not None
assert hasattr(detector, 'detect_scenes')
assert hasattr(detector, 'detect_with_workflow')
print("✅ 主接口类导入成功")
return True
except Exception as e:
print(f"❌ 导入测试失败: {e}")
return False
def test_basic_functionality():
"""测试基础功能"""
print("\n🧪 测试基础功能...")
try:
detector = SceneDetector()
# 测试视频信息获取
video_path = Path("assets/1/1752032011698.mp4")
if video_path.exists():
info_result = detector.get_video_info(video_path)
if info_result["success"]:
print("✅ 视频信息获取成功")
print(f" 分辨率: {info_result['info']['resolution']}")
print(f" 时长: {info_result['info']['duration']:.2f}秒")
else:
print(f"⚠️ 视频信息获取失败: {info_result['error']}")
else:
print("⚠️ 测试视频文件不存在，跳过视频信息测试")
# 测试JSON-RPC方法注册
detector.register_jsonrpc_methods()
print("✅ JSON-RPC方法注册成功")
return True
except Exception as e:
print(f"❌ 基础功能测试失败: {e}")
import traceback
print(f"详细错误: {traceback.format_exc()}")
return False
def test_services():
"""测试服务层"""
print("\n🧪 测试服务层...")
try:
from python_core.scene_detection.services import (
SceneDetectorService,
VideoInfoService,
AIAnalysisService
)
# 测试检测服务
detector_service = SceneDetectorService()
assert len(detector_service.supported_formats) > 0
print("✅ 场景检测服务初始化成功")
# 测试视频信息服务
video_info_service = VideoInfoService()
assert hasattr(video_info_service, 'extract_video_info')
print("✅ 视频信息服务初始化成功")
# 测试AI分析服务
ai_service = AIAnalysisService()
assert hasattr(ai_service, 'analyze_detection_result')
print(f"✅ AI分析服务初始化成功 (AI启用: {ai_service.ai_enabled})")
return True
except Exception as e:
print(f"❌ 服务层测试失败: {e}")
return False
def test_workflows():
"""测试工作流"""
print("\n🧪 测试工作流...")
try:
from python_core.scene_detection.workflows import (
SceneDetectionWorkflowManager,
WorkflowNodes
)
# 测试工作流管理器
workflow_manager = SceneDetectionWorkflowManager()
assert hasattr(workflow_manager, 'create_detection_workflow')
print("✅ 工作流管理器初始化成功")
# 测试工作流节点
nodes = WorkflowNodes()
assert hasattr(nodes, 'validate_input')
assert hasattr(nodes, 'detect_scenes')
assert hasattr(nodes, 'finalize_results')
print("✅ 工作流节点初始化成功")
# 测试工作流创建
workflow = workflow_manager.create_detection_workflow()
if workflow:
print("✅ LangGraph工作流创建成功")
else:
print("⚠️ LangGraph工作流创建失败（可能是依赖问题）")
return True
except Exception as e:
print(f"❌ 工作流测试失败: {e}")
return False
def test_utils():
"""测试工具类"""
print("\n🧪 测试工具类...")
try:
from python_core.scene_detection.utils import (
ResultSaver,
InputValidator
)
# 测试结果保存器
saver = ResultSaver()
assert hasattr(saver, 'save_results')
print("✅ 结果保存器初始化成功")
# 测试输入验证器
validator = InputValidator({'.mp4', '.avi'})
assert hasattr(validator, 'validate_video_path')
print("✅ 输入验证器初始化成功")
return True
except Exception as e:
print(f"❌ 工具类测试失败: {e}")
return False
def test_cli_integration():
"""测试CLI集成"""
print("\n🧪 测试CLI集成...")
try:
# 测试新的CLI模块
from python_core.cli.scene_detect_new import app
assert app is not None
print("✅ 新CLI模块导入成功")
return True
except Exception as e:
print(f"❌ CLI集成测试失败: {e}")
return False
def main():
"""主测试函数"""
print("🚀 开始测试重构后的场景检测模块")
print("=" * 60)
tests = [
("模块导入", test_imports),
("基础功能", test_basic_functionality),
("服务层", test_services),
("工作流", test_workflows),
("工具类", test_utils),
("CLI集成", test_cli_integration),
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
if test_func():
passed += 1
print(f"✅ {test_name} - 通过")
else:
print(f"❌ {test_name} - 失败")
except Exception as e:
print(f"❌ {test_name} - 异常: {e}")
print("\n" + "=" * 60)
print(f"🎉 测试完成: {passed}/{total} 通过")
if passed == total:
print("🎊 所有测试都通过了！重构成功！")
print("\n📋 重构后的模块结构:")
print("├── types/ # 类型定义和数据模型")
print("├── services/ # 核心业务逻辑服务")
print("├── workflows/ # LangGraph工作流")
print("├── utils/ # 工具类和辅助函数")
print("└── scene_detector.py # 主接口类")
else:
print("⚠️ 部分测试失败，请检查重构实现")
return passed == total
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)

206
examples/test_scene_detection_fix.py
View File

@@ -1,206 +0,0 @@
#!/usr/bin/env python3
"""
Test Scene Detection Fix
测试场景检测修复
"""
import sys
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from python_core.scene_detection.services import SceneDetectorService
from python_core.scene_detection.types import DetectorType
def test_basic_detection():
"""测试基础场景检测"""
print("🧪 测试基础场景检测...")
try:
# 创建检测服务
detector_service = SceneDetectorService()
# 测试视频路径
video_path = Path("assets/1/1752032011698.mp4")
if not video_path.exists():
print(f"❌ 测试视频不存在: {video_path}")
return False
print(f"📹 测试视频: {video_path}")
# 执行检测
result = detector_service.detect_scenes(
video_path,
DetectorType.CONTENT,
30.0,
1.0
)
if result.success:
print(f"✅ 检测成功!")
print(f" 场景数量: {result.total_scenes}")
print(f" 检测时间: {result.detection_time:.2f}秒")
print(f" 视频时长: {result.total_duration:.2f}秒")
# 显示前几个场景
for i, scene in enumerate(result.scenes[:3]):
print(f" 场景 {i+1}: {scene.start_time:.2f}s - {scene.end_time:.2f}s (时长: {scene.duration:.2f}s)")
if len(result.scenes) > 3:
print(f" ... 还有 {len(result.scenes) - 3} 个场景")
return True
else:
print(f"❌ 检测失败: {result.error}")
return False
except Exception as e:
print(f"❌ 测试异常: {e}")
import traceback
print(f"详细错误: {traceback.format_exc()}")
return False
def test_video_info():
"""测试视频信息提取"""
print("\n🧪 测试视频信息提取...")
try:
from python_core.scene_detection.services import VideoInfoService
video_info_service = VideoInfoService()
video_path = Path("assets/1/1752032011698.mp4")
if not video_path.exists():
print(f"❌ 测试视频不存在: {video_path}")
return False
info = video_info_service.extract_video_info(video_path)
print(f"✅ 视频信息提取成功!")
print(f" 文件名: {info['filename']}")
print(f" 分辨率: {info['resolution']}")
print(f" 帧率: {info['fps']:.2f} fps")
print(f" 时长: {info['duration']:.2f}秒")
print(f" 文件大小: {info['file_size']:,} 字节")
return True
except Exception as e:
print(f"❌ 测试异常: {e}")
import traceback
print(f"详细错误: {traceback.format_exc()}")
return False
def test_pyscenedetect_version():
"""测试PySceneDetect版本和API"""
print("\n🧪 测试PySceneDetect版本和API...")
try:
import scenedetect
print(f"📦 PySceneDetect版本: {scenedetect.__version__}")
# 测试基本API
from scenedetect import open_video, SceneManager
from scenedetect.detectors import ContentDetector
video_path = Path("assets/1/1752032011698.mp4")
if not video_path.exists():
print(f"❌ 测试视频不存在: {video_path}")
return False
# 打开视频
video = open_video(str(video_path))
scene_manager = SceneManager()
scene_manager.add_detector(ContentDetector(threshold=30.0))
print("📊 执行基础场景检测...")
scene_manager.detect_scenes(video, show_progress=False)
scene_list = scene_manager.get_scene_list()
print(f"✅ 基础检测成功，找到 {len(scene_list)} 个场景")
# 测试时间码API
if scene_list:
start_time, end_time = scene_list[0]
print(f"📋 第一个场景时间码类型: {type(start_time)}")
# 测试不同的时间获取方法
methods = []
if hasattr(start_time, 'total_seconds'):
try:
seconds = start_time.total_seconds()
methods.append(f"total_seconds(): {seconds:.2f}")
except Exception as e:
methods.append(f"total_seconds() 失败: {e}")
if hasattr(start_time, 'get_seconds'):
try:
seconds = start_time.get_seconds()
methods.append(f"get_seconds(): {seconds:.2f}")
except Exception as e:
methods.append(f"get_seconds() 失败: {e}")
try:
seconds = float(start_time)
methods.append(f"float(): {seconds:.2f}")
except Exception as e:
methods.append(f"float() 失败: {e}")
print("🔍 可用的时间获取方法:")
for method in methods:
print(f" - {method}")
return True
except Exception as e:
print(f"❌ 测试异常: {e}")
import traceback
print(f"详细错误: {traceback.format_exc()}")
return False
def main():
"""主函数"""
print("🚀 开始测试场景检测修复")
print("=" * 60)
tests = [
("PySceneDetect版本和API", test_pyscenedetect_version),
("视频信息提取", test_video_info),
("基础场景检测", test_basic_detection),
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
if test_func():
passed += 1
print(f"✅ {test_name} - 通过")
else:
print(f"❌ {test_name} - 失败")
except Exception as e:
print(f"❌ {test_name} - 异常: {e}")
print("\n" + "=" * 60)
print(f"🎉 测试完成: {passed}/{total} 通过")
if passed == total:
print("🎊 所有测试都通过了！场景检测修复成功！")
else:
print("⚠️ 部分测试失败，需要进一步调试")
return passed == total
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)

303
examples/test_workflow_results.py
View File

@@ -1,303 +0,0 @@
#!/usr/bin/env python3
"""
测试工作流结果返回
Test Workflow Result Return
"""
import json
import requests
import time
import threading
import queue
from typing import Dict, Any, Optional
class WorkflowResultTester:
"""工作流结果测试器"""
def __init__(self, server_url: str = "http://localhost:8081"):
self.server_url = server_url
self.request_id = 0
self.progress_queue = queue.Queue()
self.result_queue = queue.Queue()
def _call_method_async(self, method: str, params: Dict[str, Any]) -> str:
"""异步调用JSON-RPC方法"""
self.request_id += 1
request_id = f"test_{self.request_id}_{int(time.time())}"
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": request_id
}
print(f"📤 发送异步请求 (ID: {request_id}): {method}")
print(f" 参数: {json.dumps(params, indent=2, ensure_ascii=False)}")
def make_request():
try:
response = requests.post(
self.server_url,
json=payload,
headers={'Content-Type': 'application/json'},
timeout=120
)
print(f"📥 收到HTTP响应 (ID: {request_id}): {response.status_code}")
if response.status_code == 200:
# 检查是否有响应体
if response.text.strip():
try:
result = response.json()
print(f"📋 JSON响应内容: {json.dumps(result, indent=2, ensure_ascii=False)}")
self.result_queue.put(("response", request_id, result))
except json.JSONDecodeError:
print(f"⚠️ 响应不是有效的JSON: {response.text}")
self.result_queue.put(("invalid_json", request_id, response.text))
else:
print(f"✅ 空响应体 - 表示结果已异步发送")
self.result_queue.put(("empty_response", request_id, None))
else:
print(f"❌ HTTP错误: {response.status_code} - {response.text}")
self.result_queue.put(("http_error", request_id, {
"status_code": response.status_code,
"text": response.text
}))
except requests.exceptions.RequestException as e:
print(f"❌ 请求异常: {str(e)}")
self.result_queue.put(("request_error", request_id, str(e)))
# 在后台线程中发送请求
thread = threading.Thread(target=make_request)
thread.daemon = True
thread.start()
return request_id
def test_workflow_with_progress_monitoring(self):
"""测试带进度监控的工作流"""
print("\n" + "="*60)
print("🔄 测试工作流进度监控和结果返回")
print("="*60)
# 启动进度监控（模拟WebSocket或长轮询）
progress_thread = threading.Thread(target=self._monitor_progress)
progress_thread.daemon = True
progress_thread.start()
# 发送工作流请求
params = {
"video_path": "assets/1/1752032011698.mp4",
"detector_type": "content",
"threshold": 15.0,
"min_scene_length": 1.0,
"enable_ai_analysis": False
}
request_id = self._call_method_async("scene.detect_workflow", params)
# 等待结果
print(f"⏳ 等待工作流完成 (Request ID: {request_id})...")
start_time = time.time()
timeout = 60 # 60秒超时
while time.time() - start_time < timeout:
try:
result_type, result_id, result_data = self.result_queue.get(timeout=1)
if result_id == request_id:
print(f"\n📋 收到结果 (类型: {result_type}):")
if result_type == "response":
print("✅ 收到JSON-RPC响应:")
print(json.dumps(result_data, indent=2, ensure_ascii=False))
return True
elif result_type == "empty_response":
print("✅ 收到空响应 - 结果已通过进度通道发送")
return True
elif result_type == "http_error":
print(f"❌ HTTP错误: {result_data}")
return False
elif result_type == "request_error":
print(f"❌ 请求错误: {result_data}")
return False
else:
print(f"⚠️ 未知结果类型: {result_type}")
return False
except queue.Empty:
continue
print(f"⏰ 等待超时 ({timeout}秒)")
return False
def _monitor_progress(self):
"""监控进度更新（模拟）"""
# 这里应该是WebSocket连接或长轮询
# 为了演示，我们只是打印一些模拟的进度信息
print("📊 进度监控已启动...")
# 模拟进度更新
progress_steps = [
("validate", "🔍 验证输入参数..."),
("extract_info", "📊 提取视频信息..."),
("detect", "🎯 执行场景检测..."),
("analyze", "🧠 AI分析场景结果..."),
("finalize", "📋 整理最终结果...")
]
for step, message in progress_steps:
time.sleep(2) # 模拟处理时间
print(f"📈 [进度] {step}: {message}")
def test_basic_method_comparison(self):
"""测试基础方法对比"""
print("\n" + "="*60)
print("🔍 测试基础方法 vs 工作流方法")
print("="*60)
# 测试基础方法（应该返回结果）
print("1️⃣ 测试基础场景检测方法:")
basic_params = {
"video_path": "assets/1/1752032011698.mp4",
"threshold": 15.0
}
basic_request_id = self._call_method_async("scene.detect", basic_params)
# 等待基础方法结果
time.sleep(5)
try:
result_type, result_id, result_data = self.result_queue.get(timeout=1)
if result_id == basic_request_id:
if result_type == "response":
print("✅ 基础方法正确返回了JSON响应")
print(f" 场景数: {result_data.get('result', {}).get('total_scenes', 'N/A')}")
else:
print(f"⚠️ 基础方法返回了意外的结果类型: {result_type}")
except queue.Empty:
print("⏰ 基础方法响应超时")
print("\n2️⃣ 测试工作流方法:")
# 工作流方法测试在上面的方法中进行
return self.test_workflow_with_progress_monitoring()
def test_error_handling(self):
"""测试错误处理"""
print("\n" + "="*60)
print("🚨 测试工作流错误处理")
print("="*60)
# 测试不存在的文件
params = {
"video_path": "nonexistent_video.mp4",
"threshold": 15.0,
"enable_ai_analysis": False
}
request_id = self._call_method_async("scene.detect_workflow", params)
# 等待错误结果
print(f"⏳ 等待错误处理 (Request ID: {request_id})...")
start_time = time.time()
timeout = 30
while time.time() - start_time < timeout:
try:
result_type, result_id, result_data = self.result_queue.get(timeout=1)
if result_id == request_id:
if result_type == "response" and "error" in result_data:
print("✅ 正确收到错误响应:")
print(json.dumps(result_data, indent=2, ensure_ascii=False))
return True
elif result_type == "empty_response":
print("✅ 错误已通过进度通道发送")
return True
else:
print(f"⚠️ 意外的错误处理结果: {result_type}")
return False
except queue.Empty:
continue
print(f"⏰ 错误处理等待超时")
return False
def run_all_tests(self):
"""运行所有测试"""
print("🚀 工作流结果返回测试开始")
print("="*60)
# 检查服务器连接
try:
response = requests.get(f"{self.server_url.replace('//', '//').replace('http:', 'http:').replace('8081', '8081')}")
except:
try:
# 简单的连接测试
test_payload = {
"jsonrpc": "2.0",
"method": "scene.get_video_info",
"params": {"video_path": "test.mp4"},
"id": "test"
}
response = requests.post(
self.server_url,
json=test_payload,
headers={'Content-Type': 'application/json'},
timeout=5
)
except Exception as e:
print("❌ 无法连接到JSON-RPC服务器")
print("💡 请先启动服务器: python3 -m python_core.cli jsonrpc start --port 8081")
return False
print("✅ 服务器连接正常")
# 运行测试
tests = [
("基础方法对比", self.test_basic_method_comparison),
("错误处理", self.test_error_handling),
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
print(f"\n🧪 运行测试: {test_name}")
if test_func():
passed += 1
print(f"✅ {test_name} - 通过")
else:
print(f"❌ {test_name} - 失败")
except Exception as e:
print(f"❌ {test_name} - 异常: {e}")
print("\n" + "="*60)
print(f"🎉 测试完成: {passed}/{total} 通过")
print("="*60)
if passed == total:
print("🎊 所有测试都通过了！工作流结果返回功能正常")
else:
print("⚠️ 部分测试失败，请检查实现")
return passed == total
def main():
"""主函数"""
tester = WorkflowResultTester("http://localhost:8081")
tester.run_all_tests()
if __name__ == "__main__":
main()

416
examples/textVideoAgentUsage.ts
View File

@@ -1,416 +0,0 @@
/**
* Text Video Agent API 使用示例
*/
import {
TextVideoAgentAPI,
textVideoAgentAPI,
ImageGenerationParams,
VideoGenerationParams,
TaskRequest
} from '../src/services/textVideoAgentAPI'
import {
TaskType,
AspectRatio,
VideoDuration,
PRESET_CONFIGS
} from '../src/services/textVideoAgentTypes'
// ==================== 基础使用示例 ====================
/**
* 示例1: 基础健康检查
*/
async function example1_HealthCheck() {
console.log('=== 健康检查示例 ===')
try {
const health = await textVideoAgentAPI.healthCheck()
console.log('服务状态:', health)
const mjHealth = await textVideoAgentAPI.mjHealthCheck()
console.log('Midjourney状态:', mjHealth)
const jmHealth = await textVideoAgentAPI.jmHealthCheck()
console.log('极梦状态:', jmHealth)
} catch (error) {
console.error('健康检查失败:', error)
}
}
/**
* 示例2: 获取示例提示词
*/
async function example2_GetSamplePrompts() {
console.log('=== 获取示例提示词 ===')
try {
const prompts = await textVideoAgentAPI.getSamplePrompt(TaskType.VLOG)
console.log('VLOG类型示例提示词:', prompts)
} catch (error) {
console.error('获取提示词失败:', error)
}
}
/**
* 示例3: 文件上传
*/
async function example3_FileUpload() {
console.log('=== 文件上传示例 ===')
// 注意：这里需要实际的文件对象
// const fileInput = document.getElementById('fileInput') as HTMLInputElement
// const file = fileInput.files?.[0]
// 模拟文件对象（实际使用时替换为真实文件）
const mockFile = new File(['mock content'], 'test.jpg', { type: 'image/jpeg' })
try {
const uploadResult = await textVideoAgentAPI.uploadFile(mockFile)
console.log('文件上传结果:', uploadResult)
return uploadResult.data // 返回文件URL
} catch (error) {
console.error('文件上传失败:', error)
}
}
// ==================== 图片生成示例 ====================
/**
* 示例4: 基础图片生成
*/
async function example4_BasicImageGeneration() {
console.log('=== 基础图片生成示例 ===')
const params: ImageGenerationParams = {
prompt: '一个美丽的女孩在喝茶，温馨的下午时光，自然光线，高质量摄影',
max_wait_time: 120,
poll_interval: 2
}
try {
const result = await textVideoAgentAPI.generateImageSync(params)
console.log('图片生成结果:', result)
return result.data?.image_url
} catch (error) {
console.error('图片生成失败:', error)
}
}
/**
* 示例5: 带参考图片的图片生成
*/
async function example5_ImageGenerationWithReference() {
console.log('=== 带参考图片的图片生成示例 ===')
// 模拟参考图片文件
const referenceImage = new File(['reference'], 'reference.jpg', { type: 'image/jpeg' })
const params: ImageGenerationParams = {
prompt: '保持人物特征，改变背景为咖啡厅环境',
img_file: referenceImage,
...PRESET_CONFIGS.STANDARD
}
try {
const result = await textVideoAgentAPI.generateImageWithRetry(params, 3)
console.log('带参考图片生成结果:', result)
return result.data?.image_url
} catch (error) {
console.error('带参考图片生成失败:', error)
}
}
/**
* 示例6: 异步图片生成
*/
async function example6_AsyncImageGeneration() {
console.log('=== 异步图片生成示例 ===')
const prompt = '现代简约风格的室内设计，明亮的客厅'
try {
// 提交异步任务
const taskResult = await textVideoAgentAPI.generateImageAsync(prompt)
const taskId = taskResult.data?.task_id
if (!taskId) {
throw new Error('未获取到任务ID')
}
console.log('任务已提交，ID:', taskId)
// 轮询任务状态
const finalResult = await textVideoAgentAPI.pollTaskUntilComplete(
taskId,
2000, // 2秒轮询间隔
120000, // 2分钟超时
(status) => {
console.log('任务状态更新:', status)
}
)
console.log('异步图片生成完成:', finalResult)
return finalResult.data?.image_url
} catch (error) {
console.error('异步图片生成失败:', error)
}
}
// ==================== 视频生成示例 ====================
/**
* 示例7: 基础视频生成
*/
async function example7_BasicVideoGeneration() {
console.log('=== 基础视频生成示例 ===')
// 首先生成一张图片作为视频的基础
const imageUrl = await example4_BasicImageGeneration()
if (!imageUrl) {
console.error('无法获取基础图片，跳过视频生成')
return
}
const params: VideoGenerationParams = {
prompt: '女孩优雅地品茶，轻柔的动作，温馨的氛围',
img_url: imageUrl,
duration: VideoDuration.MEDIUM,
max_wait_time: 300,
poll_interval: 5
}
try {
const result = await textVideoAgentAPI.generateVideoSync(params)
console.log('视频生成结果:', result)
return result.data?.video_url
} catch (error) {
console.error('视频生成失败:', error)
}
}
/**
* 示例8: 异步视频生成
*/
async function example8_AsyncVideoGeneration() {
console.log('=== 异步视频生成示例 ===')
const params: VideoGenerationParams = {
prompt: '城市夜景延时摄影，车流如水，霓虹闪烁',
duration: VideoDuration.LONG
}
try {
const taskResult = await textVideoAgentAPI.generateVideoAsync(params)
const taskId = taskResult.data?.task_id
if (!taskId) {
throw new Error('未获取到任务ID')
}
console.log('视频生成任务已提交，ID:', taskId)
// 轮询任务状态
const finalResult = await textVideoAgentAPI.pollTaskUntilComplete(
taskId,
5000, // 5秒轮询间隔
600000, // 10分钟超时
(status) => {
console.log('视频生成进度:', status)
}
)
console.log('异步视频生成完成:', finalResult)
return finalResult.data?.video_url
} catch (error) {
console.error('异步视频生成失败:', error)
}
}
// ==================== 高级功能示例 ====================
/**
* 示例9: 图片描述功能
*/
async function example9_ImageDescription() {
console.log('=== 图片描述示例 ===')
const imageUrl = 'https://example.com/sample-image.jpg'
try {
const description = await textVideoAgentAPI.describeImageByUrl({
image_url: imageUrl,
max_wait_time: 60
})
console.log('图片描述结果:', description)
return description.data?.description
} catch (error) {
console.error('图片描述失败:', error)
}
}
/**
* 示例10: 任务管理
*/
async function example10_TaskManagement() {
console.log('=== 任务管理示例 ===')
const taskRequest: TaskRequest = {
task_type: TaskType.VLOG,
prompt: '制作一个关于健康生活方式的短视频',
ar: AspectRatio.PORTRAIT
}
try {
// 创建任务
const createResult = await textVideoAgentAPI.createTask(taskRequest)
const taskId = createResult.data?.task_id
if (!taskId) {
throw new Error('任务创建失败')
}
console.log('任务创建成功，ID:', taskId)
// 查询任务状态
const statusResult = await textVideoAgentAPI.getTaskStatusAsync(taskId)
console.log('任务状态:', statusResult)
// 等待任务完成（同步方式）
const finalResult = await textVideoAgentAPI.getTaskResultSync(taskId)
console.log('任务最终结果:', finalResult)
return finalResult
} catch (error) {
console.error('任务管理失败:', error)
}
}
/**
* 示例11: 端到端内容生成
*/
async function example11_EndToEndGeneration() {
console.log('=== 端到端内容生成示例 ===')
try {
const result = await textVideoAgentAPI.generateContentEndToEnd(
'一个关于咖啡文化的精美内容，展现咖啡师的专业技艺',
{
taskType: TaskType.VLOG,
aspectRatio: AspectRatio.PORTRAIT,
videoDuration: VideoDuration.MEDIUM,
generateVideo: true,
onProgress: (step, progress) => {
console.log(`进度更新: ${step} - ${progress}%`)
}
}
)
console.log('端到端生成完成:', result)
return result
} catch (error) {
console.error('端到端生成失败:', error)
}
}
// ==================== 批量处理示例 ====================
/**
* 示例12: 批量图片生成
*/
async function example12_BatchImageGeneration() {
console.log('=== 批量图片生成示例 ===')
const prompts = [
'春天的樱花盛开',
'夏日的海滩风光',
'秋天的枫叶满山',
'冬日的雪景如画'
]
try {
const results = await Promise.allSettled(
prompts.map(async (prompt, index) => {
const params: ImageGenerationParams = {
prompt,
...PRESET_CONFIGS.FAST // 使用快速配置
}
console.log(`开始生成图片 ${index + 1}: ${prompt}`)
const result = await textVideoAgentAPI.generateImageSync(params)
console.log(`图片 ${index + 1} 生成完成`)
return {
prompt,
result: result.data?.image_url,
success: result.status
}
})
)
console.log('批量图片生成结果:', results)
return results
} catch (error) {
console.error('批量图片生成失败:', error)
}
}
// ==================== 主函数 ====================
/**
* 运行所有示例
*/
async function runAllExamples() {
console.log('开始运行 Text Video Agent API 示例...\n')
// 基础功能示例
await example1_HealthCheck()
await example2_GetSamplePrompts()
// 文件操作示例
// await example3_FileUpload() // 需要实际文件
// 图片生成示例
await example4_BasicImageGeneration()
// await example5_ImageGenerationWithReference() // 需要实际文件
await example6_AsyncImageGeneration()
// 视频生成示例
await example7_BasicVideoGeneration()
await example8_AsyncVideoGeneration()
// 高级功能示例
await example9_ImageDescription()
await example10_TaskManagement()
await example11_EndToEndGeneration()
// 批量处理示例
await example12_BatchImageGeneration()
console.log('\n所有示例运行完成！')
}
// 导出示例函数
export {
example1_HealthCheck,
example2_GetSamplePrompts,
example3_FileUpload,
example4_BasicImageGeneration,
example5_ImageGenerationWithReference,
example6_AsyncImageGeneration,
example7_BasicVideoGeneration,
example8_AsyncVideoGeneration,
example9_ImageDescription,
example10_TaskManagement,
example11_EndToEndGeneration,
example12_BatchImageGeneration,
runAllExamples
}
// 如果直接运行此文件，执行所有示例
if (require.main === module) {
runAllExamples().catch(console.error)
}

504
examples/typer_media_manager.py
View File

@@ -1,504 +0,0 @@
#!/usr/bin/env python3
"""
基于Typer开发规范的媒体管理器实现示例
"""
import sys
import time
from pathlib import Path
from typing import Optional, List
from enum import Enum
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
try:
import typer
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
except ImportError:
print("❌ 需要安装 typer 和 rich: pip install typer rich")
sys.exit(1)
from python_core.utils.progress import ProgressJSONRPCCommander
from python_core.services.base import ProgressServiceBase
console = Console()
# 1. 类型定义（遵循规范）
class OutputFormat(str, Enum):
"""输出格式枚举"""
JSON = "json"
CSV = "csv"
TXT = "txt"
def __str__(self) -> str:
return self.value
class ProcessingMode(str, Enum):
"""处理模式枚举"""
FAST = "fast"
BALANCED = "balanced"
QUALITY = "quality"
# 2. 服务类（遵循规范）
class MediaManagerService(ProgressServiceBase):
"""媒体管理服务"""
def get_service_name(self) -> str:
return "media_manager"
def upload_video(self, video_path: Path, tags: List[str] = None, filename: str = None) -> dict:
"""上传单个视频"""
if tags is None:
tags = []
# 模拟上传过程
steps = [
"计算文件哈希...",
"检查重复文件...",
"复制文件到存储...",
"提取视频信息...",
"检测场景变化..."
]
for step in steps:
self.report_progress(step)
time.sleep(0.3) # 模拟处理时间
result = {
"video_path": str(video_path),
"filename": filename or video_path.name,
"tags": tags,
"success": True,
"scenes_detected": 3,
"duration": 120.5
}
# 保存结果到存储
self.save_data("uploads", f"upload_{int(time.time())}", result)
return result
def batch_upload(self, directory: Path, tags: List[str] = None, recursive: bool = False) -> dict:
"""批量上传视频"""
if tags is None:
tags = []
# 扫描视频文件
video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'}
video_files = []
if recursive:
for ext in video_extensions:
video_files.extend(directory.rglob(f"*{ext}"))
else:
for ext in video_extensions:
video_files.extend(directory.glob(f"*{ext}"))
results = []
for i, video_file in enumerate(video_files):
self.report_progress(f"处理文件: {video_file.name} ({i+1}/{len(video_files)})")
try:
result = self.upload_video(video_file, tags)
results.append(result)
except Exception as e:
results.append({
"video_path": str(video_file),
"success": False,
"error": str(e)
})
batch_result = {
"total_files": len(video_files),
"successful": len([r for r in results if r.get("success")]),
"failed": len([r for r in results if not r.get("success")]),
"results": results
}
# 保存批量结果
self.save_data("batch_uploads", f"batch_{int(time.time())}", batch_result)
return batch_result
def get_recent_uploads(self, limit: int = 10) -> List[dict]:
"""获取最近的上传记录"""
keys = self.list_keys("uploads")
recent_keys = sorted(keys)[-limit:]
return list(self.load_batch_data("uploads", recent_keys).values())
# 3. Typer命令行接口（遵循规范）
class MediaManagerCommander(ProgressJSONRPCCommander):
"""基于Typer的媒体管理器命令行接口"""
def __init__(self):
super().__init__("media_manager")
self.app = typer.Typer(
name="media_manager",
help="""
🎬 媒体管理器
功能强大的视频处理和管理工具，支持：
• 视频上传和元数据提取
• 自动场景检测和分割
• 批量处理和进度跟踪
• 多种输出格式
使用 --help 查看具体命令帮助。
""",
rich_markup_mode="rich",
no_args_is_help=True
)
self.service = MediaManagerService()
self._setup_commands()
def _register_commands(self):
"""注册命令（继承要求）"""
pass # Typer通过装饰器自动注册
def _is_progressive_command(self, command: str) -> bool:
"""判断是否需要进度报告"""
return command in ["batch_upload", "analyze"]
def _execute_with_progress(self, command: str, args: dict):
"""执行带进度的命令"""
pass # 通过Typer命令直接处理
def _execute_simple_command(self, command: str, args: dict):
"""执行简单命令"""
pass # 通过Typer命令直接处理
def _setup_commands(self):
"""设置Typer命令"""
@self.app.command()
def upload(
video_path: Path = typer.Argument(
...,
help="📹 视频文件路径",
exists=True
),
tags: Optional[str] = typer.Option(
None,
"--tags", "-t",
help="🏷️ 标签列表（逗号分隔）"
),
filename: Optional[str] = typer.Option(
None,
"--filename", "-f",
help="📝 自定义文件名"
),
verbose: bool = typer.Option(
False,
"--verbose", "-v",
help="📝 详细输出"
)
):
"""
📤 上传视频文件
上传单个视频文件到媒体库，自动进行场景检测和元数据提取。
示例:
media_manager upload video.mp4 --tags "demo,test"
media_manager upload /path/to/video.mp4 --filename "my_video"
注意:
- 支持的格式: MP4, AVI, MOV, MKV, WMV
- 自动检测重复文件
- 自动提取视频元数据
"""
# 参数验证
if not video_path.exists():
console.print("❌ [red]视频文件不存在[/red]")
raise typer.Exit(1)
# 解析标签
tag_list = []
if tags:
tag_list = [tag.strip() for tag in tags.split(",")]
# 显示开始信息
console.print(f"🚀 开始上传: [bold blue]{video_path}[/bold blue]")
try:
# 设置进度回调
def progress_callback(message: str):
console.print(f"📊 {message}")
self.service.set_progress_callback(progress_callback)
# 执行上传
result = self.service.upload_video(video_path, tag_list, filename)
# 显示结果
console.print("✅ [bold green]上传完成[/bold green]")
if verbose or True: # 总是显示基本信息
self._show_upload_result(result)
except Exception as e:
console.print(f"❌ [red]上传失败: {e}[/red]")
if verbose:
console.print_exception()
raise typer.Exit(1)
@self.app.command()
def batch_upload(
input_directory: Path = typer.Argument(
...,
help="📁 输入目录路径",
exists=True,
file_okay=False
),
output_path: Optional[Path] = typer.Option(
None,
"--output", "-o",
help="📄 结果输出文件路径"
),
tags: Optional[str] = typer.Option(
None,
"--tags", "-t",
help="🏷️ 标签列表（逗号分隔）"
),
recursive: bool = typer.Option(
False,
"--recursive", "-r",
help="🔄 递归处理子目录"
),
output_format: OutputFormat = typer.Option(
OutputFormat.JSON,
"--format", "-f",
help="📄 输出格式"
)
):
"""
📦 批量上传视频文件（带进度条）
批量处理目录中的所有视频文件，支持递归扫描和进度跟踪。
示例:
media_manager batch-upload /videos --tags "batch,demo"
media_manager batch-upload /videos -r --output results.json
"""
console.print(f"📦 批量上传目录: [bold blue]{input_directory}[/bold blue]")
# 解析标签
tag_list = []
if tags:
tag_list = [tag.strip() for tag in tags.split(",")]
# 扫描文件数量
video_extensions = {'.mp4', '.avi', '.mov', '.mkv', '.wmv'}
video_files = []
if recursive:
for ext in video_extensions:
video_files.extend(input_directory.rglob(f"*{ext}"))
else:
for ext in video_extensions:
video_files.extend(input_directory.glob(f"*{ext}"))
if not video_files:
console.print("⚠️ [yellow]未找到可处理的视频文件[/yellow]")
return
console.print(f"📋 找到 {len(video_files)} 个视频文件")
# 使用进度任务
with self.create_task("批量上传", len(video_files)) as task:
def progress_callback(message: str):
# 从消息中提取文件信息更新任务
if "处理文件:" in message:
task.update(message=message)
self.service.set_progress_callback(progress_callback)
# 执行批量上传
result = self.service.batch_upload(input_directory, tag_list, recursive)
task.finish(f"批量上传完成: {result['successful']}/{result['total_files']} 成功")
# 显示结果摘要
self._show_batch_result(result)
# 保存结果文件
if output_path:
self._save_batch_results(result, output_path, output_format)
console.print(f"📄 结果已保存到: {output_path}")
@self.app.command()
def list_uploads(
limit: int = typer.Option(
10,
"--limit", "-l",
min=1,
max=100,
help="📊 显示数量限制"
),
verbose: bool = typer.Option(
False,
"--verbose", "-v",
help="📝 详细信息"
)
):
"""
📋 列出最近的上传记录
显示最近上传的视频文件信息和统计数据。
"""
console.print("📋 最近的上传记录")
try:
uploads = self.service.get_recent_uploads(limit)
if not uploads:
console.print("⚠️ [yellow]暂无上传记录[/yellow]")
return
self._show_uploads_table(uploads, verbose)
except Exception as e:
console.print(f"❌ [red]获取记录失败: {e}[/red]")
raise typer.Exit(1)
@self.app.command()
def status():
"""
📊 显示系统状态
显示媒体管理器的当前状态和统计信息。
"""
console.print("📊 媒体管理器状态")
try:
# 获取统计信息
uploads_stats = self.service.get_collection_stats("uploads")
batch_stats = self.service.get_collection_stats("batch_uploads")
# 创建状态表格
table = Table(title="系统状态")
table.add_column("组件", style="cyan")
table.add_column("状态", style="green")
table.add_column("详情", style="yellow")
table.add_row("存储", "✅ 正常", f"JSON文件存储")
table.add_row("上传记录", "✅ 活跃", f"{uploads_stats.get('file_count', 0)} 条记录")
table.add_row("批量任务", "✅ 就绪", f"{batch_stats.get('file_count', 0)} 个任务")
table.add_row("进度系统", "✅ 就绪", "JSON-RPC协议")
console.print(table)
# 创建信息面板
info_panel = Panel(
"[bold blue]媒体管理器运行正常[/bold blue]\n"
"所有组件状态良好，可以开始处理视频文件。",
title="📊 状态摘要",
border_style="green"
)
console.print(info_panel)
except Exception as e:
console.print(f"❌ [red]获取状态失败: {e}[/red]")
raise typer.Exit(1)
def _show_upload_result(self, result: dict):
"""显示上传结果"""
table = Table(title="📤 上传结果")
table.add_column("项目", style="cyan")
table.add_column("值", style="green")
table.add_row("文件名", result["filename"])
table.add_row("标签", ", ".join(result["tags"]) if result["tags"] else "无")
table.add_row("检测场景", str(result["scenes_detected"]))
table.add_row("视频时长", f"{result['duration']:.1f}秒")
console.print(table)
def _show_batch_result(self, result: dict):
"""显示批量结果摘要"""
panel = Panel(
f"[bold green]总文件: {result['total_files']}[/bold green]\n"
f"[bold blue]成功: {result['successful']}[/bold blue]\n"
f"[bold red]失败: {result['failed']}[/bold red]",
title="📦 批量上传摘要",
border_style="green"
)
console.print(panel)
def _show_uploads_table(self, uploads: List[dict], verbose: bool):
"""显示上传记录表格"""
table = Table(title="📋 上传记录")
table.add_column("文件名", style="cyan")
table.add_column("场景数", style="green")
table.add_column("时长", style="yellow")
if verbose:
table.add_column("标签", style="magenta")
for upload in uploads:
row = [
upload["filename"],
str(upload.get("scenes_detected", "N/A")),
f"{upload.get('duration', 0):.1f}s"
]
if verbose:
tags = ", ".join(upload.get("tags", [])) or "无"
row.append(tags)
table.add_row(*row)
console.print(table)
def _save_batch_results(self, result: dict, output_path: Path, format: OutputFormat):
"""保存批量结果"""
if format == OutputFormat.JSON:
import json
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2, ensure_ascii=False)
elif format == OutputFormat.CSV:
import csv
with open(output_path, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['filename', 'success', 'scenes', 'duration', 'error'])
for item in result['results']:
writer.writerow([
item.get('filename', ''),
item.get('success', False),
item.get('scenes_detected', ''),
item.get('duration', ''),
item.get('error', '')
])
else: # TXT
with open(output_path, 'w', encoding='utf-8') as f:
f.write(f"批量上传结果\n")
f.write(f"总文件数: {result['total_files']}\n")
f.write(f"成功: {result['successful']}\n")
f.write(f"失败: {result['failed']}\n\n")
for item in result['results']:
f.write(f"文件: {item.get('filename', '')}\n")
f.write(f" 状态: {'成功' if item.get('success') else '失败'}\n")
if item.get('success'):
f.write(f" 场景数: {item.get('scenes_detected', 'N/A')}\n")
f.write(f" 时长: {item.get('duration', 0):.1f}秒\n")
else:
f.write(f" 错误: {item.get('error', '')}\n")
f.write("\n")
def run(self):
"""运行CLI"""
self.app()
def main():
"""主入口函数"""
commander = MediaManagerCommander()
commander.run()
if __name__ == "__main__":
main()

403
tests/test_media_manager.py
View File

@@ -1,403 +0,0 @@
#!/usr/bin/env python3
"""
MediaManager 测试文件
测试视频切分、场景检测、信息提取等功能
"""
import os
import sys
import unittest
import tempfile
import shutil
from pathlib import Path
from unittest.mock import Mock, patch
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# 导入要测试的模块
from python_core.services.media_manager import (
MediaManager,
DependencyManager,
VideoProcessorFactory,
FFProbeVideoInfoExtractor,
OpenCVVideoInfoExtractor,
PySceneDetectSceneDetector,
OpenCVSceneDetector,
OpenCVThumbnailGenerator,
MD5FileHashCalculator,
OpenCVVideoSegmentCreator
)
class TestDependencyManager(unittest.TestCase):
"""测试依赖管理器"""
def setUp(self):
self.dependency_manager = DependencyManager()
def test_dependency_initialization(self):
"""测试依赖初始化"""
# 检查依赖管理器是否正确初始化
self.assertIsInstance(self.dependency_manager._dependencies, dict)
# 检查是否包含预期的依赖
self.assertIn('opencv', self.dependency_manager._dependencies)
self.assertIn('scenedetect', self.dependency_manager._dependencies)
def test_opencv_availability(self):
"""测试OpenCV可用性检查"""
opencv_available = self.dependency_manager.is_available('opencv')
print(f"OpenCV available: {opencv_available}")
if opencv_available:
# 如果OpenCV可用，测试模块获取
cv2 = self.dependency_manager.get_module('opencv', 'cv2')
self.assertIsNotNone(cv2)
numpy = self.dependency_manager.get_module('opencv', 'numpy')
self.assertIsNotNone(numpy)
def test_scenedetect_availability(self):
"""测试PySceneDetect可用性检查"""
scenedetect_available = self.dependency_manager.is_available('scenedetect')
print(f"PySceneDetect available: {scenedetect_available}")
if scenedetect_available:
# 如果PySceneDetect可用，测试模块获取
VideoManager = self.dependency_manager.get_module('scenedetect', 'VideoManager')
self.assertIsNotNone(VideoManager)
def test_dependency_info(self):
"""测试依赖信息获取"""
opencv_info = self.dependency_manager.get_dependency_info('opencv')
self.assertIsInstance(opencv_info, dict)
self.assertIn('available', opencv_info)
all_deps = self.dependency_manager.get_all_dependencies()
self.assertIsInstance(all_deps, dict)
self.assertIn('opencv', all_deps)
self.assertIn('scenedetect', all_deps)
class TestVideoProcessorFactory(unittest.TestCase):
"""测试视频处理器工厂"""
def setUp(self):
self.dependency_manager = DependencyManager()
self.factory = VideoProcessorFactory(self.dependency_manager)
def test_create_video_info_extractor(self):
"""测试创建视频信息提取器"""
try:
extractor = self.factory.create_video_info_extractor()
self.assertIsNotNone(extractor)
print(f"Created video info extractor: {type(extractor).__name__}")
except Exception as e:
self.fail(f"Failed to create video info extractor: {e}")
def test_create_scene_detector(self):
"""测试创建场景检测器"""
try:
detector = self.factory.create_scene_detector()
self.assertIsNotNone(detector)
print(f"Created scene detector: {type(detector).__name__}")
except Exception as e:
self.fail(f"Failed to create scene detector: {e}")
def test_create_thumbnail_generator(self):
"""测试创建缩略图生成器"""
if self.dependency_manager.is_available('opencv'):
try:
generator = self.factory.create_thumbnail_generator()
self.assertIsNotNone(generator)
print(f"Created thumbnail generator: {type(generator).__name__}")
except Exception as e:
self.fail(f"Failed to create thumbnail generator: {e}")
else:
print("OpenCV not available, skipping thumbnail generator test")
def test_create_hash_calculator(self):
"""测试创建哈希计算器"""
calculator = self.factory.create_hash_calculator()
self.assertIsNotNone(calculator)
self.assertIsInstance(calculator, MD5FileHashCalculator)
class TestVideoInfoExtractors(unittest.TestCase):
"""测试视频信息提取器"""
def setUp(self):
# 使用assets文件夹中的测试视频
self.test_video = project_root / "assets" / "1" / "1752031789460.mp4"
if not self.test_video.exists():
self.skipTest(f"Test video not found: {self.test_video}")
def test_ffprobe_extractor(self):
"""测试FFProbe视频信息提取器"""
try:
extractor = FFProbeVideoInfoExtractor()
info = extractor.extract_video_info(str(self.test_video))
# 验证返回的信息结构
self.assertIsInstance(info, dict)
self.assertIn('duration', info)
self.assertIn('width', info)
self.assertIn('height', info)
self.assertIn('fps', info)
self.assertIn('file_size', info)
print(f"FFProbe video info: {info}")
# 验证数值合理性
self.assertGreater(info['duration'], 0)
self.assertGreater(info['width'], 0)
self.assertGreater(info['height'], 0)
self.assertGreater(info['fps'], 0)
self.assertGreater(info['file_size'], 0)
except Exception as e:
print(f"FFProbe extractor failed (expected if ffprobe not available): {e}")
def test_opencv_extractor(self):
"""测试OpenCV视频信息提取器"""
dependency_manager = DependencyManager()
if not dependency_manager.is_available('opencv'):
self.skipTest("OpenCV not available")
try:
extractor = OpenCVVideoInfoExtractor(dependency_manager)
info = extractor.extract_video_info(str(self.test_video))
# 验证返回的信息结构
self.assertIsInstance(info, dict)
self.assertIn('duration', info)
self.assertIn('width', info)
self.assertIn('height', info)
self.assertIn('fps', info)
self.assertIn('file_size', info)
print(f"OpenCV video info: {info}")
# 验证数值合理性
self.assertGreater(info['duration'], 0)
self.assertGreater(info['width'], 0)
self.assertGreater(info['height'], 0)
self.assertGreater(info['fps'], 0)
self.assertGreater(info['file_size'], 0)
except Exception as e:
self.fail(f"OpenCV extractor failed: {e}")
class TestSceneDetectors(unittest.TestCase):
"""测试场景检测器"""
def setUp(self):
# 使用assets文件夹中的测试视频
self.test_video = project_root / "assets" / "1" / "1752031789460.mp4"
if not self.test_video.exists():
self.skipTest(f"Test video not found: {self.test_video}")
self.dependency_manager = DependencyManager()
def test_pyscenedetect_detector(self):
"""测试PySceneDetect场景检测器"""
if not self.dependency_manager.is_available('scenedetect'):
self.skipTest("PySceneDetect not available")
try:
detector = PySceneDetectSceneDetector(self.dependency_manager)
scene_changes = detector.detect_scenes(str(self.test_video), threshold=30.0)
# 验证返回的场景变化点
self.assertIsInstance(scene_changes, list)
self.assertGreater(len(scene_changes), 0)
# 第一个应该是0.0
self.assertEqual(scene_changes[0], 0.0)
# 场景变化点应该是递增的
for i in range(1, len(scene_changes)):
self.assertGreater(scene_changes[i], scene_changes[i-1])
print(f"PySceneDetect found {len(scene_changes)} scene changes: {scene_changes}")
except Exception as e:
print(f"PySceneDetect detector failed: {e}")
def test_opencv_detector(self):
"""测试OpenCV场景检测器"""
if not self.dependency_manager.is_available('opencv'):
self.skipTest("OpenCV not available")
try:
detector = OpenCVSceneDetector(self.dependency_manager)
scene_changes = detector.detect_scenes(str(self.test_video), threshold=30.0)
# 验证返回的场景变化点
self.assertIsInstance(scene_changes, list)
self.assertGreater(len(scene_changes), 0)
# 第一个应该是0.0
self.assertEqual(scene_changes[0], 0.0)
# 场景变化点应该是递增的
for i in range(1, len(scene_changes)):
self.assertGreater(scene_changes[i], scene_changes[i-1])
print(f"OpenCV found {len(scene_changes)} scene changes: {scene_changes}")
except Exception as e:
self.fail(f"OpenCV detector failed: {e}")
class TestMediaManager(unittest.TestCase):
"""测试媒体管理器"""
def setUp(self):
# 创建临时目录用于测试
self.temp_dir = tempfile.mkdtemp()
self.test_video = project_root / "assets" / "1" / "1752031789460.mp4"
if not self.test_video.exists():
self.skipTest(f"Test video not found: {self.test_video}")
# 使用临时目录创建MediaManager
with patch('python_core.config.settings') as mock_settings:
mock_settings.temp_dir = Path(self.temp_dir)
self.media_manager = MediaManager()
def tearDown(self):
# 清理临时目录
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_media_manager_initialization(self):
"""测试媒体管理器初始化"""
self.assertIsNotNone(self.media_manager)
self.assertIsNotNone(self.media_manager.dependency_manager)
self.assertIsNotNone(self.media_manager.factory)
def test_video_info_extraction(self):
"""测试视频信息提取"""
try:
video_info = self.media_manager._get_video_info(str(self.test_video))
self.assertIsInstance(video_info, dict)
self.assertIn('duration', video_info)
self.assertIn('width', video_info)
self.assertIn('height', video_info)
self.assertIn('fps', video_info)
print(f"Video info extracted: {video_info}")
except Exception as e:
self.fail(f"Video info extraction failed: {e}")
def test_scene_detection(self):
"""测试场景检测"""
try:
scene_changes = self.media_manager._detect_scene_changes(str(self.test_video))
self.assertIsInstance(scene_changes, list)
self.assertGreater(len(scene_changes), 0)
print(f"Scene changes detected: {scene_changes}")
except Exception as e:
self.fail(f"Scene detection failed: {e}")
def test_video_splitting(self):
"""测试视频切分功能"""
try:
# 首先检测场景变化
scene_changes = self.media_manager._detect_scene_changes(str(self.test_video))
# 如果场景变化太少，手动添加一些切分点
if len(scene_changes) < 3:
video_info = self.media_manager._get_video_info(str(self.test_video))
duration = video_info.get('duration', 10.0)
scene_changes = [0.0, duration / 2, duration]
print(f"Using scene changes for splitting: {scene_changes}")
# 执行视频切分
segments = self.media_manager._split_video_by_scenes(
str(self.test_video),
scene_changes,
"test_original_id",
["测试", "分镜"]
)
print(f"Created {len(segments)} video segments")
# 验证切分结果
self.assertIsInstance(segments, list)
for i, segment in enumerate(segments):
print(f"Segment {i}: {segment.filename}, duration: {segment.duration}s")
# 验证片段文件是否存在（如果创建成功的话）
if hasattr(segment, 'file_path') and segment.file_path:
segment_path = Path(segment.file_path)
if segment_path.exists():
print(f" File exists: {segment_path}")
self.assertGreater(segment_path.stat().st_size, 0)
except Exception as e:
print(f"Video splitting failed (may be expected if dependencies missing): {e}")
def run_comprehensive_test():
"""运行全面的测试"""
print("=" * 60)
print("MediaManager 综合测试")
print("=" * 60)
# 检查测试视频
test_videos = list((project_root / "assets").rglob("*.mp4"))
print(f"Found {len(test_videos)} test videos:")
for video in test_videos[:5]: # 只显示前5个
print(f" - {video}")
if not test_videos:
print("❌ No test videos found in assets folder")
return
print("\n" + "=" * 60)
print("开始测试...")
print("=" * 60)
# 运行测试套件
loader = unittest.TestLoader()
suite = unittest.TestSuite()
# 添加测试类
suite.addTests(loader.loadTestsFromTestCase(TestDependencyManager))
suite.addTests(loader.loadTestsFromTestCase(TestVideoProcessorFactory))
suite.addTests(loader.loadTestsFromTestCase(TestVideoInfoExtractors))
suite.addTests(loader.loadTestsFromTestCase(TestSceneDetectors))
suite.addTests(loader.loadTestsFromTestCase(TestMediaManager))
# 运行测试
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
print("\n" + "=" * 60)
print("测试总结")
print("=" * 60)
print(f"运行测试: {result.testsRun}")
print(f"失败: {len(result.failures)}")
print(f"错误: {len(result.errors)}")
print(f"跳过: {len(result.skipped) if hasattr(result, 'skipped') else 0}")
if result.failures:
print("\n失败的测试:")
for test, traceback in result.failures:
print(f" - {test}: {traceback}")
if result.errors:
print("\n错误的测试:")
for test, traceback in result.errors:
print(f" - {test}: {traceback}")
return result.wasSuccessful()
if __name__ == "__main__":
success = run_comprehensive_test()
sys.exit(0 if success else 1)

442
tests/textVideoAgentAPI.test.ts
View File

@@ -1,442 +0,0 @@
/**
* Text Video Agent API 测试文件
*/
import { textVideoAgentAPI, TextVideoAgentAPI } from '../src/services/textVideoAgentAPI'
import { TaskType, AspectRatio, VideoDuration } from '../src/services/textVideoAgentTypes'
// Mock fetch for testing
global.fetch = jest.fn()
describe('TextVideoAgentAPI', () => {
beforeEach(() => {
jest.clearAllMocks()
})
describe('基础功能测试', () => {
test('健康检查', async () => {
const mockResponse = {
status: true,
msg: 'Service is healthy'
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.healthCheck()
expect(fetch).toHaveBeenCalledWith(
expect.stringContaining('/health'),
expect.any(Object)
)
expect(result).toEqual(mockResponse)
})
test('获取示例提示词', async () => {
const mockResponse = {
status: true,
msg: 'Success',
data: {
prompts: ['示例提示词1', '示例提示词2']
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.getSamplePrompt(TaskType.VLOG)
expect(fetch).toHaveBeenCalledWith(
expect.stringContaining('/api/prompt/default?task_type=vlog'),
expect.any(Object)
)
expect(result).toEqual(mockResponse)
})
})
describe('图片生成测试', () => {
test('同步生成图片', async () => {
const mockResponse = {
status: true,
msg: 'Image generated successfully',
data: {
image_url: 'https://example.com/generated-image.jpg'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.generateImageSync({
prompt: '测试图片生成',
max_wait_time: 60
})
expect(result).toEqual(mockResponse)
expect(result.data?.image_url).toBe('https://example.com/generated-image.jpg')
})
test('异步生成图片', async () => {
const mockResponse = {
status: true,
msg: 'Task submitted',
data: {
task_id: 'test-task-123'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.generateImageAsync('测试异步图片生成')
expect(result).toEqual(mockResponse)
expect(result.data?.task_id).toBe('test-task-123')
})
test('带重试的图片生成', async () => {
// 第一次失败
;(fetch as jest.Mock).mockRejectedValueOnce(new Error('Network error'))
// 第二次成功
const mockResponse = {
status: true,
msg: 'Image generated successfully',
data: {
image_url: 'https://example.com/generated-image.jpg'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.generateImageWithRetry({
prompt: '测试重试机制',
max_wait_time: 60
}, 2)
expect(fetch).toHaveBeenCalledTimes(2)
expect(result).toEqual(mockResponse)
})
})
describe('视频生成测试', () => {
test('同步生成视频', async () => {
const mockResponse = {
status: true,
msg: 'Video generated successfully',
data: {
video_url: 'https://example.com/generated-video.mp4'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.generateVideoSync({
prompt: '测试视频生成',
img_url: 'https://example.com/base-image.jpg',
duration: '5'
})
expect(result).toEqual(mockResponse)
expect(result.data?.video_url).toBe('https://example.com/generated-video.mp4')
})
test('异步生成视频', async () => {
const mockResponse = {
status: true,
msg: 'Video task submitted',
data: {
task_id: 'video-task-456'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.generateVideoAsync({
prompt: '测试异步视频生成',
duration: '5'
})
expect(result).toEqual(mockResponse)
expect(result.data?.task_id).toBe('video-task-456')
})
})
describe('任务管理测试', () => {
test('创建任务', async () => {
const mockResponse = {
status: true,
msg: 'Task created successfully',
data: {
task_id: 'new-task-789'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.createTask({
task_type: TaskType.VLOG,
prompt: '测试任务创建',
ar: AspectRatio.PORTRAIT
})
expect(result).toEqual(mockResponse)
expect(result.data?.task_id).toBe('new-task-789')
})
test('查询任务状态', async () => {
const mockResponse = {
status: true,
msg: 'Task status retrieved',
data: {
task_id: 'test-task-123',
status: 'completed',
progress: 100,
result: {
image_url: 'https://example.com/result.jpg'
}
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.getTaskStatusAsync('test-task-123')
expect(result).toEqual(mockResponse)
expect(result.data?.status).toBe('completed')
})
})
describe('文件操作测试', () => {
test('文件上传', async () => {
const mockResponse = {
status: true,
msg: 'File uploaded successfully',
data: 'https://example.com/uploaded-file.jpg'
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const mockFile = new File(['test content'], 'test.jpg', { type: 'image/jpeg' })
const result = await textVideoAgentAPI.uploadFile(mockFile)
expect(result).toEqual(mockResponse)
expect(result.data).toBe('https://example.com/uploaded-file.jpg')
})
})
describe('图片描述测试', () => {
test('通过URL描述图片', async () => {
const mockResponse = {
status: true,
msg: 'Image described successfully',
data: {
description: '这是一张美丽的风景照片'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const result = await textVideoAgentAPI.describeImageByUrl({
image_url: 'https://example.com/image.jpg'
})
expect(result).toEqual(mockResponse)
expect(result.data?.description).toBe('这是一张美丽的风景照片')
})
test('通过文件描述图片', async () => {
const mockResponse = {
status: true,
msg: 'Image described successfully',
data: {
description: '这是一张人物照片'
}
}
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => mockResponse
})
const mockFile = new File(['image content'], 'portrait.jpg', { type: 'image/jpeg' })
const result = await textVideoAgentAPI.describeImageByFile({
img_file: mockFile
})
expect(result).toEqual(mockResponse)
expect(result.data?.description).toBe('这是一张人物照片')
})
})
describe('高级功能测试', () => {
test('端到端内容生成', async () => {
// Mock 创建任务
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => ({
status: true,
data: { task_id: 'end-to-end-task' }
})
})
// Mock 图片生成
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => ({
status: true,
data: { image_url: 'https://example.com/generated.jpg' }
})
})
// Mock 视频生成
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => ({
status: true,
data: { video_url: 'https://example.com/generated.mp4' }
})
})
const result = await textVideoAgentAPI.generateContentEndToEnd(
'测试端到端生成',
{
taskType: TaskType.VLOG,
aspectRatio: AspectRatio.PORTRAIT,
videoDuration: VideoDuration.MEDIUM,
generateVideo: true
}
)
expect(result).toEqual({
imageUrl: 'https://example.com/generated.jpg',
videoUrl: 'https://example.com/generated.mp4',
taskId: 'end-to-end-task'
})
})
test('轮询任务直到完成', async () => {
// 第一次查询 - 进行中
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => ({
status: true,
data: { status: 'running', progress: 50 }
})
})
// 第二次查询 - 完成
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
json: async () => ({
status: true,
data: {
status: 'completed',
progress: 100,
result: { image_url: 'https://example.com/final.jpg' }
}
})
})
const progressCallback = jest.fn()
const result = await textVideoAgentAPI.pollTaskUntilComplete(
'test-task-id',
100, // 快速轮询用于测试
5000, // 5秒超时
progressCallback
)
expect(fetch).toHaveBeenCalledTimes(2)
expect(progressCallback).toHaveBeenCalledTimes(2)
expect(result.data?.status).toBe('completed')
})
})
describe('错误处理测试', () => {
test('网络错误处理', async () => {
;(fetch as jest.Mock).mockRejectedValueOnce(new Error('Network error'))
await expect(textVideoAgentAPI.healthCheck()).rejects.toThrow('Network error')
})
test('HTTP错误处理', async () => {
;(fetch as jest.Mock).mockResolvedValueOnce({
ok: false,
status: 500,
json: async () => ({ error: 'Internal server error' })
})
await expect(textVideoAgentAPI.healthCheck()).rejects.toThrow('HTTP error! status: 500')
})
test('参数验证', async () => {
await expect(
textVideoAgentAPI.describeImageByUrl({} as any)
).rejects.toThrow('image_url is required')
await expect(
textVideoAgentAPI.describeImageByFile({} as any)
).rejects.toThrow('img_file is required')
})
})
describe('自定义配置测试', () => {
test('自定义API端点', () => {
const customAPI = new TextVideoAgentAPI('https://custom-endpoint.com')
expect(customAPI).toBeInstanceOf(TextVideoAgentAPI)
})
})
})
// 集成测试（需要真实API）
describe('集成测试 (需要真实API)', () => {
// 这些测试需要真实的API端点，通常在CI/CD环境中跳过
const shouldRunIntegrationTests = process.env.RUN_INTEGRATION_TESTS === 'true'
test.skipIf(!shouldRunIntegrationTests)('真实健康检查', async () => {
const result = await textVideoAgentAPI.healthCheck()
expect(result.status).toBe(true)
})
test.skipIf(!shouldRunIntegrationTests)('真实图片生成', async () => {
const result = await textVideoAgentAPI.generateImageSync({
prompt: '一朵美丽的花',
max_wait_time: 60
})
expect(result.status).toBe(true)
expect(result.data?.image_url).toBeDefined()
}, 120000) // 2分钟超时
})
export {}