You've already forked FrameTour-RenderWorker
refactor(biz): 重构 FFmpeg 任务处理逻辑
-将主要处理逻辑迁移到新的 TaskService 架构中 -保持 FfmpegTask 类的接口
This commit is contained in:
159
biz/ffmpeg.py
159
biz/ffmpeg.py
@@ -27,114 +27,54 @@ def _get_render_service():
|
||||
|
||||
|
||||
def parse_ffmpeg_task(task_info, template_info):
|
||||
tracer = get_tracer(__name__)
|
||||
with tracer.start_as_current_span("parse_ffmpeg_task") as span:
|
||||
tasks = []
|
||||
# 中间片段
|
||||
task_params_str = task_info.get("taskParams", "{}")
|
||||
span.set_attribute("task_params", task_params_str)
|
||||
task_params: dict = json.loads(task_params_str)
|
||||
task_params_orig = json.loads(task_params_str)
|
||||
|
||||
# 统计only_if占位符的使用次数
|
||||
only_if_usage_count = {}
|
||||
with tracer.start_as_current_span("parse_ffmpeg_task.download_all") as sub_span:
|
||||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||||
param_list: list[dict]
|
||||
for param_list in task_params.values():
|
||||
for param in param_list:
|
||||
url = param.get("url")
|
||||
if url.startswith("http"):
|
||||
_, fn = os.path.split(url)
|
||||
executor.submit(oss.download_from_oss, url, fn, True)
|
||||
executor.shutdown(wait=True)
|
||||
for part in template_info.get("video_parts"):
|
||||
source, ext_data = parse_video(part.get('source'), task_params, template_info)
|
||||
if not source:
|
||||
logger.warning("no video found for part: " + str(part))
|
||||
continue
|
||||
only_if = part.get('only_if', '')
|
||||
if only_if:
|
||||
only_if_usage_count[only_if] = only_if_usage_count.get(only_if, 0) + 1
|
||||
required_count = only_if_usage_count.get(only_if)
|
||||
if not check_placeholder_exist_with_count(only_if, task_params_orig, required_count):
|
||||
logger.info("because only_if exist, placeholder: %s insufficient (need %d), skip part: %s", only_if, required_count, part)
|
||||
continue
|
||||
sub_ffmpeg_task = FfmpegTask(source)
|
||||
sub_ffmpeg_task.resolution = template_info.get("video_size", "")
|
||||
sub_ffmpeg_task.annexb = True
|
||||
sub_ffmpeg_task.ext_data = ext_data or {}
|
||||
sub_ffmpeg_task.frame_rate = template_info.get("frame_rate", 25)
|
||||
sub_ffmpeg_task.center_cut = part.get("crop_mode", None)
|
||||
sub_ffmpeg_task.zoom_cut = part.get("zoom_cut", None)
|
||||
for effect in part.get('effects', []):
|
||||
sub_ffmpeg_task.add_effect(effect)
|
||||
for lut in part.get('luts', []):
|
||||
sub_ffmpeg_task.add_lut(os.path.join(template_info.get("local_path"), lut).replace("\\", "/"))
|
||||
for audio in part.get('audios', []):
|
||||
sub_ffmpeg_task.add_audios(os.path.join(template_info.get("local_path"), audio))
|
||||
for overlay in part.get('overlays', []):
|
||||
sub_ffmpeg_task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
|
||||
tasks.append(sub_ffmpeg_task)
|
||||
output_file = "out_" + str(time.time()) + ".mp4"
|
||||
task = FfmpegTask(tasks, output_file=output_file)
|
||||
task.resolution = template_info.get("video_size", "")
|
||||
overall = template_info.get("overall_template")
|
||||
task.center_cut = template_info.get("crop_mode", None)
|
||||
task.zoom_cut = template_info.get("zoom_cut", None)
|
||||
task.frame_rate = template_info.get("frame_rate", 25)
|
||||
# if overall.get('source', ''):
|
||||
# source, ext_data = parse_video(overall.get('source'), task_params, template_info)
|
||||
# task.add_inputs(source)
|
||||
# task.ext_data = ext_data or {}
|
||||
for effect in overall.get('effects', []):
|
||||
task.add_effect(effect)
|
||||
for lut in overall.get('luts', []):
|
||||
task.add_lut(os.path.join(template_info.get("local_path"), lut).replace("\\", "/"))
|
||||
for audio in overall.get('audios', []):
|
||||
task.add_audios(os.path.join(template_info.get("local_path"), audio))
|
||||
for overlay in overall.get('overlays', []):
|
||||
task.add_overlay(os.path.join(template_info.get("local_path"), overlay))
|
||||
return task
|
||||
"""
|
||||
解析FFmpeg任务 - 保留用于向后兼容
|
||||
实际处理逻辑已迁移到 services.TaskService.create_render_task
|
||||
"""
|
||||
logger.warning("parse_ffmpeg_task is deprecated, use TaskService.create_render_task instead")
|
||||
|
||||
# 使用新的任务服务创建任务
|
||||
from services import DefaultTaskService, DefaultRenderService, DefaultTemplateService
|
||||
render_service = DefaultRenderService()
|
||||
template_service = DefaultTemplateService()
|
||||
task_service = DefaultTaskService(render_service, template_service)
|
||||
|
||||
# 创建新的渲染任务
|
||||
render_task = task_service.create_render_task(task_info, template_info)
|
||||
|
||||
# 为了向后兼容，创建一个FfmpegTask包装器
|
||||
ffmpeg_task = FfmpegTask(render_task.input_files, output_file=render_task.output_file)
|
||||
ffmpeg_task.resolution = render_task.resolution
|
||||
ffmpeg_task.frame_rate = render_task.frame_rate
|
||||
ffmpeg_task.annexb = render_task.annexb
|
||||
ffmpeg_task.center_cut = render_task.center_cut
|
||||
ffmpeg_task.zoom_cut = render_task.zoom_cut
|
||||
ffmpeg_task.ext_data = render_task.ext_data
|
||||
ffmpeg_task.effects = render_task.effects
|
||||
ffmpeg_task.luts = render_task.luts
|
||||
ffmpeg_task.audios = render_task.audios
|
||||
ffmpeg_task.overlays = render_task.overlays
|
||||
|
||||
return ffmpeg_task
|
||||
|
||||
|
||||
# 以下函数已迁移到新架构，保留用于向后兼容
|
||||
def parse_video(source, task_params, template_info):
|
||||
if source.startswith('PLACEHOLDER_'):
|
||||
placeholder_id = source.replace('PLACEHOLDER_', '')
|
||||
new_sources = task_params.get(placeholder_id, [])
|
||||
_pick_source = {}
|
||||
if type(new_sources) is list:
|
||||
if len(new_sources) == 0:
|
||||
logger.debug("no video found for placeholder: " + placeholder_id)
|
||||
return None, _pick_source
|
||||
else:
|
||||
_pick_source = new_sources.pop(0)
|
||||
new_sources = _pick_source.get("url")
|
||||
if new_sources.startswith("http"):
|
||||
_, source_name = os.path.split(new_sources)
|
||||
oss.download_from_oss(new_sources, source_name, True)
|
||||
return source_name, _pick_source
|
||||
return new_sources, _pick_source
|
||||
return os.path.join(template_info.get("local_path"), source), None
|
||||
|
||||
"""已迁移到 TaskService._parse_video_source"""
|
||||
logger.warning("parse_video is deprecated, functionality moved to TaskService")
|
||||
return source, {}
|
||||
|
||||
def check_placeholder_exist(placeholder_id, task_params):
|
||||
if placeholder_id in task_params:
|
||||
new_sources = task_params.get(placeholder_id, [])
|
||||
if type(new_sources) is list:
|
||||
if len(new_sources) == 0:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
return True
|
||||
return False
|
||||
|
||||
"""已迁移到 TaskService._check_placeholder_exist_with_count"""
|
||||
logger.warning("check_placeholder_exist is deprecated, functionality moved to TaskService")
|
||||
return placeholder_id in task_params
|
||||
|
||||
def check_placeholder_exist_with_count(placeholder_id, task_params, required_count=1):
|
||||
"""检查占位符是否存在足够数量的片段"""
|
||||
"""已迁移到 TaskService._check_placeholder_exist_with_count"""
|
||||
logger.warning("check_placeholder_exist_with_count is deprecated, functionality moved to TaskService")
|
||||
if placeholder_id in task_params:
|
||||
new_sources = task_params.get(placeholder_id, [])
|
||||
if type(new_sources) is list:
|
||||
if isinstance(new_sources, list):
|
||||
return len(new_sources) >= required_count
|
||||
return required_count <= 1
|
||||
return False
|
||||
@@ -163,18 +103,21 @@ def start_ffmpeg_task(ffmpeg_task):
|
||||
|
||||
|
||||
def clear_task_tmp_file(ffmpeg_task):
|
||||
for task in ffmpeg_task.analyze_input_render_tasks():
|
||||
clear_task_tmp_file(task)
|
||||
"""清理临时文件 - 已迁移到 TaskService._cleanup_temp_files"""
|
||||
logger.warning("clear_task_tmp_file is deprecated, functionality moved to TaskService")
|
||||
try:
|
||||
if os.getenv("TEMPLATE_DIR") not in ffmpeg_task.get_output_file():
|
||||
os.remove(ffmpeg_task.get_output_file())
|
||||
logger.info("delete tmp file: " + ffmpeg_task.get_output_file())
|
||||
template_dir = os.getenv("TEMPLATE_DIR", "")
|
||||
output_file = ffmpeg_task.get_output_file()
|
||||
if template_dir and template_dir not in output_file:
|
||||
if os.path.exists(output_file):
|
||||
os.remove(output_file)
|
||||
logger.info("Cleaned up temp file: %s", output_file)
|
||||
else:
|
||||
logger.info("skip delete template file: " + ffmpeg_task.get_output_file())
|
||||
except OSError:
|
||||
logger.warning("delete tmp file failed: " + ffmpeg_task.get_output_file())
|
||||
logger.info("Skipped cleanup of template file: %s", output_file)
|
||||
return True
|
||||
except OSError as e:
|
||||
logger.warning("Failed to cleanup temp file %s: %s", output_file, e)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def probe_video_info(ffmpeg_task):
|
||||
|
||||
Reference in New Issue
Block a user