ai-flow-python/app/service/task/impl/video_editor_task.py

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)

            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)
        else:
            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