import json import logging import os import subprocess import time from typing import Any from app.service.task.base_task import BaseTask from app.service.task.task_param import TaskParam from app.service.task.task_result import TaskResult from app.util.download_util import get_relative_path from app.util.file_util import copy_file, move_file from app.util.res_util import get_resources_path logger = logging.getLogger(__name__) class VideoEditorTask(BaseTask): def get_type(self) -> str: return "video_editor" def execute(self, param: TaskParam) -> TaskResult: images = param.inputs.get("images") if isinstance(param.inputs.get("images"), list) else [] videos = param.inputs.get("videos") if isinstance(param.inputs.get("videos"), list) else [] logger.info("[video_editor] Starting: %d images, %d videos, resources=%s", len(images), len(videos), param.resources) t0 = time.time() try: src = os.path.join(get_resources_path(), "video_editor", "compile") dst = os.path.join(param.workspace, "compile") copy_file(src, dst) os.chmod(dst, 0o755) logger.info("[video_editor] Copied compile to workspace") self._update_json(param.workspace, param.resources, videos, images) logger.info("[video_editor] Updated JSON config with media paths") self._execute_compile(param.workspace, param.resources) output_path = self._move_output(param.workspace, param.resources, param.step_id) if not output_path: raise RuntimeError("No output path generated by video_editor") logger.info("[video_editor] Completed: output=%s in %.2fs", output_path, time.time() - t0) return TaskResult.success({"videos": [output_path]}) except Exception as e: logger.error("[video_editor] Failed in %.2fs: %s", time.time() - t0, e, exc_info=True) return TaskResult.failure(str(e)) @staticmethod def _update_json(workspace: str, resources: str, videos: list, images: list): json_path = os.path.join(workspace, resources) with open(json_path, "r", encoding="utf-8") as f: data = json.load(f) video_track = data.get("videoTrack", []) vi, ii = 0, 0 for track in video_track: t = track.get("type") if t == "video" and vi < len(videos): logger.info("[video_editor] Assigning video[%d] -> track path: %s", vi, videos[vi]) track["path"] = videos[vi] vi += 1 elif t == "image" and ii < len(images): logger.info("[video_editor] Assigning image[%d] -> track path: %s", ii, images[ii]) track["path"] = images[ii] ii += 1 with open(json_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) @staticmethod def _execute_compile(workspace: str, resources: str): cmd = f"./compile -m {resources}" logger.info("[video_editor] Executing: %s (cwd=%s)", cmd, workspace) cmd_t0 = time.time() result = subprocess.run( ["sh", "-c", cmd], cwd=workspace, capture_output=True, text=True, ) cmd_elapsed = time.time() - cmd_t0 if result.stdout: logger.info("[video_editor] stdout: %s", result.stdout.strip()) if result.stderr: logger.warning("[video_editor] stderr: %s", result.stderr.strip()) if result.returncode != 0: logger.error("[video_editor] compile failed: exit_code=%d in %.2fs", result.returncode, cmd_elapsed) raise RuntimeError(f"compile failed with exit code {result.returncode}") logger.info("[video_editor] compile completed: exit_code=0 in %.2fs", cmd_elapsed) @staticmethod def _move_output(workspace: str, resources: str, step_id: str) -> str | None: json_path = os.path.join(workspace, resources) with open(json_path, "r", encoding="utf-8") as f: data = json.load(f) output_cfg = data.get("output", {}) output_path = output_cfg.get("path") if not output_path: logger.warning("[video_editor] No output path found in config") return None src = os.path.join(workspace, output_path) dest_dir = os.path.join(workspace, step_id) os.makedirs(dest_dir, exist_ok=True) dest = os.path.join(dest_dir, os.path.basename(output_path)) move_file(src, dest) rel_path = get_relative_path(dest, workspace) logger.info("[video_editor] Output moved: %s -> %s", output_path, rel_path) return rel_path