q792602257/FrameTour-RenderWorker
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

basic cpu encode support

Browse Source
This commit is contained in:
Jerry Yan
2024-11-27 16:47:39 +08:00
parent a8abb92b84
commit 9139727dc6
3 changed files with 136 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
biz/ffmpeg.py
View File

@ -1,3 +1,5 @@
import os.path
from entity.ffmpeg import FfmpegTask
import logging
@ -10,32 +12,32 @@ def parse_ffmpeg_task(task_info, template_info):
tasks = []
# 中间片段
for part in template_info.get("video_parts"):
sub_ffmpeg_task = parse_video_part(part, task_info)
if not sub_ffmpeg_task:
source = select_video_if_needed(part.get('source'), task_info, template_info)
if not source:
logger.warning("no video found for part: " + str(part))
continue
sub_ffmpeg_task = FfmpegTask(source)
sub_ffmpeg_task.frame_rate = template_info.get("frame_rate", 25)
for lut in part.get('filters', []):
sub_ffmpeg_task.add_lut(os.path.join(template_info.get("local_path"), lut))
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)
task = FfmpegTask(tasks, output_file="test.mp4")
task.correct_task_type()
overall = template_info.get("overall_template")
task.add_lut(*overall.get('luts', []))
task.add_audios(*overall.get('audios', []))
task.add_overlay(*overall.get('overlays', []))
task.frame_rate = template_info.get("frame_rate", 25)
for lut in overall.get('filters', []):
task.add_lut(os.path.join(template_info.get("local_path"), lut))
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
def parse_video_part(video_part, task_info):
source = select_video_if_needed(video_part.get('source'), task_info)
if not source:
logger.warning("no video found for part: " + str(video_part))
return None
task = FfmpegTask(source)
task.add_lut(*video_part.get('luts', []))
task.add_audios(*video_part.get('audios', []))
task.add_overlay(*video_part.get('overlays', []))
return task
def select_video_if_needed(source, task_info):
def select_video_if_needed(source, task_info, template_info):
if source.startswith('PLACEHOLDER_'):
placeholder_id = source.replace('PLACEHOLDER_', '')
new_sources = task_info.get('user_videos', {}).get(placeholder_id, [])
@ -47,7 +49,7 @@ def select_video_if_needed(source, task_info):
# TODO: Random Pick / Policy Pick
new_sources = new_sources[0]
return new_sources
return source
return os.path.join(template_info.get("local_path"), source)
def start_ffmpeg_task(ffmpeg_task):