修复

biz/ffmpeg.py

@@ -24,7 +24,7 @@ def parse_ffmpeg_task(task_info, template_info):
         task_params = json.loads(task_params_str)
         task_params_orig = json.loads(task_params_str)
         for part in template_info.get("video_parts"):
-            source = parse_video(part.get('source'), task_params, template_info)
+            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
@@ -36,7 +36,7 @@ def parse_ffmpeg_task(task_info, template_info):
             sub_ffmpeg_task = FfmpegTask(source)
             sub_ffmpeg_task.resolution = template_info.get("video_size", "")
             sub_ffmpeg_task.annexb = True
-            sub_ffmpeg_task.ext_data = find_placeholder_params(part.get('source'), task_params) or {}
+            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', []):
@@ -55,8 +55,9 @@ def parse_ffmpeg_task(task_info, template_info):
         task.center_cut = template_info.get("crop_mode", None)
         task.frame_rate = template_info.get("frame_rate", 25)
         if overall.get('source', ''):
-            source = parse_video(overall.get('source'), task_params, template_info)
+            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', []):
@@ -68,37 +69,24 @@ def parse_ffmpeg_task(task_info, template_info):
         return task
 
 
-def find_placeholder_params(source, task_params):
-    if source.startswith('PLACEHOLDER_'):
-        placeholder_id = source.replace('PLACEHOLDER_', '')
-        new_sources = task_params.get(placeholder_id, [])
-        if type(new_sources) is list:
-            if len(new_sources) == 0:
-                logger.debug("no video found for placeholder: " + placeholder_id)
-                return {}
-            else:
-                return new_sources[0]
-    return {}
-
-
 def parse_video(source, task_params, template_info):
-    print(source)
     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
+                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)
-            return source_name
-        return new_sources
-    return os.path.join(template_info.get("local_path"), source)
+            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):
@@ -121,6 +109,11 @@ def start_ffmpeg_task(ffmpeg_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))

biz/task.py

@@ -23,6 +23,9 @@ def start_task(task_info):
             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)
@@ -38,3 +41,4 @@ def start_task(task_info):
             "duration": duration
         })
         span.set_status(Status(StatusCode.OK))
+        return None

util/oss.py

@@ -23,6 +23,25 @@ def upload_to_oss(url, 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)