修复

.env

@@ -1,4 +1,9 @@
 TEMPLATE_DIR=template/
-API_ENDPOINT=/task
-API_TOKEN=123456
-TEMP_DIR=tmp/
+API_ENDPOINT=https://zhentuai.com/task/v1
+ACCESS_KEY=TEST_ACCESS_KEY
+TEMP_DIR=tmp/
+#REDIRECT_TO_URL=https://worker-renderworker-re-kekuflqjxx.cn-shanghai.fcapp.run/
+# QSV
+ENCODER_ARGS="-c:v h264_qsv -global_quality 28 -look_ahead 1"
+# NVENC
+#ENCODER_ARGS="-c:v h264_nvenc -cq:v 24 -preset:v p7 -tune:v hq -profile:v high"

.gitignore

@@ -6,6 +6,11 @@ __pycache__/
 *.so
 .Python
 build/
+dist/
+*.mp4
+*.ts
+rand*.ts
+tmp_concat_*.txt
 *.egg-info/
 *.egg
 *.manifest

Dockerfile

@@ -0,0 +1,21 @@
+FROM linuxserver/ffmpeg:7.1.1
+LABEL authors="Jerry Yan"
+
+RUN sed -i 's@//.*archive.ubuntu.com@//mirrors.ustc.edu.cn@g' /etc/apt/sources.list && \
+    sed -i 's/security.ubuntu.com/mirrors.ustc.edu.cn/g' /etc/apt/sources.list
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+      python3-pip \
+      python3-dev \
+      python3-setuptools \
+      python3-wheel \
+      python3-venv
+RUN apt-get clean && \
+    rm -rf /var/lib/apt/lists/*
+COPY . /app/
+RUN python3 -m venv /app/venv
+RUN /app/venv/bin/python -m pip config set global.index-url https://mirrors.ustc.edu.cn/pypi/simple
+RUN /app/venv/bin/python -m pip install -r /app/requirements.txt
+
+WORKDIR /app
+ENTRYPOINT ["/app/venv/bin/python", "app.py"]

app.py

@@ -0,0 +1,40 @@
+import time
+
+import flask
+
+import config
+import biz.task
+import template
+from telemetry import init_opentelemetry
+from template import load_local_template
+from util import api
+
+load_local_template()
+import logging
+
+LOGGER = logging.getLogger(__name__)
+init_opentelemetry(batch=False)
+app = flask.Flask(__name__)
+
+@app.get('/health/check')
+def health_check():
+    return api.sync_center()
+
+@app.post('/')
+def do_nothing():
+    return "NOOP"
+
+@app.post('/<task_id>')
+def do_task(task_id):
+    task_info = api.get_task_info(task_id)
+    local_template_info = template.get_template_def(task_info.get("templateId"))
+    template_info = api.get_template_info(task_info.get("templateId"))
+    if local_template_info:
+        if local_template_info.get("updateTime") != template_info.get("updateTime"):
+            template.download_template(task_info.get("templateId"))
+    biz.task.start_task(task_info)
+    return "OK"
+
+
+if __name__ == '__main__':
+    app.run(host="0.0.0.0", port=9998)

biz/ffmpeg.py

@@ -1,58 +1,144 @@
+import json
 import os.path
+import time
+
+from opentelemetry.trace import Status, StatusCode
 
 from entity.ffmpeg import FfmpegTask
 import logging
 
-from util import ffmpeg
+from util import ffmpeg, oss
+from util.ffmpeg import fade_out_audio
+from telemetry import get_tracer
 
 logger = logging.getLogger('biz/ffmpeg')
 
 
 def parse_ffmpeg_task(task_info, template_info):
-    tasks = []
-    # 中间片段
-    for part in template_info.get("video_parts"):
-        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")
-    overall = template_info.get("overall_template")
-    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
+    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 = json.loads(task_params_str)
+        task_params_orig = json.loads(task_params_str)
+        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:
+                if not check_placeholder_exist(only_if, task_params_orig):
+                    logger.info("because only_if exist, placeholder: %s not exist, skip part: %s", only_if, 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)
+            for effect in part.get('effects', []):
+                sub_ffmpeg_task.add_effect(effect)
+            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)
+        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.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('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 select_video_if_needed(source, task_info, template_info):
+def parse_video(source, task_params, template_info):
     if source.startswith('PLACEHOLDER_'):
         placeholder_id = source.replace('PLACEHOLDER_', '')
-        new_sources = task_info.get('user_videos', {}).get(placeholder_id, [])
+        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
+                return None, _pick_source
             else:
-                # TODO: Random Pick / Policy Pick
-                new_sources = new_sources[0]
-        return new_sources
-    return os.path.join(template_info.get("local_path"), source)
+                _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)
+            return source_name, _pick_source
+        return new_sources, _pick_source
+    return os.path.join(template_info.get("local_path"), source), None
+
+
+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
 
 
 def start_ffmpeg_task(ffmpeg_task):
-   for task in ffmpeg_task.analyze_input_render_tasks():
-       start_ffmpeg_task(task)
-   ffmpeg.start_render(ffmpeg_task)
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_ffmpeg_task") as span:
+        for task in ffmpeg_task.analyze_input_render_tasks():
+            result = start_ffmpeg_task(task)
+            if not result:
+                return False
+        ffmpeg_task.correct_task_type()
+        span.set_attribute("task.type", ffmpeg_task.task_type)
+        span.set_attribute("task.center_cut", str(ffmpeg_task.center_cut))
+        span.set_attribute("task.frame_rate", ffmpeg_task.frame_rate)
+        span.set_attribute("task.resolution", str(ffmpeg_task.resolution))
+        span.set_attribute("task.ext_data", json.dumps(ffmpeg_task.ext_data))
+        result = ffmpeg.start_render(ffmpeg_task)
+        if not result:
+            span.set_status(Status(StatusCode.ERROR))
+            return False
+        span.set_status(Status(StatusCode.OK))
+        return True
+
+
+def clear_task_tmp_file(ffmpeg_task):
+    for task in ffmpeg_task.analyze_input_render_tasks():
+        clear_task_tmp_file(task)
+    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())
+        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())
+        return False
+    return True
+
+
+def probe_video_info(ffmpeg_task):
+    # 获取视频长度宽度和时长
+    return ffmpeg.probe_video_info(ffmpeg_task.get_output_file())
+
+

biz/task.py

@@ -1,15 +1,44 @@
+import json
+
+from opentelemetry.trace import Status, StatusCode
+
+from biz.ffmpeg import parse_ffmpeg_task, start_ffmpeg_task, clear_task_tmp_file, probe_video_info, fade_out_audio
+from telemetry import get_tracer
 from template import get_template_def
-
-
-def normalize_task(task_info):
-    ...
-    return task_info
+from util import api
 
 
 def start_task(task_info):
-    from biz.ffmpeg import parse_ffmpeg_task, start_ffmpeg_task
-    task_info = normalize_task(task_info)
-    task_template = "test_template"
-    template_info = get_template_def(task_template)
-    ffmpeg_task = parse_ffmpeg_task(task_info, template_info)
-    result = start_ffmpeg_task(ffmpeg_task)
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_task") as span:
+        task_info = api.normalize_task(task_info)
+        span.set_attribute("task", json.dumps(task_info))
+        span.set_attribute("scenicId", task_info.get("scenicId", "?"))
+        span.set_attribute("templateId", task_info.get("templateId"))
+        template_info = get_template_def(task_info.get("templateId"))
+        api.report_task_start(task_info)
+        ffmpeg_task = parse_ffmpeg_task(task_info, template_info)
+        result = start_ffmpeg_task(ffmpeg_task)
+        if not result:
+            span.set_status(Status(StatusCode.ERROR))
+            return api.report_task_failed(task_info)
+        width, height, duration = probe_video_info(ffmpeg_task)
+        span.set_attribute("probe.width", width)
+        span.set_attribute("probe.height", height)
+        span.set_attribute("probe.duration", duration)
+        # 音频淡出
+        new_fn = fade_out_audio(ffmpeg_task.get_output_file(), duration)
+        ffmpeg_task.set_output_file(new_fn)
+        oss_result = api.upload_task_file(task_info, ffmpeg_task)
+        if not oss_result:
+            span.set_status(Status(StatusCode.ERROR))
+            return api.report_task_failed(task_info)
+        # 获取视频长度宽度和时长
+        clear_task_tmp_file(ffmpeg_task)
+        api.report_task_success(task_info, videoInfo={
+            "width": width,
+            "height": height,
+            "duration": duration
+        })
+        span.set_status(Status(StatusCode.OK))
+        return None

config/__init__.py

@@ -4,7 +4,7 @@ from logging.handlers import TimedRotatingFileHandler
 from dotenv import load_dotenv
 
 load_dotenv()
-logging.basicConfig(level=logging.DEBUG)
+logging.basicConfig(level=logging.INFO)
 root_logger = logging.getLogger()
 rf_handler = TimedRotatingFileHandler('all_log.log', when='midnight')
 rf_handler.setFormatter(logging.Formatter("[%(asctime)s][%(name)s]%(levelname)s - %(message)s"))

entity/ffmpeg.py

@@ -1,23 +1,44 @@
+import json
+import os
+import time
+import uuid
+from typing import Any
+
+DEFAULT_ARGS = ("-shortest",)
+ENCODER_ARGS = ("-c:v", "h264", ) if not os.getenv("ENCODER_ARGS", False) else os.getenv("ENCODER_ARGS", "").split(" ")
+VIDEO_ARGS = ("-profile:v", "high", "-level:v", "4", )
+AUDIO_ARGS = ("-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", )
+MUTE_AUDIO_INPUT = ("-f", "lavfi", "-i", "anullsrc=cl=stereo:r=48000", )
+
 
 class FfmpegTask(object):
 
+    effects: list[str]
+
     def __init__(self, input_file, task_type='copy', output_file=''):
+        self.annexb = False
         if type(input_file) is str:
+            if input_file.endswith(".ts"):
+                self.annexb = True
             self.input_file = [input_file]
         elif type(input_file) is list:
             self.input_file = input_file
         else:
             self.input_file = []
+        self.zoom_cut = None
+        self.center_cut = None
+        self.ext_data = {}
         self.task_type = task_type
         self.output_file = output_file
         self.mute = True
         self.speed = 1
         self.frame_rate = 25
+        self.resolution = None
         self.subtitles = []
         self.luts = []
         self.audios = []
         self.overlays = []
-        self.annexb = False
+        self.effects = []
 
     def __repr__(self):
         _str = f'FfmpegTask(input_file={self.input_file}, task_type={self.task_type}'
@@ -29,8 +50,11 @@ class FfmpegTask(object):
             _str += f', overlays={self.overlays}'
         if self.annexb:
             _str += f', annexb={self.annexb}'
+        if self.effects:
+            _str += f', effects={self.effects}'
         if self.mute:
             _str += f', mute={self.mute}'
+        _str += f', center_cut={self.center_cut}'
         return _str + ')'
 
     def analyze_input_render_tasks(self):
@@ -56,7 +80,11 @@ class FfmpegTask(object):
         self.input_file.extend(inputs)
 
     def add_overlay(self, *overlays):
-        self.overlays.extend(overlays)
+        for overlay in overlays:
+            if str(overlay).endswith('.ass'):
+                self.subtitles.append(overlay)
+            else:
+                self.overlays.append(overlay)
         self.correct_task_type()
 
     def add_audios(self, *audios):
@@ -68,6 +96,10 @@ class FfmpegTask(object):
         self.luts.extend(luts)
         self.correct_task_type()
 
+    def add_effect(self, *effects):
+        self.effects.extend(effects)
+        self.correct_task_type()
+
     def get_output_file(self):
         if self.task_type == 'copy':
             return self.input_file[0]
@@ -90,8 +122,14 @@ class FfmpegTask(object):
             return False
         if len(self.subtitles) > 0:
             return False
+        if len(self.effects) > 0:
+            return False
         if self.speed != 1:
             return False
+        if self.zoom_cut is not None:
+            return False
+        if self.center_cut is not None:
+            return False
         return True
 
     def check_can_copy(self):
@@ -101,114 +139,242 @@ class FfmpegTask(object):
             return False
         if len(self.subtitles) > 0:
             return False
+        if len(self.effects) > 0:
+            return False
         if self.speed != 1:
             return False
-        if len(self.audios) > 1:
+        if len(self.audios) >= 1:
             return False
         if len(self.input_file) > 1:
             return False
+        if self.zoom_cut is not None:
+            return False
+        if self.center_cut is not None:
+            return False
         return True
 
     def check_audio_track(self):
-        if len(self.audios) > 0:
-            self.mute = False
+        ...
 
     def get_ffmpeg_args(self):
         args = ['-y', '-hide_banner']
-        video_output_str = "[0:v]"
         if self.task_type == 'encode':
             input_args = []
             filter_args = []
-            output_args = []
+            output_args = [*VIDEO_ARGS, *AUDIO_ARGS, *ENCODER_ARGS, *DEFAULT_ARGS]
+            if self.annexb:
+                output_args.append("-bsf:v")
+                output_args.append("h264_mp4toannexb")
+                output_args.append("-reset_timestamps")
+                output_args.append("1")
             video_output_str = "[0:v]"
-            audio_output_str = "[0:v]"
-            video_input_count = 0
-            audio_input_count = 0
+            audio_output_str = ""
+            audio_track_index = 0
+            effect_index = 0
             for input_file in self.input_file:
                 input_args.append("-i")
                 if type(input_file) is str:
                     input_args.append(input_file)
                 elif isinstance(input_file, FfmpegTask):
                     input_args.append(input_file.get_output_file())
+            if self.center_cut == 1:
+                pos_json_str = self.ext_data.get('posJson', '{}')
+                pos_json = json.loads(pos_json_str)
+                _v_w = pos_json.get('imgWidth', 1)
+                _f_x = pos_json.get('ltX', 0)
+                _f_x2 = pos_json.get('rbX', 0)
+                _x = f'{float((_f_x2 + _f_x)/(2 * _v_w)) :.4f}*iw-ih*ih/(2*iw)'
+                filter_args.append(f"{video_output_str}crop=x={_x}:y=0:w=ih*ih/iw:h=ih[v_cut{effect_index}]")
+                video_output_str = f"[v_cut{effect_index}]"
+                effect_index += 1
+            for effect in self.effects:
+                if effect.startswith("cameraShot:"):
+                    param = effect.split(":", 2)[1]
+                    if param == '':
+                        param = "3,1,0"
+                    _split = param.split(",")
+                    start = 3
+                    duration = 1
+                    rotate_deg = 0
+                    if len(_split) >= 3:
+                        if _split[2] == '':
+                            rotate_deg = 0
+                        else:
+                            rotate_deg = int(_split[2])
+                    if len(_split) >= 2:
+                        duration = float(_split[1])
+                    if len(_split) >= 1:
+                        start = float(_split[0])
+                    _start_out_str = "[eff_s]"
+                    _mid_out_str = "[eff_m]"
+                    _end_out_str = "[eff_e]"
+                    filter_args.append(f"{video_output_str}split=3{_start_out_str}{_mid_out_str}{_end_out_str}")
+                    filter_args.append(f"{_start_out_str}select=lt(n\\,{int(start*self.frame_rate)}){_start_out_str}")
+                    filter_args.append(f"{_end_out_str}select=gt(n\\,{int(start*self.frame_rate)}){_end_out_str}")
+                    filter_args.append(f"{_mid_out_str}select=eq(n\\,{int(start*self.frame_rate)}){_mid_out_str}")
+                    filter_args.append(f"{_mid_out_str}tpad=start_mode=clone:start_duration={duration:.4f}{_mid_out_str}")
+                    if rotate_deg != 0:
+                        filter_args.append(f"{_mid_out_str}rotate=PI*{rotate_deg}/360{_mid_out_str}")
+                    # filter_args.append(f"{video_output_str}trim=start=0:end={start+duration},tpad=stop_mode=clone:stop_duration={duration},setpts=PTS-STARTPTS{_start_out_str}")
+                    # filter_args.append(f"tpad=start_mode=clone:start_duration={duration},setpts=PTS-STARTPTS{_start_out_str}")
+                    # filter_args.append(f"{_end_out_str}trim=start={start}{_end_out_str}")
+                    video_output_str = f"[v_eff{effect_index}]"
+                    # filter_args.append(f"{_end_out_str}{_start_out_str}overlay=eof_action=pass{video_output_str}")
+                    filter_args.append(f"{_start_out_str}{_mid_out_str}{_end_out_str}concat=n=3:v=1:a=0,setpts=N/{self.frame_rate}/TB{video_output_str}")
+                    effect_index += 1
+                elif effect.startswith("ospeed:"):
+                    param = effect.split(":", 2)[1]
+                    if param == '':
+                        param = "1"
+                    if param != "1":
+                        # 视频变速
+                        effect_index += 1
+                        filter_args.append(f"{video_output_str}setpts={param}*PTS[v_eff{effect_index}]")
+                        video_output_str = f"[v_eff{effect_index}]"
+                elif effect.startswith("zoom:"):
+                    ...
+                ...
             for lut in self.luts:
-                filter_args.append("[0:v]lut3d=file=" + lut + "[0:v]")
+                filter_args.append(f"{video_output_str}lut3d=file={lut}{video_output_str}")
+            if self.resolution:
+                filter_args.append(f"{video_output_str}scale={self.resolution.replace('x', ':')}[v]")
+                video_output_str = "[v]"
             for overlay in self.overlays:
                 input_index = input_args.count("-i")
                 input_args.append("-i")
                 input_args.append(overlay)
-                filter_args.append(f"{video_output_str}[{input_index}:v]scale=rw:rh[v]")
+                if os.getenv("OLD_FFMPEG"):
+                    filter_args.append(f"{video_output_str}[{input_index}:v]scale2ref=iw:ih[v]")
+                else:
+                    filter_args.append(f"{video_output_str}[{input_index}:v]scale=rw:rh[v]")
                 filter_args.append(f"[v][{input_index}:v]overlay=1:eof_action=endall[v]")
                 video_output_str = "[v]"
+            for subtitle in self.subtitles:
+                filter_args.append(f"{video_output_str}ass={subtitle}[v]")
+                video_output_str = "[v]"
             output_args.append("-map")
             output_args.append(video_output_str)
             output_args.append("-r")
             output_args.append(f"{self.frame_rate}")
             if self.mute:
-                output_args.append("-an")
+                input_index = input_args.count("-i")
+                input_args += MUTE_AUDIO_INPUT
+                filter_args.append(f"[{input_index}:a]acopy[a]")
+                audio_track_index += 1
+                audio_output_str = "[a]"
             else:
-                input_index = 0
-                for audio in self.audios:
-                    input_index = input_args.count("-i")
-                    input_args.append("-i")
-                    input_args.append(audio.replace("\\", "/"))
-                    if audio_input_count > 0:
-                        filter_args.append(f"{audio_output_str}[{input_index}:a]amix[a]")
-                        audio_output_str = "[a]"
-                    else:
-                        audio_output_str = f"[{input_index}:a]"
-                    audio_input_count += 1
-                if audio_input_count == 1:
-                    audio_output_str = f"{input_index}"
-                output_args.append(f"-map")
-                output_args.append(audio_output_str)
-            return args + input_args + ["-filter_complex", ";".join(filter_args)] + output_args + [self.get_output_file()]
-        elif self.task_type == 'concat':
-            input_args = []
-            output_args = []
-            filter_args = []
-            video_output_str = "[0:v]"
-            audio_output_str = "[0:v]"
-            video_input_count = 0
-            audio_input_count = 0
-            for input_file in self.input_file:
+                audio_output_str = "[0:a]"
+                audio_track_index += 1
+            for audio in self.audios:
                 input_index = input_args.count("-i")
                 input_args.append("-i")
-                if type(input_file) is str:
-                    input_args.append(input_file.replace("\\", "/"))
-                elif isinstance(input_file, FfmpegTask):
-                    input_args.append(input_file.get_output_file().replace("\\", "/"))
-                if video_input_count > 0:
-                    filter_args.append(f"{video_output_str}[{input_index}:v]concat=n=2:v=1:a=0[v]")
-                    video_output_str = "[v]"
-                else:
-                    video_output_str = f"[{input_index}:v]"
-                video_input_count += 1
-            output_args.append("-map")
-            output_args.append(video_output_str)
-            if self.mute:
-                output_args.append("-an")
-            else:
-                input_index = 0
-                for audio in self.audios:
-                    input_index = input_args.count("-i")
-                    input_args.append("-i")
-                    input_args.append(audio.replace("\\", "/"))
-                    if audio_input_count > 0:
-                        filter_args.append(f"{audio_output_str}[{input_index}:a]amix[a]")
-                        audio_output_str = "[a]"
-                    else:
-                        audio_output_str = f"[{input_index}:a]"
-                    audio_input_count += 1
-                if audio_input_count == 1:
-                    audio_output_str = f"{input_index}"
-                output_args.append(f"-map")
+                input_args.append(audio.replace("\\", "/"))
+                audio_track_index += 1
+                filter_args.append(f"{audio_output_str}[{input_index}:a]amix=duration=shortest:dropout_transition=0:normalize=0[a]")
+                audio_output_str = "[a]"
+            if audio_output_str:
+                output_args.append("-map")
                 output_args.append(audio_output_str)
-            return args + input_args + ["-filter_complex", ";".join(filter_args)] +  output_args + [self.get_output_file()]
+            _filter_args = [] if len(filter_args) == 0 else ["-filter_complex", ";".join(filter_args)]
+            return args + input_args + _filter_args + output_args + [self.get_output_file()]
+        elif self.task_type == 'concat':
+            # 无法通过 annexb 合并的
+            input_args = []
+            output_args = [*DEFAULT_ARGS]
+            filter_args = []
+            audio_output_str = ""
+            audio_track_index = 0
+            # output_args
+            if len(self.input_file) == 1:
+                _file = self.input_file[0]
+                from util.ffmpeg import probe_video_audio
+                if type(_file) is str:
+                    input_args += ["-i", _file]
+                    self.mute = not probe_video_audio(_file)
+                elif isinstance(_file, FfmpegTask):
+                    input_args += ["-i", _file.get_output_file()]
+                    self.mute = not probe_video_audio(_file.get_output_file())
+            else:
+                _tmp_file = "tmp_concat_" + str(time.time()) + ".txt"
+                from util.ffmpeg import probe_video_audio
+                with open(_tmp_file, "w", encoding="utf-8") as f:
+                    for input_file in self.input_file:
+                        if type(input_file) is str:
+                            f.write("file '" + input_file + "'\n")
+                        elif isinstance(input_file, FfmpegTask):
+                            f.write("file '" + input_file.get_output_file() + "'\n")
+                input_args += ["-f", "concat", "-safe", "0", "-i", _tmp_file]
+                self.mute = not probe_video_audio(_tmp_file, "concat")
+            output_args.append("-map")
+            output_args.append("0:v")
+            output_args.append("-c:v")
+            output_args.append("copy")
+            if self.mute:
+                input_index = input_args.count("-i")
+                input_args += MUTE_AUDIO_INPUT
+                audio_output_str = f"[{input_index}:a]"
+                audio_track_index += 1
+            else:
+                audio_output_str = "[0:a]"
+                audio_track_index += 1
+            for audio in self.audios:
+                input_index = input_args.count("-i")
+                input_args.append("-i")
+                input_args.append(audio.replace("\\", "/"))
+                audio_track_index += 1
+                filter_args.append(f"{audio_output_str}[{input_index}:a]amix=duration=shortest:dropout_transition=0:normalize=0[a]")
+                audio_output_str = "[a]"
+            if audio_output_str:
+                output_args.append("-map")
+                if audio_track_index <= 1:
+                    output_args.append(audio_output_str[1:-1])
+                else:
+                    output_args.append(audio_output_str)
+            output_args += AUDIO_ARGS
+            if self.annexb:
+                output_args.append("-bsf:v")
+                output_args.append("h264_mp4toannexb")
+                output_args.append("-bsf:a")
+                output_args.append("setts=pts=DTS")
+            output_args.append("-f")
+            output_args.append("mpegts" if self.annexb else "mp4")
+            _filter_args = [] if len(filter_args) == 0 else ["-filter_complex", ";".join(filter_args)]
+            return args + input_args + _filter_args + output_args + [self.get_output_file()]
+        elif self.task_type == 'copy':
+            if len(self.input_file) == 1:
+                if type(self.input_file[0]) is str:
+                    if self.input_file[0] == self.get_output_file():
+                        return []
+            return args + ["-i", self.input_file[0]] + ["-c", "copy", self.get_output_file()]
+        return []
 
     def set_output_file(self, file=None):
         if file is None:
             if self.output_file == '':
-                # TODO: Random Filename
-                self.output_file = "rand.mp4"
+                if self.annexb:
+                    self.output_file = "rand_" + str(uuid.uuid4()) + ".ts"
+                else:
+                    self.output_file = "rand_" + str(uuid.uuid4()) + ".mp4"
         else:
-            self.output_file = file
+            if isinstance(file, FfmpegTask):
+                if file == self:
+                    return
+                self.output_file = file.get_output_file()
+            if type(file) is str:
+                self.output_file = file
+
+    def check_annexb(self):
+        for input_file in self.input_file:
+            if type(input_file) is str:
+                if self.task_type == 'encode':
+                    return self.annexb
+                elif self.task_type == 'concat':
+                    return False
+                elif self.task_type == 'copy':
+                    return self.annexb
+                else:
+                    return False
+            elif isinstance(input_file, FfmpegTask):
+                if not input_file.check_annexb():
+                    return False
+        return True

index.py

@@ -1,19 +1,42 @@
 from time import sleep
 
-import biz.task
 import config
+import biz.task
+from telemetry import init_opentelemetry
 from template import load_local_template
 from util import api
-from util.system import get_sys_info
+
+import os
+import glob
 
 load_local_template()
+import logging
 
+LOGGER = logging.getLogger(__name__)
+init_opentelemetry()
 
 while True:
     # print(get_sys_info())
     print("waiting for task...")
-    task_list = api.get_render_task()
+    try:
+        task_list = api.sync_center()
+    except Exception as e:
+        LOGGER.error("sync_center error", exc_info=e)
+        sleep(5)
+        continue
+    if len(task_list) == 0:
+        # 删除当前文件夹下所有以.mp4、.ts结尾的文件
+        for file_globs in ['*.mp4', '*.ts', 'tmp_concat*.txt']:
+            for file_path in glob.glob(file_globs):
+                try:
+                    os.remove(file_path)
+                    print(f"Deleted file: {file_path}")
+                except Exception as e:
+                    LOGGER.error(f"Error deleting file {file_path}", exc_info=e)
+        sleep(5)
     for task in task_list:
         print("start task:", task)
-        biz.task.start_task(task)
-    break
+        try:
+            biz.task.start_task(task)
+        except Exception as e:
+            LOGGER.error("task_start error", exc_info=e)

requirements.txt

@@ -1,3 +1,7 @@
 requests~=2.32.3
 psutil~=6.1.0
-python-dotenv~=1.0.1
+python-dotenv~=1.0.1
+opentelemetry-api~=1.30.0
+opentelemetry-sdk~=1.30.0
+opentelemetry-exporter-otlp~=1.30.0
+flask~=3.1.0

telemetry/__init__.py

@@ -0,0 +1,35 @@
+import os
+
+from constant import SOFTWARE_VERSION
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as OTLPSpanHttpExporter
+from opentelemetry.sdk.resources import DEPLOYMENT_ENVIRONMENT, HOST_NAME, Resource, SERVICE_NAME, SERVICE_VERSION
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor
+
+
+def get_tracer(name):
+    return trace.get_tracer(name)
+
+# 初始化 OpenTelemetry
+def init_opentelemetry(batch=True):
+    # 设置服务名、主机名
+    resource = Resource(attributes={
+        SERVICE_NAME: "RENDER_WORKER",
+        SERVICE_VERSION: SOFTWARE_VERSION,
+        DEPLOYMENT_ENVIRONMENT: "Python",
+        HOST_NAME: os.getenv("ACCESS_KEY"),
+    })
+
+    # 使用HTTP协议上报
+    if batch:
+        span_processor = BatchSpanProcessor(OTLPSpanHttpExporter(
+            endpoint="https://oltp.jerryyan.top/v1/traces",
+        ))
+    else:
+        span_processor = SimpleSpanProcessor(OTLPSpanHttpExporter(
+            endpoint="https://oltp.jerryyan.top/v1/traces",
+        ))
+
+    trace_provider = TracerProvider(resource=resource, active_span_processor=span_processor)
+    trace.set_tracer_provider(trace_provider)

template/__init__.py

@@ -2,6 +2,9 @@ import json
 import os
 import logging
 
+from telemetry import get_tracer
+from util import api, oss
+
 TEMPLATES = {}
 logger = logging.getLogger("template")
 
@@ -68,11 +71,55 @@ def load_local_template():
 
 
 def get_template_def(template_id):
+    if template_id not in TEMPLATES:
+        download_template(template_id)
     return TEMPLATES.get(template_id)
 
 def download_template(template_id):
-    logger.info(f"下载模板：{template_id}")
-    ...
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("download_template"):
+        template_info = api.get_template_info(template_id)
+        if not os.path.isdir(template_info['local_path']):
+            os.makedirs(template_info['local_path'])
+        # download template assets
+        overall_template = template_info['overall_template']
+        video_parts = template_info['video_parts']
+        def _download_assets(_template):
+            if 'source' in _template:
+                if str(_template['source']).startswith("http"):
+                    _, _fn = os.path.split(_template['source'])
+                    new_fp = os.path.join(template_info['local_path'], _fn)
+                    oss.download_from_oss(_template['source'], new_fp)
+                    if _fn.endswith(".mp4"):
+                        from util.ffmpeg import re_encode_and_annexb
+                        new_fp = re_encode_and_annexb(new_fp)
+                    _template['source'] = os.path.relpath(new_fp, template_info['local_path'])
+            if 'overlays' in _template:
+                for i in range(len(_template['overlays'])):
+                    overlay = _template['overlays'][i]
+                    if str(overlay).startswith("http"):
+                        _, _fn = os.path.split(overlay)
+                        oss.download_from_oss(overlay, os.path.join(template_info['local_path'], _fn))
+                        _template['overlays'][i] = _fn
+            if 'luts' in _template:
+                for i in range(len(_template['luts'])):
+                    lut = _template['luts'][i]
+                    if str(lut).startswith("http"):
+                        _, _fn = os.path.split(lut)
+                        oss.download_from_oss(lut, os.path.join(template_info['local_path'], _fn))
+                        _template['luts'][i] = _fn
+            if 'audios' in _template:
+                for i in range(len(_template['audios'])):
+                    if str(_template['audios'][i]).startswith("http"):
+                        _, _fn = os.path.split(_template['audios'][i])
+                        oss.download_from_oss(_template['audios'][i], os.path.join(template_info['local_path'], _fn))
+                        _template['audios'][i] = _fn
+        _download_assets(overall_template)
+        for video_part in video_parts:
+            _download_assets(video_part)
+        with open(os.path.join(template_info['local_path'], 'template.json'), 'w', encoding='utf-8') as f:
+            json.dump(template_info, f)
+        load_template(template_id, template_info['local_path'])
 
 
 def analyze_template(template_id):

util/api.py

@@ -1,19 +1,62 @@
+import json
+import logging
+import os
+import threading
+
 import requests
+from opentelemetry.trace import Status, StatusCode
+
+import util.system
+from telemetry import get_tracer
+from util import oss
 
 session = requests.Session()
+logger = logging.getLogger(__name__)
 
 
-def get_render_task():
+def normalize_task(task_info):
+    ...
+    return task_info
+
+
+def sync_center():
     """
     通过接口获取任务
     :return: 任务列表
     """
-    tasks = []
-    tasks.append({
-        'user_videos': {
-            'CAM_ID': 'paper-planes.mp4'
-        }
-    })
+    from template import TEMPLATES, download_template
+    try:
+        response = session.post(os.getenv('API_ENDPOINT') + "/sync", json={
+            'accessKey': os.getenv('ACCESS_KEY'),
+            'clientStatus': util.system.get_sys_info(),
+            'templateList': [{'id': t.get('id', ''), 'updateTime': t.get('updateTime', '')} for t in
+                             TEMPLATES.values()]
+        }, timeout=10)
+        response.raise_for_status()
+    except requests.RequestException as e:
+        logger.error("请求失败！", e)
+        return []
+    data = response.json()
+    logger.debug("获取任务结果：【%s】", data)
+    if data.get('code', 0) == 200:
+        templates = data.get('data', {}).get('templates', [])
+        tasks = data.get('data', {}).get('tasks', [])
+    else:
+        tasks = []
+        templates = []
+        logger.warning("获取任务失败")
+    if os.getenv("REDIRECT_TO_URL", False) != False:
+        for task in tasks:
+            _sess = requests.Session()
+            logger.info("重定向任务【%s】至配置的地址：%s", task.get("id"), os.getenv("REDIRECT_TO_URL"))
+            url = f"{os.getenv('REDIRECT_TO_URL')}{task.get('id')}"
+            threading.Thread(target=requests.post, args=(url,)).start()
+        return []
+    for template in templates:
+        template_id = template.get('id', '')
+        if template_id:
+            logger.info("更新模板：【%s】", template_id)
+            download_template(template_id)
     return tasks
 
 
@@ -25,39 +68,189 @@ def get_template_info(template_id):
     :type template_id: str
     :return: 模板信息
     """
-    template = {
-        'id': template_id,
-        'name': '模板名称',
-        'description': '模板描述',
-        'video_size': '1920x1080',
-        'frame_rate': 30,
-        'overall_duration': 30,
-        'video_parts': [
-            {
-                'source': './template/test_template/1.mp4',
-                'mute': True,
-            },
-            {
-                'source': 'PLACEHOLDER_CAM_ID',
-                'mute': True,
-                'overlays': [
-                    './template/test_template/2.mov'
-                ],
-                'luts': [
-                    './template/test_template/cube.cube'
-                ]
-            },
-            {
-                'source': './template/test_template/3.mp4',
-                'mute': True,
-            }
-        ],
-        'overall_template': {
-            'source': None,
-            'mute': False,
-            'audios': [
-                './template/test_template/bgm.acc'
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("get_template_info"):
+        with tracer.start_as_current_span("get_template_info.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url", '{0}/template/{1}'.format(os.getenv('API_ENDPOINT'), template_id))
+                response = session.post('{0}/template/{1}'.format(os.getenv('API_ENDPOINT'), template_id), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                req_span.set_attribute("http.response", response.text)
+                response.raise_for_status()
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
+        data = response.json()
+        logger.debug("获取模板信息结果：【%s】", data)
+        remote_template_info = data.get('data', {})
+        template = {
+            'id': template_id,
+            'updateTime': remote_template_info.get('updateTime', template_id),
+            'scenic_name': remote_template_info.get('scenicName', '景区'),
+            'name': remote_template_info.get('name', '模版'),
+            'video_size': remote_template_info.get('resolution', '1920x1080'),
+            'frame_rate': 25,
+            'overall_duration': 30,
+            'video_parts': [
+
             ]
-        },
-    }
-    return template
+        }
+
+        def _template_normalizer(template_info):
+            _template = {}
+            _placeholder_type = template_info.get('isPlaceholder', -1)
+            if _placeholder_type == 0:
+                # 固定视频
+                _template['source'] = template_info.get('sourceUrl', '')
+            elif _placeholder_type == 1:
+                # 占位符
+                _template['source'] = "PLACEHOLDER_" + template_info.get('sourceUrl', '')
+                _template['mute'] = template_info.get('mute', True)
+                _template['crop_mode'] = template_info.get('cropEnable', None)
+            else:
+                _template['source'] = None
+            _overlays = template_info.get('overlays', '')
+            if _overlays:
+                _template['overlays'] = _overlays.split(",")
+            _audios = template_info.get('audios', '')
+            if _audios:
+                _template['audios'] = _audios.split(",")
+            _luts = template_info.get('luts', '')
+            if _luts:
+                _template['luts'] = _luts.split(",")
+            _only_if = template_info.get('onlyIf', '')
+            if _only_if:
+                _template['only_if'] = _only_if
+            _effects = template_info.get('effects', '')
+            if _effects:
+                _template['effects'] = _effects.split("|")
+            return _template
+
+        # outer template definition
+        overall_template = _template_normalizer(remote_template_info)
+        template['overall_template'] = overall_template
+        # inter template definition
+        inter_template_list = remote_template_info.get('children', [])
+        for children_template in inter_template_list:
+            parts = _template_normalizer(children_template)
+            template['video_parts'].append(parts)
+        template['local_path'] = os.path.join(os.getenv('TEMPLATE_DIR'), str(template_id))
+        with get_tracer("api").start_as_current_span("get_template_info.template") as res_span:
+            res_span.set_attribute("normalized.response", json.dumps(template))
+        return template
+
+
+def report_task_success(task_info, **kwargs):
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_success"):
+        with tracer.start_as_current_span("report_task_success.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/success'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/success'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                    **kwargs
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                req_span.set_attribute("http.response", response.text)
+                response.raise_for_status()
+                req_span.set_status(Status(StatusCode.OK))
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
+
+
+def report_task_start(task_info):
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_start"):
+        with tracer.start_as_current_span("report_task_start.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/start'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/start'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                req_span.set_attribute("http.response", response.text)
+                response.raise_for_status()
+                req_span.set_status(Status(StatusCode.OK))
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                logger.error("请求失败！", e)
+                return None
+
+
+def report_task_failed(task_info, reason=''):
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("report_task_failed") as span:
+        span.set_attribute("task_id", task_info.get("id"))
+        span.set_attribute("reason", reason)
+        with tracer.start_as_current_span("report_task_failed.request") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/fail'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/fail'.format(os.getenv('API_ENDPOINT'), task_info.get("id")), json={
+                    'accessKey': os.getenv('ACCESS_KEY'),
+                    'reason': reason
+                }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                req_span.set_attribute("http.response", response.text)
+                response.raise_for_status()
+                req_span.set_status(Status(StatusCode.OK))
+            except requests.RequestException as e:
+                req_span.set_attribute("api.error", str(e))
+                req_span.set_status(Status(StatusCode.ERROR))
+                logger.error("请求失败！", e)
+                return None
+
+
+def upload_task_file(task_info, ffmpeg_task):
+    tracer = get_tracer(__name__)
+    with get_tracer("api").start_as_current_span("upload_task_file") as span:
+        logger.info("开始上传文件: %s", task_info.get("id"))
+        span.set_attribute("file.id", task_info.get("id"))
+        with tracer.start_as_current_span("upload_task_file.request_upload_url") as req_span:
+            try:
+                req_span.set_attribute("http.method", "POST")
+                req_span.set_attribute("http.url",
+                                       '{0}/{1}/uploadUrl'.format(os.getenv('API_ENDPOINT'), task_info.get("id")))
+                response = session.post('{0}/{1}/uploadUrl'.format(os.getenv('API_ENDPOINT'), task_info.get("id")),
+                                        json={
+                                            'accessKey': os.getenv('ACCESS_KEY'),
+                                        }, timeout=10)
+                req_span.set_attribute("http.status_code", response.status_code)
+                req_span.set_attribute("http.response", response.text)
+                response.raise_for_status()
+                req_span.set_status(Status(StatusCode.OK))
+            except requests.RequestException as e:
+                span.set_attribute("api.error", str(e))
+                req_span.set_status(Status(StatusCode.ERROR))
+                logger.error("请求失败！", e)
+                return False
+        data = response.json()
+        url = data.get('data', "")
+        logger.info("开始上传文件: %s 至 %s", task_info.get("id"), url)
+        return oss.upload_to_oss(url, ffmpeg_task.get_output_file())
+
+
+def get_task_info(id):
+    try:
+        response = session.get(os.getenv('API_ENDPOINT') + "/" + id + "/info", params={
+            'accessKey': os.getenv('ACCESS_KEY'),
+        }, timeout=10)
+        response.raise_for_status()
+    except requests.RequestException as e:
+        logger.error("请求失败！", e)
+        return []
+    data = response.json()
+    logger.debug("获取任务结果：【%s】", data)
+    if data.get('code', 0) == 200:
+        return data.get('data', {})

util/ffmpeg.py

@@ -1,23 +1,95 @@
+import json
+import logging
 import os
+import subprocess
 from datetime import datetime
 from typing import Optional, IO
 
-from entity.ffmpeg import FfmpegTask
+from opentelemetry.trace import Status, StatusCode
 
+from entity.ffmpeg import FfmpegTask, ENCODER_ARGS, VIDEO_ARGS, AUDIO_ARGS, MUTE_AUDIO_INPUT
+from telemetry import get_tracer
+
+logger = logging.getLogger(__name__)
+
+def re_encode_and_annexb(file):
+    with get_tracer("ffmpeg").start_as_current_span("re_encode_and_annexb") as span:
+        span.set_attribute("file.path", file)
+        if not os.path.exists(file):
+            span.set_status(Status(StatusCode.ERROR))
+            return file
+        logger.info("ReEncodeAndAnnexb: %s", file)
+        has_audio = not not probe_video_audio(file)
+        ffmpeg_process = subprocess.run(["ffmpeg", "-y", "-hide_banner", "-vsync", "cfr", "-i", file,
+                                         *(set() if has_audio else MUTE_AUDIO_INPUT),
+                                         "-map", "0:v", "-map", "0:a" if has_audio else "1:a",
+                                         *VIDEO_ARGS, "-bsf:v", "h264_mp4toannexb",
+                                         *AUDIO_ARGS, "-bsf:a", "setts=pts=DTS",
+                                         *ENCODER_ARGS, "-shortest", "-fflags", "+genpts",
+                                         "-f", "mpegts", file + ".ts"])
+        logger.info(" ".join(ffmpeg_process.args))
+        span.set_attribute("ffmpeg.args", json.dumps(ffmpeg_process.args))
+        logger.info("ReEncodeAndAnnexb: %s, returned: %s", file, ffmpeg_process.returncode)
+        span.set_attribute("ffmpeg.code", ffmpeg_process.returncode)
+        if ffmpeg_process.returncode == 0:
+            span.set_status(Status(StatusCode.OK))
+            span.set_attribute("file.size", os.path.getsize(file+".ts"))
+            # os.remove(file)
+            return file+".ts"
+        else:
+            span.set_status(Status(StatusCode.ERROR))
+            return file
 
 def start_render(ffmpeg_task: FfmpegTask):
-    print(ffmpeg_task)
-    print(ffmpeg_task.get_ffmpeg_args())
-    os.system("ffmpeg.exe "+" ".join(ffmpeg_task.get_ffmpeg_args()))
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("start_render") as span:
+        span.set_attribute("ffmpeg.task", str(ffmpeg_task))
+        if not ffmpeg_task.need_run():
+            ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
+            span.set_status(Status(StatusCode.OK))
+            return True
+        ffmpeg_args = ffmpeg_task.get_ffmpeg_args()
+        if len(ffmpeg_args) == 0:
+            ffmpeg_task.set_output_file(ffmpeg_task.input_file[0])
+            span.set_status(Status(StatusCode.OK))
+            return True
+        ffmpeg_process = subprocess.run(["ffmpeg", "-progress", "-", "-loglevel", "error", *ffmpeg_args], stderr=subprocess.PIPE, **subprocess_args(True))
+        span.set_attribute("ffmpeg.args", json.dumps(ffmpeg_process.args))
+        logger.info(" ".join(ffmpeg_process.args))
+        ffmpeg_final_out = handle_ffmpeg_output(ffmpeg_process.stdout)
+        span.set_attribute("ffmpeg.out", ffmpeg_final_out)
+        logger.info("FINISH TASK, OUTPUT IS %s", ffmpeg_final_out)
+        code = ffmpeg_process.returncode
+        span.set_attribute("ffmpeg.code", code)
+        if code != 0:
+            span.set_attribute("ffmpeg.err", str(ffmpeg_process.stderr))
+            span.set_status(Status(StatusCode.ERROR, "FFMPEG异常退出"))
+            logger.error("FFMPEG ERROR: %s", ffmpeg_process.stderr)
+            return False
+        span.set_attribute("ffmpeg.out_file", ffmpeg_task.output_file)
+        try:
+            file_size = os.path.getsize(ffmpeg_task.output_file)
+            span.set_attribute("file.size", file_size)
+            if file_size < 4096:
+                span.set_status(Status(StatusCode.ERROR, "输出文件过小"))
+                logger.error("FFMPEG ERROR: OUTPUT FILE IS TOO SMALL")
+                return False
+        except OSError as e:
+            span.set_attribute("file.size", 0)
+            span.set_attribute("file.error", e.strerror)
+            span.set_status(Status(StatusCode.ERROR, "输出文件不存在"))
+            logger.error("FFMPEG ERROR: OUTPUT FILE NOT FOUND")
+            return False
+        span.set_status(Status(StatusCode.OK))
+        return True
 
-def handle_ffmpeg_output(stdout: Optional[IO[bytes]]) -> str:
+def handle_ffmpeg_output(stdout: Optional[bytes]) -> str:
     out_time = "0:0:0.0"
     if stdout is None:
         print("[!]STDOUT is null")
         return out_time
     speed = "0"
-    while True:
-        line = stdout.readline()
+    for line in stdout.split(b"\n"):
         if line == b"":
             break
         if line.strip() == b"progress=end":
@@ -28,8 +100,145 @@ def handle_ffmpeg_output(stdout: Optional[IO[bytes]]) -> str:
         if line.startswith(b"speed="):
             speed = line.replace(b"speed=", b"").decode().strip()
     print("[ ]Speed:", out_time, "@", speed)
-    return out_time
+    return out_time+"@"+speed
 
 def duration_str_to_float(duration_str: str) -> float:
     _duration = datetime.strptime(duration_str, "%H:%M:%S.%f") - datetime(1900, 1, 1)
     return _duration.total_seconds()
+
+
+def probe_video_info(video_file):
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("probe_video_info") as span:
+        span.set_attribute("video.file", video_file)
+        # 获取宽度和高度
+        result = subprocess.run(
+            ["ffprobe", '-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height:format=duration', '-of',
+             'csv=s=x:p=0', video_file],
+            stderr=subprocess.STDOUT,
+            **subprocess_args(True)
+        )
+        span.set_attribute("ffprobe.args", json.dumps(result.args))
+        span.set_attribute("ffprobe.code", result.returncode)
+        if result.returncode != 0:
+            span.set_status(Status(StatusCode.ERROR))
+            return 0, 0, 0
+        all_result = result.stdout.decode('utf-8').strip()
+        span.set_attribute("ffprobe.out", all_result)
+        if all_result == '':
+            span.set_status(Status(StatusCode.ERROR))
+            return 0, 0, 0
+        span.set_status(Status(StatusCode.OK))
+        wh, duration = all_result.split('\n')
+        width, height = wh.strip().split('x')
+        return int(width), int(height), float(duration)
+
+
+def probe_video_audio(video_file, type=None):
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("probe_video_audio") as span:
+        span.set_attribute("video.file", video_file)
+        args = ["ffprobe", "-hide_banner", "-v", "error", "-select_streams", "a", "-show_entries", "stream=index", "-of", "csv=p=0"]
+        if type == 'concat':
+            args.append("-safe")
+            args.append("0")
+            args.append("-f")
+            args.append("concat")
+        args.append(video_file)
+        logger.info(" ".join(args))
+        result = subprocess.run(args, stderr=subprocess.STDOUT, **subprocess_args(True))
+        span.set_attribute("ffprobe.args", json.dumps(result.args))
+        span.set_attribute("ffprobe.code", result.returncode)
+        logger.info("probe_video_audio: %s", result.stdout.decode('utf-8').strip())
+        if result.returncode != 0:
+            return False
+        if result.stdout.decode('utf-8').strip() == '':
+            return False
+        return True
+
+
+# 音频淡出2秒
+def fade_out_audio(file, duration, fade_out_sec = 2):
+    if type(duration) == str:
+        try:
+            duration = float(duration)
+        except Exception as e:
+            logger.error("duration is not float: %s", e)
+            return file
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("fade_out_audio") as span:
+        span.set_attribute("audio.file", file)
+        if duration <= fade_out_sec:
+            return file
+        else:
+            new_fn = file + "_.mp4"
+            if os.path.exists(new_fn):
+                os.remove(new_fn)
+                logger.info("delete tmp file: " + new_fn)
+            try:
+                process = subprocess.run(["ffmpeg", "-i", file, "-c:v", "copy", "-c:a", "aac", "-af", "afade=t=out:st=" + str(duration - fade_out_sec) + ":d=" + str(fade_out_sec), "-y", new_fn], **subprocess_args(True))
+                span.set_attribute("ffmpeg.args", json.dumps(process.args))
+                logger.info(" ".join(process.args))
+                if process.returncode != 0:
+                    span.set_status(Status(StatusCode.ERROR))
+                    logger.error("FFMPEG ERROR: %s", process.stderr)
+                    return file
+                else:
+                    span.set_status(Status(StatusCode.OK))
+                    return new_fn
+            except Exception as e:
+                span.set_status(Status(StatusCode.ERROR))
+                logger.error("FFMPEG ERROR: %s", e)
+                return file
+
+
+
+# Create a set of arguments which make a ``subprocess.Popen`` (and
+# variants) call work with or without Pyinstaller, ``--noconsole`` or
+# not, on Windows and Linux. Typical use::
+#
+#   subprocess.call(['program_to_run', 'arg_1'], **subprocess_args())
+#
+# When calling ``check_output``::
+#
+#   subprocess.check_output(['program_to_run', 'arg_1'],
+#                           **subprocess_args(False))
+def subprocess_args(include_stdout=True):
+    # The following is true only on Windows.
+    if hasattr(subprocess, 'STARTUPINFO'):
+        # On Windows, subprocess calls will pop up a command window by default
+        # when run from Pyinstaller with the ``--noconsole`` option. Avoid this
+        # distraction.
+        si = subprocess.STARTUPINFO()
+        si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+        # Windows doesn't search the path by default. Pass it an environment so
+        # it will.
+        env = os.environ
+    else:
+        si = None
+        env = None
+
+    # ``subprocess.check_output`` doesn't allow specifying ``stdout``::
+    #
+    #   Traceback (most recent call last):
+    #     File "test_subprocess.py", line 58, in <module>
+    #       **subprocess_args(stdout=None))
+    #     File "C:\Python27\lib\subprocess.py", line 567, in check_output
+    #       raise ValueError('stdout argument not allowed, it will be overridden.')
+    #   ValueError: stdout argument not allowed, it will be overridden.
+    #
+    # So, add it only if it's needed.
+    if include_stdout:
+        ret = {'stdout': subprocess.PIPE}
+    else:
+        ret = {}
+
+    # On Windows, running this from the binary produced by Pyinstaller
+    # with the ``--noconsole`` option requires redirecting everything
+    # (stdin, stdout, stderr) to avoid an OSError exception
+    # "[Error 6] the handle is invalid."
+    ret.update({'stdin': subprocess.PIPE,
+                'startupinfo': si,
+                'env': env})
+    return ret
+

util/oss.py

@@ -1,17 +1,116 @@
+import logging
+import os
+
 import requests
+from opentelemetry.trace import Status, StatusCode
+
+from telemetry import get_tracer
+
+logger = logging.getLogger(__name__)
 
 
-def upload_to_oss_use_signed_url(url, file_path):
+def upload_to_oss(url, file_path):
     """
     使用签名URL上传文件到OSS
     :param str url: 签名URL
     :param str file_path: 文件路径
     :return bool: 是否成功
     """
-    with open(file_path, 'rb') as f:
-        try:
-            response = requests.put(url, data=f)
-            return response.status_code == 200
-        except Exception as e:
-            print(e)
-            return False
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("upload_to_oss") as span:
+        span.set_attribute("file.url", url)
+        span.set_attribute("file.path", file_path)
+        span.set_attribute("file.size", os.path.getsize(file_path))
+        max_retries = 5
+        retries = 0
+        if os.getenv("UPLOAD_METHOD") == "rclone":
+            with tracer.start_as_current_span("rclone_to_oss") as r_span:
+                replace_map = os.getenv("RCLONE_REPLACE_MAP")
+                r_span.set_attribute("rclone.replace_map", replace_map)
+                if replace_map != "":
+                    replace_list = [i.split("|", 1) for i in replace_map.split(",")]
+                    new_url = url
+                    for (_src, _dst) in replace_list:
+                        new_url = new_url.replace(_src, _dst)
+                    new_url = new_url.split("?", 1)[0]
+                    r_span.set_attribute("rclone.target_dir", new_url)
+                    if new_url != url:
+                        result = os.system(f"rclone copyto --no-check-dest --ignore-existing --multi-thread-chunk-size 32M --multi-thread-streams 8 {file_path} {new_url}")
+                        r_span.set_attribute("rclone.result", result)
+                        if result == 0:
+                            span.set_status(Status(StatusCode.OK))
+                            return True
+                        else:
+                            span.set_status(Status(StatusCode.ERROR))
+        while retries < max_retries:
+            with tracer.start_as_current_span("upload_to_oss.request") as req_span:
+                req_span.set_attribute("http.retry_count", retries)
+                try:
+                    req_span.set_attribute("http.method", "PUT")
+                    req_span.set_attribute("http.url", url)
+                    with open(file_path, 'rb') as f:
+                        response = requests.put(url, data=f, stream=True, timeout=60, headers={"Content-Type": "video/mp4"})
+                        req_span.set_attribute("http.status_code", response.status_code)
+                        req_span.set_attribute("http.response", response.text)
+                        response.raise_for_status()
+                        req_span.set_status(Status(StatusCode.OK))
+                        span.set_status(Status(StatusCode.OK))
+                        return True
+                except requests.exceptions.Timeout:
+                    req_span.set_attribute("http.error", "Timeout")
+                    req_span.set_status(Status(StatusCode.ERROR))
+                    retries += 1
+                    logger.warning(f"Upload timed out. Retrying {retries}/{max_retries}...")
+                except Exception as e:
+                    req_span.set_attribute("http.error", str(e))
+                    req_span.set_status(Status(StatusCode.ERROR))
+                    retries += 1
+                    logger.warning(f"Upload failed. Retrying {retries}/{max_retries}...")
+        span.set_status(Status(StatusCode.ERROR))
+    return False
+
+
+def download_from_oss(url, file_path):
+    """
+    使用签名URL下载文件到OSS
+    :param str url: 签名URL
+    :param Union[LiteralString, str, bytes] file_path: 文件路径
+    :return bool: 是否成功
+    """
+    tracer = get_tracer(__name__)
+    with tracer.start_as_current_span("download_from_oss") as span:
+        span.set_attribute("file.url", url)
+        span.set_attribute("file.path", file_path)
+        logging.info("download_from_oss: %s", url)
+        file_dir, file_name = os.path.split(file_path)
+        if file_dir:
+            if not os.path.exists(file_dir):
+                os.makedirs(file_dir)
+        max_retries = 5
+        retries = 0
+        while retries < max_retries:
+            with tracer.start_as_current_span("download_from_oss.request") as req_span:
+                req_span.set_attribute("http.retry_count", retries)
+                try:
+                    req_span.set_attribute("http.method", "GET")
+                    req_span.set_attribute("http.url", url)
+                    response = requests.get(url, timeout=15)  # 设置超时时间
+                    req_span.set_attribute("http.status_code", response.status_code)
+                    with open(file_path, 'wb') as f:
+                        f.write(response.content)
+                    req_span.set_attribute("file.size", os.path.getsize(file_path))
+                    req_span.set_status(Status(StatusCode.OK))
+                    span.set_status(Status(StatusCode.OK))
+                    return True
+                except requests.exceptions.Timeout:
+                    req_span.set_attribute("http.error", "Timeout")
+                    req_span.set_status(Status(StatusCode.ERROR))
+                    retries += 1
+                    logger.warning(f"Download timed out. Retrying {retries}/{max_retries}...")
+                except Exception as e:
+                    req_span.set_attribute("http.error", str(e))
+                    req_span.set_status(Status(StatusCode.ERROR))
+                    retries += 1
+                    logger.warning(f"Download failed. Retrying {retries}/{max_retries}...")
+        span.set_status(Status(StatusCode.ERROR))
+    return False

util/system.py

@@ -1,5 +1,7 @@
 import os
 import platform
+from datetime import datetime
+
 import psutil
 from constant import SUPPORT_FEATURE, SOFTWARE_VERSION
 
@@ -10,6 +12,7 @@ def get_sys_info():
     """
     info = {
         'version': SOFTWARE_VERSION,
+        'client_datetime': datetime.now().isoformat(),
         'platform': platform.system(),
         'runtime_version': 'Python ' + platform.python_version(),
         'cpu_count': os.cpu_count(),