47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
"""Temporary file management for web operations."""
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from backend.config import settings
|
|
|
|
|
|
class FileManager:
|
|
@staticmethod
|
|
def get_task_dir(task_id: str) -> Path:
|
|
return settings.temp_dir / task_id
|
|
|
|
@staticmethod
|
|
def get_input_path(task_id: str) -> Path:
|
|
task_dir = FileManager.get_task_dir(task_id)
|
|
for f in task_dir.iterdir():
|
|
if f.name.startswith("input."):
|
|
return f
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_output_dir(task_id: str) -> Path:
|
|
return FileManager.get_task_dir(task_id) / "output"
|
|
|
|
@staticmethod
|
|
def ensure_output_dir(task_id: str) -> Path:
|
|
output_dir = FileManager.get_output_dir(task_id)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
return output_dir
|
|
|
|
@staticmethod
|
|
def cleanup_task(task_id: str) -> None:
|
|
task_dir = FileManager.get_task_dir(task_id)
|
|
if task_dir.exists():
|
|
shutil.rmtree(task_dir, ignore_errors=True)
|
|
|
|
@staticmethod
|
|
def get_output_files(task_id: str) -> list:
|
|
output_dir = FileManager.get_output_dir(task_id)
|
|
if not output_dir.exists():
|
|
return []
|
|
files = []
|
|
for f in output_dir.iterdir():
|
|
if f.is_file():
|
|
files.append({"filename": f.name, "size": f.stat().st_size})
|
|
return files |