Merge branch 'master' of ssh://gitea.bowongai.com:222/bowong/mxivideo

python_core/services/media_manager.py

@@ -26,6 +26,16 @@ except ImportError:
     logger.warning("Video processing libraries not available. Install opencv-python for full functionality.")
     VIDEO_LIBS_AVAILABLE = False
 
+# PySceneDetect库
+try:
+    from scenedetect import VideoManager, SceneManager
+    from scenedetect.detectors import ContentDetector
+    SCENEDETECT_AVAILABLE = True
+    logger.info("PySceneDetect is available for advanced scene detection")
+except ImportError:
+    logger.warning("PySceneDetect not available. Install scenedetect for better scene detection.")
+    SCENEDETECT_AVAILABLE = False
+
 
 @dataclass
 class VideoSegment:
@@ -210,54 +220,130 @@ class MediaManager:
             }
     
     def _detect_scene_changes(self, file_path: str, threshold: float = 30.0) -> List[float]:
-        """检测场景变化点（转场镜头）"""
+        """使用PySceneDetect检测场景变化点（转场镜头）"""
+        if SCENEDETECT_AVAILABLE:
+            try:
+                # 创建视频管理器
+                video_manager = VideoManager([file_path])
+                scene_manager = SceneManager()
+
+                # 添加内容检测器，threshold参数控制敏感度
+                scene_manager.add_detector(ContentDetector(threshold=threshold))
+
+                # 开始检测
+                video_manager.start()
+                scene_manager.detect_scenes(frame_source=video_manager)
+
+                # 获取场景列表
+                scene_list = scene_manager.get_scene_list()
+
+                # 转换为时间戳列表
+                scene_changes = [0.0]  # 开始时间
+
+                for scene in scene_list:
+                    start_time = scene[0].get_seconds()
+                    end_time = scene[1].get_seconds()
+
+                    # 添加场景开始时间（跳过第一个，因为已经有0.0了）
+                    if start_time > 0 and start_time not in scene_changes:
+                        scene_changes.append(start_time)
+
+                    # 添加场景结束时间
+                    if end_time not in scene_changes:
+                        scene_changes.append(end_time)
+
+                # 确保时间戳排序
+                scene_changes = sorted(list(set(scene_changes)))
+
+                video_manager.release()
+
+                logger.info(f"PySceneDetect found {len(scene_changes)-1} scene changes in video")
+                logger.debug(f"Scene change timestamps: {scene_changes}")
+
+                return scene_changes
+
+            except Exception as e:
+                logger.error(f"PySceneDetect failed: {e}, falling back to OpenCV method")
+                return self._detect_scene_changes_opencv(file_path, threshold)
+        else:
+            logger.warning("PySceneDetect not available, using OpenCV method")
+            return self._detect_scene_changes_opencv(file_path, threshold)
+
+    def _detect_scene_changes_opencv(self, file_path: str, threshold: float = 30.0) -> List[float]:
+        """使用OpenCV检测场景变化点（备用方案）"""
         if not VIDEO_LIBS_AVAILABLE:
-            logger.warning("Video processing not available, returning empty scene changes")
-            return []
-        
+            logger.warning("Video processing not available, returning basic scene changes")
+            # 获取视频时长，返回开始和结束时间
+            try:
+                video_info = self._get_video_info(file_path)
+                duration = video_info.get('duration', 0)
+                return [0.0, duration] if duration > 0 else [0.0]
+            except:
+                return [0.0]
+
         try:
             cap = cv2.VideoCapture(file_path)
             fps = cap.get(cv2.CAP_PROP_FPS)
-            
+
+            if fps <= 0:
+                cap.release()
+                logger.warning(f"Invalid fps ({fps}) for video {file_path}")
+                return [0.0]
+
             scene_changes = [0.0]  # 开始时间
             prev_frame = None
             frame_count = 0
-            
+
+            # 每隔几帧检测一次，提高性能
+            frame_skip = max(1, int(fps / 2))  # 每秒检测2次
+
             while True:
                 ret, frame = cap.read()
                 if not ret:
                     break
-                
-                # 转换为灰度图
-                gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-                
-                if prev_frame is not None:
-                    # 计算帧间差异
-                    diff = cv2.absdiff(prev_frame, gray)
-                    mean_diff = np.mean(diff)
-                    
-                    # 如果差异超过阈值，认为是场景变化
-                    if mean_diff > threshold:
-                        timestamp = frame_count / fps
-                        scene_changes.append(timestamp)
-                        logger.debug(f"Scene change detected at {timestamp:.2f}s (diff: {mean_diff:.2f})")
-                
-                prev_frame = gray
+
+                # 跳帧处理
+                if frame_count % frame_skip == 0:
+                    # 转换为灰度图并缩小尺寸以提高性能
+                    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+                    gray = cv2.resize(gray, (320, 240))
+
+                    if prev_frame is not None:
+                        # 计算帧间差异
+                        diff = cv2.absdiff(prev_frame, gray)
+                        mean_diff = np.mean(diff)
+
+                        # 如果差异超过阈值，认为是场景变化
+                        if mean_diff > threshold:
+                            timestamp = frame_count / fps
+                            # 避免过于接近的场景变化点
+                            if not scene_changes or timestamp - scene_changes[-1] > 1.0:
+                                scene_changes.append(timestamp)
+                                logger.debug(f"Scene change detected at {timestamp:.2f}s (diff: {mean_diff:.2f})")
+
+                    prev_frame = gray
+
                 frame_count += 1
-            
+
             cap.release()
-            
+
             # 添加结束时间
             duration = frame_count / fps if fps > 0 else 0
-            if duration > 0:
+            if duration > 0 and (not scene_changes or duration - scene_changes[-1] > 0.5):
                 scene_changes.append(duration)
-            
-            logger.info(f"Detected {len(scene_changes)-1} scene changes in video")
+
+            logger.info(f"OpenCV detected {len(scene_changes)-1} scene changes in video")
             return scene_changes
-            
+
         except Exception as e:
-            logger.error(f"Failed to detect scene changes: {e}")
-            return [0.0]
+            logger.error(f"Failed to detect scene changes with OpenCV: {e}")
+            # 返回基本的开始和结束时间
+            try:
+                video_info = self._get_video_info(file_path)
+                duration = video_info.get('duration', 0)
+                return [0.0, duration] if duration > 0 else [0.0]
+            except:
+                return [0.0]
     
     def _generate_thumbnail(self, video_path: str, timestamp: float, output_path: str) -> bool:
         """生成视频缩略图"""

scripts/install_scenedetect.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""
+安装PySceneDetect的脚本
+用于改进视频场景检测功能
+"""
+
+import subprocess
+import sys
+import os
+from pathlib import Path
+
+def install_scenedetect():
+    """安装PySceneDetect库"""
+    print("🎬 安装PySceneDetect...")
+    
+    try:
+        # 尝试导入，检查是否已安装
+        import scenedetect
+        print("✅ PySceneDetect已经安装")
+        print(f"版本: {scenedetect.__version__}")
+        return True
+    except ImportError:
+        print("📦 PySceneDetect未安装，开始安装...")
+    
+    try:
+        # 安装PySceneDetect
+        subprocess.check_call([
+            sys.executable, '-m', 'pip', 'install', 'scenedetect[opencv]'
+        ])
+        
+        print("✅ PySceneDetect安装成功！")
+        
+        # 验证安装
+        try:
+            import scenedetect
+            print(f"✅ 验证成功，版本: {scenedetect.__version__}")
+            return True
+        except ImportError:
+            print("❌ 安装验证失败")
+            return False
+            
+    except subprocess.CalledProcessError as e:
+        print(f"❌ 安装失败: {e}")
+        return False
+    except Exception as e:
+        print(f"❌ 安装过程中出现错误: {e}")
+        return False
+
+def test_scenedetect():
+    """测试PySceneDetect功能"""
+    print("\n🧪 测试PySceneDetect功能...")
+    
+    try:
+        from scenedetect import VideoManager, SceneManager
+        from scenedetect.detectors import ContentDetector
+        
+        print("✅ 导入成功")
+        
+        # 查找测试视频文件
+        project_root = Path(__file__).parent.parent
+        test_video_paths = [
+            project_root / "assets" / "templates" / "template1" / "resources" / "video1.mp4",
+            project_root / "assets" / "templates" / "template2" / "resources" / "video1.mp4",
+            "/root/.mixvideo/temp/video_segments",
+        ]
+        
+        test_video = None
+        for path in test_video_paths:
+            if isinstance(path, str):
+                # 查找目录中的视频文件
+                if os.path.exists(path):
+                    for file in os.listdir(path):
+                        if file.lower().endswith(('.mp4', '.avi', '.mov')):
+                            test_video = os.path.join(path, file)
+                            break
+            else:
+                if path.exists():
+                    test_video = str(path)
+                    break
+        
+        if test_video:
+            print(f"🎬 使用测试视频: {test_video}")
+            
+            # 创建视频管理器
+            video_manager = VideoManager([test_video])
+            scene_manager = SceneManager()
+            
+            # 添加检测器
+            scene_manager.add_detector(ContentDetector(threshold=30.0))
+            
+            # 开始检测
+            video_manager.start()
+            scene_manager.detect_scenes(frame_source=video_manager)
+            
+            # 获取结果
+            scene_list = scene_manager.get_scene_list()
+            
+            print(f"✅ 检测到 {len(scene_list)} 个场景")
+            
+            if scene_list:
+                print("📋 场景列表:")
+                for i, scene in enumerate(scene_list[:5]):  # 只显示前5个
+                    start = scene[0].get_seconds()
+                    end = scene[1].get_seconds()
+                    print(f"  场景 {i+1}: {start:.2f}s - {end:.2f}s")
+            
+            video_manager.release()
+            print("✅ PySceneDetect功能测试成功！")
+            return True
+        else:
+            print("⚠️ 没有找到测试视频文件，跳过功能测试")
+            return True
+            
+    except Exception as e:
+        print(f"❌ 功能测试失败: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def show_usage_info():
+    """显示使用信息"""
+    print("\n📖 PySceneDetect使用信息:")
+    print("1. PySceneDetect是一个专业的视频场景检测库")
+    print("2. 相比OpenCV的简单帧差检测，它提供更准确的场景分割")
+    print("3. 支持多种检测算法：ContentDetector、ThresholdDetector等")
+    print("4. 可以自动检测淡入淡出、硬切等转场类型")
+    
+    print("\n🔧 配置参数:")
+    print("- threshold: 场景变化敏感度 (默认30.0)")
+    print("  - 较低值 (10-20): 更敏感，检测更多场景变化")
+    print("  - 较高值 (40-50): 不太敏感，只检测明显的场景变化")
+    
+    print("\n📚 更多信息:")
+    print("- 官方文档: https://pyscenedetect.readthedocs.io/")
+    print("- GitHub: https://github.com/Breakthrough/PySceneDetect")
+
+def main():
+    """主函数"""
+    print("🎬 PySceneDetect安装和测试工具")
+    print("=" * 50)
+    
+    # 1. 安装PySceneDetect
+    if not install_scenedetect():
+        print("❌ 安装失败，退出")
+        return
+    
+    # 2. 测试功能
+    if not test_scenedetect():
+        print("❌ 功能测试失败")
+        return
+    
+    # 3. 显示使用信息
+    show_usage_info()
+    
+    print("\n✅ 所有步骤完成！")
+    print("\n🚀 现在可以使用改进的场景检测功能了：")
+    print("- 重新导入视频素材将使用PySceneDetect进行更准确的分镜")
+    print("- 如果PySceneDetect不可用，系统会自动回退到OpenCV方法")
+
+if __name__ == "__main__":
+    main()

scripts/test_scene_detection.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""
+测试场景检测功能的脚本
+比较PySceneDetect和OpenCV的检测效果
+"""
+
+import sys
+import os
+import time
+from pathlib import Path
+
+# 添加项目根目录到Python路径
+project_root = Path(__file__).parent.parent
+sys.path.insert(0, str(project_root))
+
+from python_core.services.media_manager import MediaManager
+
+def find_test_videos():
+    """查找测试视频文件"""
+    print("🔍 查找测试视频文件...")
+    
+    # 可能的视频文件位置
+    search_paths = [
+        Path("/root/.mixvideo/temp/video_segments"),
+        Path("/root/.mixvideo/temp/original_videos"),
+        project_root / "assets" / "templates",
+    ]
+    
+    video_files = []
+    
+    for search_path in search_paths:
+        if search_path.exists():
+            print(f"📁 搜索目录: {search_path}")
+            
+            if search_path.is_file() and search_path.suffix.lower() in ['.mp4', '.avi', '.mov', '.mkv']:
+                video_files.append(str(search_path))
+            elif search_path.is_dir():
+                # 递归查找视频文件
+                for video_file in search_path.rglob("*"):
+                    if video_file.is_file() and video_file.suffix.lower() in ['.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv']:
+                        video_files.append(str(video_file))
+                        if len(video_files) >= 5:  # 限制数量
+                            break
+    
+    print(f"📊 找到 {len(video_files)} 个视频文件")
+    for i, video in enumerate(video_files[:5]):
+        file_size = Path(video).stat().st_size / 1024 / 1024
+        print(f"  {i+1}. {Path(video).name} ({file_size:.1f} MB)")
+    
+    return video_files[:3]  # 只返回前3个用于测试
+
+def test_scene_detection_methods(video_path: str):
+    """测试不同的场景检测方法"""
+    print(f"\n🎬 测试视频: {Path(video_path).name}")
+    print("-" * 50)
+    
+    media_manager = MediaManager()
+    
+    # 获取视频基本信息
+    try:
+        video_info = media_manager._get_video_info(video_path)
+        print(f"📊 视频信息:")
+        print(f"   时长: {video_info['duration']:.2f} 秒")
+        print(f"   分辨率: {video_info['width']}x{video_info['height']}")
+        print(f"   帧率: {video_info['fps']:.2f} fps")
+        print(f"   文件大小: {video_info['file_size']/1024/1024:.1f} MB")
+    except Exception as e:
+        print(f"❌ 获取视频信息失败: {e}")
+        return
+    
+    # 测试不同的阈值
+    thresholds = [20.0, 30.0, 40.0]
+    
+    for threshold in thresholds:
+        print(f"\n🔧 测试阈值: {threshold}")
+        
+        # 测试PySceneDetect方法
+        try:
+            start_time = time.time()
+            scene_changes = media_manager._detect_scene_changes(video_path, threshold)
+            pyscene_time = time.time() - start_time
+            
+            print(f"✅ PySceneDetect: {len(scene_changes)-1} 个场景变化 (耗时: {pyscene_time:.2f}s)")
+            if scene_changes:
+                print(f"   时间点: {[f'{t:.2f}s' for t in scene_changes[:10]]}")  # 只显示前10个
+        except Exception as e:
+            print(f"❌ PySceneDetect失败: {e}")
+        
+        # 测试OpenCV方法
+        try:
+            start_time = time.time()
+            opencv_changes = media_manager._detect_scene_changes_opencv(video_path, threshold)
+            opencv_time = time.time() - start_time
+            
+            print(f"✅ OpenCV: {len(opencv_changes)-1} 个场景变化 (耗时: {opencv_time:.2f}s)")
+            if opencv_changes:
+                print(f"   时间点: {[f'{t:.2f}s' for t in opencv_changes[:10]]}")  # 只显示前10个
+        except Exception as e:
+            print(f"❌ OpenCV失败: {e}")
+
+def test_scene_detection_accuracy():
+    """测试场景检测的准确性"""
+    print("\n🎯 场景检测准确性测试")
+    print("=" * 50)
+    
+    video_files = find_test_videos()
+    
+    if not video_files:
+        print("❌ 没有找到测试视频文件")
+        print("\n💡 建议:")
+        print("1. 先导入一些视频素材")
+        print("2. 或者将测试视频文件放到以下目录:")
+        print("   - /root/.mixvideo/temp/video_segments/")
+        print("   - /root/.mixvideo/temp/original_videos/")
+        return
+    
+    for video_path in video_files:
+        try:
+            test_scene_detection_methods(video_path)
+        except Exception as e:
+            print(f"❌ 测试视频 {video_path} 失败: {e}")
+            continue
+
+def benchmark_performance():
+    """性能基准测试"""
+    print("\n⚡ 性能基准测试")
+    print("=" * 50)
+    
+    video_files = find_test_videos()
+    
+    if not video_files:
+        print("❌ 没有找到测试视频文件")
+        return
+    
+    media_manager = MediaManager()
+    
+    # 选择一个中等大小的视频进行测试
+    test_video = None
+    for video in video_files:
+        try:
+            size = Path(video).stat().st_size / 1024 / 1024  # MB
+            if 5 < size < 50:  # 5-50MB的视频
+                test_video = video
+                break
+        except:
+            continue
+    
+    if not test_video:
+        test_video = video_files[0]  # 使用第一个视频
+    
+    print(f"🎬 性能测试视频: {Path(test_video).name}")
+    
+    methods = [
+        ("PySceneDetect", media_manager._detect_scene_changes),
+        ("OpenCV", media_manager._detect_scene_changes_opencv)
+    ]
+    
+    for method_name, method_func in methods:
+        try:
+            print(f"\n🔧 测试 {method_name}...")
+            
+            # 多次运行取平均值
+            times = []
+            results = []
+            
+            for i in range(3):
+                start_time = time.time()
+                scene_changes = method_func(test_video, 30.0)
+                end_time = time.time()
+                
+                times.append(end_time - start_time)
+                results.append(len(scene_changes) - 1)
+                
+                print(f"   运行 {i+1}: {end_time - start_time:.2f}s, {len(scene_changes)-1} 个场景")
+            
+            avg_time = sum(times) / len(times)
+            avg_scenes = sum(results) / len(results)
+            
+            print(f"✅ {method_name} 平均性能:")
+            print(f"   平均耗时: {avg_time:.2f}s")
+            print(f"   平均场景数: {avg_scenes:.1f}")
+            
+        except Exception as e:
+            print(f"❌ {method_name} 性能测试失败: {e}")
+
+def show_recommendations():
+    """显示使用建议"""
+    print("\n💡 使用建议")
+    print("=" * 50)
+    
+    print("🎯 选择合适的阈值:")
+    print("- 阈值 10-20: 高敏感度，适合检测细微的场景变化")
+    print("- 阈值 25-35: 中等敏感度，适合大多数情况 (推荐)")
+    print("- 阈值 40-50: 低敏感度，只检测明显的场景变化")
+    
+    print("\n⚡ 性能考虑:")
+    print("- PySceneDetect: 更准确，但处理时间较长")
+    print("- OpenCV: 处理速度快，但准确性较低")
+    print("- 建议: 对重要视频使用PySceneDetect，批量处理时使用OpenCV")
+    
+    print("\n🔧 优化建议:")
+    print("- 对于长视频，可以先用低敏感度快速分割")
+    print("- 然后对重要片段使用高敏感度精细分割")
+    print("- 定期检查分割结果，调整阈值参数")
+
+def main():
+    """主函数"""
+    print("🎬 场景检测功能测试")
+    print("=" * 50)
+    
+    try:
+        # 1. 测试场景检测准确性
+        test_scene_detection_accuracy()
+        
+        # 2. 性能基准测试
+        benchmark_performance()
+        
+        # 3. 显示使用建议
+        show_recommendations()
+        
+        print("\n✅ 测试完成！")
+        
+    except Exception as e:
+        print(f"❌ 测试过程中出现错误: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    main()