| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- 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 base64_content
- logger = logging.getLogger(__name__)
- class Base64EncodeTask(BaseTask):
- def get_type(self) -> str:
- return "base64_encode"
- 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 encoding")
- if not param.workspace:
- return TaskResult.failure("Workspace path is not provided")
- logger.info("[base64_encode] Starting: %d files to encode", len(files))
- t0 = time.time()
- encoded: list[str] = []
- for i, fp in enumerate(files):
- try:
- full_path = os.path.join(param.workspace, fp)
- if not os.path.isfile(full_path):
- logger.warning("[base64_encode] File %d/%d does not exist: %s", i + 1, len(files), full_path)
- continue
- file_size = os.path.getsize(full_path)
- b64 = base64_content(full_path)
- encoded.append(b64)
- logger.info("[base64_encode] Encoded file %d/%d: %s (file_size=%d, base64_len=%d)",
- i + 1, len(files), fp, file_size, len(b64))
- except Exception as e:
- logger.error("[base64_encode] Failed to encode file %d/%d %s: %s", i + 1, len(files), fp, e)
- if not encoded:
- return TaskResult.failure("No files were successfully encoded")
- logger.info("[base64_encode] Completed: %d/%d encoded in %.2fs",
- len(encoded), len(files), time.time() - t0)
- return TaskResult.success({"files": encoded})
- except Exception as e:
- logger.error("[base64_encode] Error: %s", e, exc_info=True)
- return TaskResult.failure(f"Base64 encoding failed: {e}")
|
