from __future__ import annotations from pathlib import Path import uuid from backend.app.api.errors import AppError from backend.app.api.schemas import TaskCreateResponse from backend.app.db.history_repository import HistoryRepository from backend.app.db.task_repository import TaskRepository from backend.app.security.path_guard import PathGuard, ResolvedPath from backend.app.tasks_runner import TaskRunner class MoveTaskService: def __init__(self, path_guard: PathGuard, repository: TaskRepository, runner: TaskRunner, history_repository: HistoryRepository | None = None): self._path_guard = path_guard self._repository = repository self._runner = runner self._history_repository = history_repository def create_move_task(self, source: str | None, destination: str | None) -> TaskCreateResponse: if not source or not destination: raise AppError( code="invalid_request", message="Source and destination are required", status_code=400, ) try: item = self._build_move_item(source=source, destination=destination) task_id = str(uuid.uuid4()) task = self._repository.create_task( operation="move", source=item["source_relative"], destination=item["destination_relative"], task_id=task_id, ) self._record_history( entry_id=task_id, operation="move", status="queued", source=item["source_relative"], destination=item["destination_relative"], ) if item["kind"] == "directory": self._runner.enqueue_move_directory(task_id=task["id"], item=item) else: self._runner.enqueue_move_file( task_id=task["id"], source=item["source_absolute"], destination=item["destination_absolute"], total_bytes=item["total_bytes"], same_root=item["same_root"], current_item=item["files"][0]["label"], ) return TaskCreateResponse(task_id=task["id"], status=task["status"]) except AppError as exc: self._record_history( operation="move", status="failed", source=source, destination=destination, error_code=exc.code, error_message=exc.message, finished_at=self._now_iso(), ) raise def create_batch_move_task(self, sources: list[str] | None, destination_base: str | None) -> TaskCreateResponse: if not sources or len(sources) < 2: raise AppError( code="invalid_request", message="Batch move requires at least 2 sources", status_code=400, ) if not destination_base: raise AppError( code="invalid_request", message="Destination base is required", status_code=400, ) resolved_destination_base = self._path_guard.resolve_directory_path(destination_base) items: list[dict] = [] resolved_sources = [self._path_guard.resolve_existing_path(source) for source in sources] source_aliases = {resolved_source.alias for resolved_source in resolved_sources} if len(source_aliases) != 1: raise AppError( code="invalid_request", message="Batch move requires all selected items to be in the same root", status_code=400, ) root_alias = next(iter(source_aliases)) if root_alias != resolved_destination_base.alias: raise AppError( code="invalid_request", message="Cross-root batch directory move is not supported in v1", status_code=400, details={"destination_base": destination_base}, ) for source, source_resolved in zip(sources, resolved_sources): destination = self._join_destination_base(destination_base, source_resolved.absolute.name) item = self._build_move_item( source=source, destination=destination, resolved_destination=resolved_destination_base, destination_base=destination_base, include_root_prefix=True, ) items.append(item) task_id = str(uuid.uuid4()) task = self._repository.create_task( operation="move", source=f"{len(items)} items", destination=resolved_destination_base.relative, task_id=task_id, ) self._record_history( entry_id=task_id, operation="move", status="queued", source=f"{len(items)} items", destination=resolved_destination_base.relative, ) self._runner.enqueue_move_batch( task_id=task["id"], items=[ { "source": item["source_absolute"], "destination": item["destination_absolute"], "kind": item["kind"], "same_root": item["same_root"], "files": item["files"], "directories": item["directories"], "progress_total_items": item["progress_total_items"], "progress_label": item["progress_label"], } for item in items ], ) return TaskCreateResponse(task_id=task["id"], status=task["status"]) def _build_move_item( self, source: str, destination: str, resolved_destination: ResolvedPath | None = None, destination_base: str | None = None, include_root_prefix: bool = False, ) -> dict: resolved_source = self._path_guard.resolve_existing_path(source) _, _, lexical_source, _ = self._path_guard.resolve_lexical_path(source) if lexical_source.is_symlink(): raise AppError( code="type_conflict", message="Source must not be a symlink", status_code=409, details={"path": source}, ) source_is_file = resolved_source.absolute.is_file() source_is_directory = resolved_source.absolute.is_dir() if not source_is_file and not source_is_directory: raise AppError( code="type_conflict", message="Unsupported source path type", status_code=409, details={"path": source}, ) resolved_destination = resolved_destination or self._path_guard.resolve_path(destination) destination_absolute = ( resolved_destination.absolute / resolved_source.absolute.name if destination_base is not None else resolved_destination.absolute ) destination_relative = self._path_guard.entry_relative_path( resolved_destination.alias, destination_absolute, display_style=resolved_destination.display_style, ) destination_parent = destination_absolute.parent parent_relative = self._path_guard.entry_relative_path( resolved_destination.alias, destination_parent, display_style=resolved_destination.display_style, ) self._map_directory_validation(parent_relative) if destination_absolute == resolved_source.absolute: raise AppError( code="invalid_request", message="Destination must differ from source", status_code=400, details={"path": source, "destination": destination_relative}, ) if destination_absolute.exists(): raise AppError( code="already_exists", message="Target path already exists", status_code=409, details={"path": destination_relative}, ) same_root = resolved_source.alias == resolved_destination.alias if source_is_directory: if not same_root: raise AppError( code="invalid_request", message="Cross-root directory move is not supported in v1", status_code=400, details={"path": source, "destination": destination_relative}, ) if self._is_nested_destination(resolved_source.absolute, destination_absolute): raise AppError( code="invalid_request", message="Destination cannot be inside source", status_code=400, details={"path": source, "destination": destination_relative}, ) progress_label = resolved_source.absolute.name if source_is_directory: files = [] directories = [] if include_root_prefix: progress_label = resolved_source.absolute.name else: files = [ { "source": str(resolved_source.absolute), "destination": str(destination_absolute), "label": resolved_source.absolute.name, } ] directories = [] progress_label = files[0]["label"] return { "source_relative": resolved_source.relative, "destination_relative": destination_relative, "source_absolute": str(resolved_source.absolute), "destination_absolute": str(destination_absolute), "kind": "directory" if source_is_directory else "file", "same_root": same_root, "total_bytes": int(resolved_source.absolute.stat().st_size) if source_is_file else None, "files": files, "directories": directories, "progress_total_items": 1, "progress_label": progress_label, } def _map_directory_validation(self, relative_path: str) -> None: try: self._path_guard.resolve_directory_path(relative_path) except AppError as exc: if exc.code == "path_type_conflict": raise AppError( code="type_conflict", message="Destination parent is not a directory", status_code=409, details=exc.details, ) raise @staticmethod def _join_destination_base(destination_base: str, name: str) -> str: return f"{destination_base.rstrip('/')}/{name}" if destination_base.rstrip("/") else f"/{name}" @staticmethod def _is_nested_destination(source: Path, destination: Path) -> bool: try: destination.relative_to(source) return True except ValueError: return False def _record_history(self, **kwargs) -> None: if self._history_repository: self._history_repository.create_entry(**kwargs) @staticmethod def _now_iso() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")