| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- import logging
- import os
- 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 generate_unique_filename, base64_to_file
- logger = logging.getLogger(__name__)
- class Base64DecodeTask(BaseTask):
- def get_type(self) -> str:
- return "base64_decode"
- def execute(self, param: TaskParam) -> TaskResult:
- try:
- files_obj = param.inputs.get("files")
- files: list[str] = files_obj if isinstance(files_obj, list) else []
- if not files:
- return TaskResult.failure("No files provided for base64 decoding")
- if not param.workspace:
- return TaskResult.failure("Workspace path is not provided")
- logger.info("[base64_decode] Starting: %d files to decode", len(files))
- t0 = time.time()
- decoded_files: list[str] = []
- for i, b64 in enumerate(files):
- try:
- name = generate_unique_filename(".png")
- rel_path = os.path.join(param.step_id, name)
- abs_path = os.path.join(param.workspace, rel_path)
- base64_to_file(b64, abs_path)
- decoded_files.append(rel_path)
- logger.info("[base64_decode] Decoded file %d/%d -> %s (base64_len=%d)",
- i + 1, len(files), rel_path, len(b64))
- except Exception as e:
- logger.error("[base64_decode] Failed to decode file %d/%d: %s", i + 1, len(files), e)
- if not decoded_files:
- return TaskResult.failure("No files were successfully decoded")
- logger.info("[base64_decode] Completed: %d/%d decoded in %.2fs",
- len(decoded_files), len(files), time.time() - t0)
- return TaskResult.success({"files": decoded_files})
- except Exception as e:
- logger.error("[base64_decode] Error: %s", e, exc_info=True)
- return TaskResult.failure(f"Base64 decoding failed: {e}")
|
