ai-flow-python/app/util/task_util.py

import logging
import os
import time

from app.util.file_util import copy_directory, delete_directory

logger = logging.getLogger(__name__)

BASE_DIR = "data/data"


def create_task_workspace(task_id: str, template_path: str) -> str:
    workspace = os.path.join(BASE_DIR, task_id)
    logger.info("[%s] Creating workspace: %s  (from template: %s)", task_id, workspace, template_path)
    t0 = time.time()
    copy_directory(template_path, workspace)
    elapsed = time.time() - t0
    logger.info("[%s] Workspace ready in %.2fs", task_id, elapsed)
    return workspace


def cleanup_workspace(workspace: str, task_id: str):
    if not workspace or not workspace.strip():
        return
    try:
        if os.path.isdir(workspace):
            logger.info("[%s] Cleaning up workspace: %s", task_id, workspace)
            delete_directory(workspace)
            logger.info("[%s] Workspace cleaned up", task_id)
    except Exception as e:
        logger.error("[%s] Error cleaning up workspace %s: %s", task_id, workspace, e)