| 12345678910111213141516171819202122232425262728293031 |
- 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("Creating workspace: %s (from template: %s)", workspace, template_path)
- t0 = time.time()
- copy_directory(template_path, workspace)
- elapsed = time.time() - t0
- logger.info("Workspace ready in %.2fs", 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("Cleaning up workspace: %s", workspace)
- delete_directory(workspace)
- logger.info("Workspace cleaned up")
- except Exception as e:
- logger.error("Error cleaning up workspace %s: %s", workspace, e)
|
