fix: 重构
This commit is contained in:
root
@@ -17,24 +17,369 @@ from python_core.config import settings
|
||||
from python_core.utils.logger import logger
|
||||
from python_core.utils.jsonrpc import create_response_handler
|
||||
|
||||
# 视频处理库
|
||||
try:
|
||||
import cv2
|
||||
import numpy as np
|
||||
VIDEO_LIBS_AVAILABLE = True
|
||||
except ImportError:
|
||||
logger.warning("Video processing libraries not available. Install opencv-python for full functionality.")
|
||||
VIDEO_LIBS_AVAILABLE = False
|
||||
# 依赖管理器 - 遵循SOLID原则
|
||||
class DependencyManager:
|
||||
"""依赖管理器 - 单一职责原则(SRP): 只负责管理依赖"""
|
||||
|
||||
# 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
|
||||
def __init__(self):
|
||||
self._dependencies = {}
|
||||
self._initialize_dependencies()
|
||||
|
||||
def _initialize_dependencies(self):
|
||||
"""初始化所有依赖"""
|
||||
self._check_opencv_dependency()
|
||||
self._check_scenedetect_dependency()
|
||||
|
||||
def _check_opencv_dependency(self):
|
||||
"""检查OpenCV依赖"""
|
||||
try:
|
||||
import cv2
|
||||
import numpy as np
|
||||
self._dependencies['opencv'] = {
|
||||
'available': True,
|
||||
'modules': {'cv2': cv2, 'numpy': np},
|
||||
'version': cv2.__version__
|
||||
}
|
||||
logger.info(f"OpenCV {cv2.__version__} is available")
|
||||
except ImportError as e:
|
||||
self._dependencies['opencv'] = {
|
||||
'available': False,
|
||||
'error': str(e),
|
||||
'modules': {}
|
||||
}
|
||||
logger.warning("OpenCV not available. Install opencv-python for full functionality.")
|
||||
|
||||
def _check_scenedetect_dependency(self):
|
||||
"""检查PySceneDetect依赖"""
|
||||
try:
|
||||
from scenedetect import VideoManager, SceneManager
|
||||
from scenedetect.detectors import ContentDetector
|
||||
import scenedetect
|
||||
|
||||
self._dependencies['scenedetect'] = {
|
||||
'available': True,
|
||||
'modules': {
|
||||
'VideoManager': VideoManager,
|
||||
'SceneManager': SceneManager,
|
||||
'ContentDetector': ContentDetector
|
||||
},
|
||||
'version': scenedetect.__version__
|
||||
}
|
||||
logger.info(f"PySceneDetect {scenedetect.__version__} is available for advanced scene detection")
|
||||
except ImportError as e:
|
||||
self._dependencies['scenedetect'] = {
|
||||
'available': False,
|
||||
'error': str(e),
|
||||
'modules': {}
|
||||
}
|
||||
logger.warning("PySceneDetect not available. Install scenedetect for better scene detection.")
|
||||
|
||||
def is_available(self, dependency_name: str) -> bool:
|
||||
"""检查依赖是否可用 - 开闭原则(OCP): 对扩展开放"""
|
||||
return self._dependencies.get(dependency_name, {}).get('available', False)
|
||||
|
||||
def get_module(self, dependency_name: str, module_name: str):
|
||||
"""获取依赖模块"""
|
||||
if not self.is_available(dependency_name):
|
||||
raise ImportError(f"Dependency '{dependency_name}' is not available")
|
||||
|
||||
modules = self._dependencies[dependency_name]['modules']
|
||||
if module_name not in modules:
|
||||
raise ImportError(f"Module '{module_name}' not found in dependency '{dependency_name}'")
|
||||
|
||||
return modules[module_name]
|
||||
|
||||
def get_dependency_info(self, dependency_name: str) -> dict:
|
||||
"""获取依赖信息"""
|
||||
return self._dependencies.get(dependency_name, {})
|
||||
|
||||
def get_all_dependencies(self) -> dict:
|
||||
"""获取所有依赖信息"""
|
||||
return self._dependencies.copy()
|
||||
|
||||
# 抽象接口 - 接口隔离原则(ISP): 定义专门的接口
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
class VideoInfoExtractor(ABC):
|
||||
"""视频信息提取器接口"""
|
||||
|
||||
@abstractmethod
|
||||
def extract_video_info(self, file_path: str) -> Dict:
|
||||
"""提取视频信息"""
|
||||
pass
|
||||
|
||||
class SceneDetector(ABC):
|
||||
"""场景检测器接口"""
|
||||
|
||||
@abstractmethod
|
||||
def detect_scenes(self, file_path: str, threshold: float = 30.0) -> List[float]:
|
||||
"""检测场景变化点"""
|
||||
pass
|
||||
|
||||
class VideoProcessor(ABC):
|
||||
"""视频处理器接口"""
|
||||
|
||||
@abstractmethod
|
||||
def split_video(self, input_path: str, output_dir: str, scene_times: List[float]) -> List[str]:
|
||||
"""分割视频"""
|
||||
pass
|
||||
|
||||
# 具体实现类 - 单一职责原则(SRP): 每个类只负责一个功能
|
||||
|
||||
class FFProbeVideoInfoExtractor(VideoInfoExtractor):
|
||||
"""使用FFProbe提取视频信息"""
|
||||
|
||||
def extract_video_info(self, file_path: str) -> Dict:
|
||||
"""使用ffprobe获取准确的视频信息"""
|
||||
file_size = os.path.getsize(file_path)
|
||||
|
||||
try:
|
||||
import subprocess
|
||||
import json
|
||||
|
||||
cmd = [
|
||||
'ffprobe',
|
||||
'-v', 'quiet',
|
||||
'-print_format', 'json',
|
||||
'-show_format',
|
||||
'-show_streams',
|
||||
file_path
|
||||
]
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
|
||||
if result.returncode == 0:
|
||||
probe_data = json.loads(result.stdout)
|
||||
|
||||
video_stream = None
|
||||
for stream in probe_data.get('streams', []):
|
||||
if stream.get('codec_type') == 'video':
|
||||
video_stream = stream
|
||||
break
|
||||
|
||||
if video_stream:
|
||||
duration = float(probe_data.get('format', {}).get('duration', 0))
|
||||
width = int(video_stream.get('width', 0))
|
||||
height = int(video_stream.get('height', 0))
|
||||
|
||||
fps_str = video_stream.get('r_frame_rate', '0/1')
|
||||
if '/' in fps_str:
|
||||
num, den = fps_str.split('/')
|
||||
fps = float(num) / float(den) if float(den) != 0 else 0
|
||||
else:
|
||||
fps = float(fps_str)
|
||||
|
||||
frame_count = int(duration * fps) if fps > 0 else 0
|
||||
|
||||
logger.info(f"ffprobe video info: duration={duration:.2f}s, fps={fps:.2f}, resolution={width}x{height}")
|
||||
|
||||
return {
|
||||
'duration': duration,
|
||||
'width': width,
|
||||
'height': height,
|
||||
'fps': fps,
|
||||
'frame_count': frame_count,
|
||||
'file_size': file_size,
|
||||
'codec': video_stream.get('codec_name', 'unknown')
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"ffprobe failed: {e}")
|
||||
raise
|
||||
|
||||
raise Exception("Failed to extract video info with ffprobe")
|
||||
|
||||
class OpenCVVideoInfoExtractor(VideoInfoExtractor):
|
||||
"""使用OpenCV提取视频信息"""
|
||||
|
||||
def __init__(self, dependency_manager: DependencyManager):
|
||||
self.dependency_manager = dependency_manager
|
||||
|
||||
def extract_video_info(self, file_path: str) -> Dict:
|
||||
"""使用OpenCV获取视频信息"""
|
||||
if not self.dependency_manager.is_available('opencv'):
|
||||
raise Exception("OpenCV not available")
|
||||
|
||||
cv2 = self.dependency_manager.get_module('opencv', 'cv2')
|
||||
file_size = os.path.getsize(file_path)
|
||||
|
||||
try:
|
||||
cap = cv2.VideoCapture(file_path)
|
||||
|
||||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||||
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
||||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||||
duration = frame_count / fps if fps > 0 else 0
|
||||
|
||||
cap.release()
|
||||
|
||||
logger.info(f"OpenCV video info: duration={duration:.2f}s, fps={fps:.2f}, resolution={width}x{height}")
|
||||
|
||||
return {
|
||||
'duration': duration,
|
||||
'width': width,
|
||||
'height': height,
|
||||
'fps': fps,
|
||||
'frame_count': frame_count,
|
||||
'file_size': file_size,
|
||||
'codec': 'unknown'
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get video info with OpenCV: {e}")
|
||||
raise
|
||||
|
||||
class PySceneDetectSceneDetector(SceneDetector):
|
||||
"""使用PySceneDetect进行场景检测"""
|
||||
|
||||
def __init__(self, dependency_manager: DependencyManager):
|
||||
self.dependency_manager = dependency_manager
|
||||
|
||||
def detect_scenes(self, file_path: str, threshold: float = 30.0) -> List[float]:
|
||||
"""使用PySceneDetect检测场景变化点"""
|
||||
if not self.dependency_manager.is_available('scenedetect'):
|
||||
raise Exception("PySceneDetect not available")
|
||||
|
||||
VideoManager = self.dependency_manager.get_module('scenedetect', 'VideoManager')
|
||||
SceneManager = self.dependency_manager.get_module('scenedetect', 'SceneManager')
|
||||
ContentDetector = self.dependency_manager.get_module('scenedetect', 'ContentDetector')
|
||||
|
||||
try:
|
||||
video_manager = VideoManager([file_path])
|
||||
scene_manager = SceneManager()
|
||||
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()
|
||||
|
||||
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}")
|
||||
raise
|
||||
|
||||
class OpenCVSceneDetector(SceneDetector):
|
||||
"""使用OpenCV进行场景检测"""
|
||||
|
||||
def __init__(self, dependency_manager: DependencyManager):
|
||||
self.dependency_manager = dependency_manager
|
||||
|
||||
def detect_scenes(self, file_path: str, threshold: float = 30.0) -> List[float]:
|
||||
"""使用OpenCV检测场景变化点"""
|
||||
if not self.dependency_manager.is_available('opencv'):
|
||||
raise Exception("OpenCV not available")
|
||||
|
||||
cv2 = self.dependency_manager.get_module('opencv', 'cv2')
|
||||
np = self.dependency_manager.get_module('opencv', 'numpy')
|
||||
|
||||
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))
|
||||
|
||||
while True:
|
||||
ret, frame = cap.read()
|
||||
if not ret:
|
||||
break
|
||||
|
||||
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 and (not scene_changes or duration - scene_changes[-1] > 0.5):
|
||||
scene_changes.append(duration)
|
||||
|
||||
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 with OpenCV: {e}")
|
||||
raise
|
||||
|
||||
# 工厂类 - 依赖倒置原则(DIP): 依赖抽象而不是具体实现
|
||||
class VideoProcessorFactory:
|
||||
"""视频处理器工厂 - 单一职责原则(SRP): 只负责创建对象"""
|
||||
|
||||
def __init__(self, dependency_manager: DependencyManager):
|
||||
self.dependency_manager = dependency_manager
|
||||
|
||||
def create_video_info_extractor(self) -> VideoInfoExtractor:
|
||||
"""创建视频信息提取器 - 开闭原则(OCP): 对扩展开放"""
|
||||
# 优先使用FFProbe，回退到OpenCV
|
||||
try:
|
||||
return FFProbeVideoInfoExtractor()
|
||||
except Exception:
|
||||
if self.dependency_manager.is_available('opencv'):
|
||||
return OpenCVVideoInfoExtractor(self.dependency_manager)
|
||||
else:
|
||||
raise Exception("No video info extractor available")
|
||||
|
||||
def create_scene_detector(self) -> SceneDetector:
|
||||
"""创建场景检测器"""
|
||||
# 优先使用PySceneDetect，回退到OpenCV
|
||||
if self.dependency_manager.is_available('scenedetect'):
|
||||
try:
|
||||
return PySceneDetectSceneDetector(self.dependency_manager)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if self.dependency_manager.is_available('opencv'):
|
||||
return OpenCVSceneDetector(self.dependency_manager)
|
||||
else:
|
||||
raise Exception("No scene detector available")
|
||||
|
||||
# 全局依赖管理器实例
|
||||
dependency_manager = DependencyManager()
|
||||
video_processor_factory = VideoProcessorFactory(dependency_manager)
|
||||
|
||||
# 向后兼容的全局变量 - 里氏替换原则(LSP): 保持接口兼容
|
||||
VIDEO_LIBS_AVAILABLE = dependency_manager.is_available('opencv')
|
||||
SCENEDETECT_AVAILABLE = dependency_manager.is_available('scenedetect')
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -87,27 +432,71 @@ class OriginalVideo:
|
||||
|
||||
|
||||
class MediaManager:
|
||||
"""媒体库管理器"""
|
||||
|
||||
def __init__(self):
|
||||
"""媒体库管理器
|
||||
|
||||
遵循SOLID原则:
|
||||
- SRP: 只负责媒体文件的管理和协调
|
||||
- OCP: 通过依赖注入支持扩展
|
||||
- LSP: 可以替换不同的处理器实现
|
||||
- ISP: 使用专门的接口
|
||||
- DIP: 依赖抽象接口而不是具体实现
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
dependency_manager: DependencyManager = None,
|
||||
video_info_extractor: VideoInfoExtractor = None,
|
||||
scene_detector: SceneDetector = None):
|
||||
"""
|
||||
初始化媒体管理器
|
||||
|
||||
Args:
|
||||
dependency_manager: 依赖管理器
|
||||
video_info_extractor: 视频信息提取器
|
||||
scene_detector: 场景检测器
|
||||
"""
|
||||
self.cache_dir = settings.temp_dir / "cache"
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
# 数据文件
|
||||
self.segments_file = self.cache_dir / "video_segments.json"
|
||||
self.original_videos_file = self.cache_dir / "original_videos.json"
|
||||
|
||||
|
||||
# 依赖注入 - 依赖倒置原则(DIP)
|
||||
self.dependency_manager = dependency_manager or globals()['dependency_manager']
|
||||
self.factory = VideoProcessorFactory(self.dependency_manager)
|
||||
|
||||
# 延迟初始化处理器 - 单一职责原则(SRP)
|
||||
self._video_info_extractor = video_info_extractor
|
||||
self._scene_detector = scene_detector
|
||||
|
||||
# 加载数据
|
||||
self.video_segments = self._load_video_segments()
|
||||
self.original_videos = self._load_original_videos()
|
||||
|
||||
|
||||
# 存储目录
|
||||
self.video_storage_dir = settings.temp_dir / "video_storage"
|
||||
|
||||
@property
|
||||
def video_info_extractor(self) -> VideoInfoExtractor:
|
||||
"""获取视频信息提取器 - 懒加载"""
|
||||
if self._video_info_extractor is None:
|
||||
self._video_info_extractor = self.factory.create_video_info_extractor()
|
||||
return self._video_info_extractor
|
||||
|
||||
@property
|
||||
def scene_detector(self) -> SceneDetector:
|
||||
"""获取场景检测器 - 懒加载"""
|
||||
if self._scene_detector is None:
|
||||
self._scene_detector = self.factory.create_scene_detector()
|
||||
return self._scene_detector
|
||||
|
||||
def _initialize_directories(self):
|
||||
"""初始化目录结构 - 单一职责原则(SRP)"""
|
||||
self.video_storage_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
self.segments_dir = settings.temp_dir / "video_segments"
|
||||
self.segments_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
self.thumbnails_dir = settings.temp_dir / "video_thumbnails"
|
||||
self.thumbnails_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -174,176 +563,38 @@ class MediaManager:
|
||||
return hash_md5.hexdigest()
|
||||
|
||||
def _get_video_info(self, file_path: str) -> Dict:
|
||||
"""获取视频基本信息"""
|
||||
if not VIDEO_LIBS_AVAILABLE:
|
||||
# 如果没有视频处理库，返回基本信息
|
||||
file_size = os.path.getsize(file_path)
|
||||
return {
|
||||
'duration': 0.0,
|
||||
'width': 0,
|
||||
'height': 0,
|
||||
'fps': 0.0,
|
||||
'frame_count': 0,
|
||||
'file_size': file_size
|
||||
}
|
||||
|
||||
"""获取视频基本信息 - 使用依赖注入的提取器"""
|
||||
try:
|
||||
cap = cv2.VideoCapture(file_path)
|
||||
|
||||
# 获取视频属性
|
||||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||||
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
||||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||||
duration = frame_count / fps if fps > 0 else 0
|
||||
|
||||
cap.release()
|
||||
|
||||
return {
|
||||
'duration': duration,
|
||||
'width': width,
|
||||
'height': height,
|
||||
'fps': fps,
|
||||
'frame_count': frame_count,
|
||||
'file_size': os.path.getsize(file_path)
|
||||
}
|
||||
return self.video_info_extractor.extract_video_info(file_path)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get video info: {e}")
|
||||
file_size = os.path.getsize(file_path)
|
||||
# 返回基本信息作为后备
|
||||
file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
|
||||
return {
|
||||
'duration': 0.0,
|
||||
'width': 0,
|
||||
'height': 0,
|
||||
'fps': 0.0,
|
||||
'frame_count': 0,
|
||||
'file_size': file_size
|
||||
'file_size': file_size,
|
||||
'codec': 'unknown'
|
||||
}
|
||||
|
||||
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 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
|
||||
|
||||
# 跳帧处理
|
||||
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 and (not scene_changes or duration - scene_changes[-1] > 0.5):
|
||||
scene_changes.append(duration)
|
||||
|
||||
logger.info(f"OpenCV detected {len(scene_changes)-1} scene changes in video")
|
||||
return scene_changes
|
||||
|
||||
return self.scene_detector.detect_scenes(file_path, threshold)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to detect scene changes with OpenCV: {e}")
|
||||
# 返回基本的开始和结束时间
|
||||
logger.error(f"Scene detection failed: {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:
|
||||
"""生成视频缩略图"""
|
||||
|
||||
Reference in New Issue
Block a user