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

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.file_util import copy_file
from app.util.res_util import get_resources_path

logger = logging.getLogger(__name__)


class ImageEditorTask(BaseTask):
    def get_type(self) -> str:
        return "image_editor"

    def execute(self, param: TaskParam) -> TaskResult:
        images_obj = param.inputs.get("images")
        images: list[str] = images_obj if isinstance(images_obj, list) else []

        logger.info("[image_editor] Starting: %d images, resources=%s", len(images), param.resources)
        t0 = time.time()

        try:
            src = os.path.join(get_resources_path(), "image_editor", "p2pconvertor")
            dst = os.path.join(param.workspace, "p2pconvertor")
            copy_file(src, dst)
            os.chmod(dst, 0o755)
            logger.info("[image_editor] Copied p2pconvertor to workspace")

            output_path = f"{param.step_id}/output.jpg"
            self._execute_p2p(images, param.resources, output_path, param.workspace)

            out_file = os.path.join(param.workspace, output_path)
            if os.path.isfile(out_file):
                file_size = os.path.getsize(out_file)
                logger.info("[image_editor] Completed: output=%s (%.1f KB) in %.2fs",
                             output_path, file_size / 1024, time.time() - t0)
            else:
                logger.warning("[image_editor] Completed but output file not found: %s", output_path)

            return TaskResult.success({"images": [output_path]})
        except Exception as e:
            logger.error("[image_editor] Failed in %.2fs: %s", time.time() - t0, e, exc_info=True)
            return TaskResult.failure(str(e))

    @staticmethod
    def _execute_p2p(images: list[str], effect_dir: str, output_path: str, workspace: str):
        cmd_parts = ["./p2pconvertor"]
        for img in images:
            cmd_parts.extend(["-i", img])
        cmd_parts.extend(["-e", effect_dir, "-o", output_path])

        cmd_str = " ".join(cmd_parts)
        logger.info("[image_editor] Executing: %s  (cwd=%s)", cmd_str, workspace)

        cmd_t0 = time.time()
        result = subprocess.run(
            ["sh", "-c", cmd_str],
            cwd=workspace, capture_output=True, text=True,
        )
        cmd_elapsed = time.time() - cmd_t0

        if result.stdout:
            logger.info("[image_editor] stdout: %s", result.stdout.strip())
        if result.stderr:
            logger.warning("[image_editor] stderr: %s", result.stderr.strip())

        if result.returncode != 0:
            logger.error("[image_editor] p2pconvertor failed: exit_code=%d in %.2fs", result.returncode, cmd_elapsed)
        else:
            logger.info("[image_editor] p2pconvertor completed: exit_code=0 in %.2fs", cmd_elapsed)