You've already forked FrameTour-RenderWorker
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Type, Any, Optional
|
|
import json
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class EffectProcessor(ABC):
|
|
"""效果处理器抽象基类"""
|
|
|
|
def __init__(self, params: str = "", ext_data: Optional[Dict[str, Any]] = None):
|
|
self.params = params
|
|
self.ext_data = ext_data or {}
|
|
self.frame_rate = 25 # 默认帧率
|
|
|
|
@abstractmethod
|
|
def validate_params(self) -> bool:
|
|
"""验证参数是否有效"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def generate_filter_args(self, video_input: str, effect_index: int) -> tuple[List[str], str]:
|
|
"""
|
|
生成FFmpeg滤镜参数
|
|
|
|
Args:
|
|
video_input: 输入视频流标识符 (例如: "[0:v]", "[v_eff1]")
|
|
effect_index: 效果索引，用于生成唯一的输出标识符
|
|
|
|
Returns:
|
|
tuple: (filter_args_list, output_stream_identifier)
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_effect_name(self) -> str:
|
|
"""获取效果名称"""
|
|
pass
|
|
|
|
def parse_params(self) -> List[str]:
|
|
"""解析参数字符串为列表"""
|
|
if not self.params:
|
|
return []
|
|
return self.params.split(',')
|
|
|
|
def get_pos_json(self) -> Dict[str, Any]:
|
|
"""获取位置JSON数据"""
|
|
pos_json_str = self.ext_data.get('posJson', '{}')
|
|
try:
|
|
return json.loads(pos_json_str) if pos_json_str != '{}' else {}
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse posJson: {e}")
|
|
return {}
|
|
|
|
class EffectRegistry:
|
|
"""效果处理器注册表"""
|
|
|
|
def __init__(self):
|
|
self._processors: Dict[str, Type[EffectProcessor]] = {}
|
|
|
|
def register(self, name: str, processor_class: Type[EffectProcessor]):
|
|
"""注册效果处理器"""
|
|
if not issubclass(processor_class, EffectProcessor):
|
|
raise ValueError(f"{processor_class} must be a subclass of EffectProcessor")
|
|
self._processors[name] = processor_class
|
|
logger.debug(f"Registered effect processor: {name}")
|
|
|
|
def get_processor(self, effect_name: str, params: str = "", ext_data: Optional[Dict[str, Any]] = None) -> Optional[EffectProcessor]:
|
|
"""获取效果处理器实例"""
|
|
if effect_name not in self._processors:
|
|
logger.warning(f"Unknown effect: {effect_name}")
|
|
return None
|
|
|
|
processor_class = self._processors[effect_name]
|
|
return processor_class(params, ext_data)
|
|
|
|
def list_effects(self) -> List[str]:
|
|
"""列出所有注册的效果"""
|
|
return list(self._processors.keys())
|
|
|
|
def parse_effect_string(self, effect_string: str) -> tuple[str, str]:
|
|
"""
|
|
解析效果字符串
|
|
|
|
Args:
|
|
effect_string: 效果字符串，格式为 "effect_name:params"
|
|
|
|
Returns:
|
|
tuple: (effect_name, params)
|
|
"""
|
|
if ':' in effect_string:
|
|
parts = effect_string.split(':', 2)
|
|
return parts[0], parts[1] if len(parts) > 1 else ""
|
|
return effect_string, "" |