kodi
131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from backend.app.api.errors import AppError
|
|
from backend.app.api.schemas import TaskCreateResponse
|
|
from backend.app.db.task_repository import TaskRepository
|
|
from backend.app.security.path_guard import PathGuard
|
|
from backend.app.tasks_runner import TaskRunner
|
|
|
|
|
|
class MoveTaskService:
|
|
def __init__(self, path_guard: PathGuard, repository: TaskRepository, runner: TaskRunner):
|
|
self._path_guard = path_guard
|
|
self._repository = repository
|
|
self._runner = runner
|
|
|
|
def create_move_task(self, source: str, destination: str) -> TaskCreateResponse:
|
|
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 = self._path_guard.resolve_path(destination)
|
|
destination_parent = resolved_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 source_is_directory and resolved_destination.absolute == resolved_source.absolute:
|
|
raise AppError(
|
|
code="invalid_request",
|
|
message="Destination must differ from source",
|
|
status_code=400,
|
|
details={"path": source, "destination": destination},
|
|
)
|
|
|
|
if resolved_destination.absolute.exists():
|
|
raise AppError(
|
|
code="already_exists",
|
|
message="Target path already exists",
|
|
status_code=409,
|
|
details={"path": resolved_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},
|
|
)
|
|
if self._is_nested_destination(resolved_source.absolute, resolved_destination.absolute):
|
|
raise AppError(
|
|
code="invalid_request",
|
|
message="Destination cannot be inside source",
|
|
status_code=400,
|
|
details={"path": source, "destination": destination},
|
|
)
|
|
|
|
task = self._repository.create_task(
|
|
operation="move",
|
|
source=resolved_source.relative,
|
|
destination=resolved_destination.relative,
|
|
)
|
|
self._runner.enqueue_move_directory(
|
|
task_id=task["id"],
|
|
source=str(resolved_source.absolute),
|
|
destination=str(resolved_destination.absolute),
|
|
)
|
|
return TaskCreateResponse(task_id=task["id"], status=task["status"])
|
|
|
|
total_bytes = int(resolved_source.absolute.stat().st_size)
|
|
task = self._repository.create_task(
|
|
operation="move",
|
|
source=resolved_source.relative,
|
|
destination=resolved_destination.relative,
|
|
)
|
|
|
|
self._runner.enqueue_move_file(
|
|
task_id=task["id"],
|
|
source=str(resolved_source.absolute),
|
|
destination=str(resolved_destination.absolute),
|
|
total_bytes=total_bytes,
|
|
same_root=same_root,
|
|
)
|
|
|
|
return TaskCreateResponse(task_id=task["id"], status=task["status"])
|
|
|
|
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 _is_nested_destination(source: Path, destination: Path) -> bool:
|
|
try:
|
|
destination.relative_to(source)
|
|
return True
|
|
except ValueError:
|
|
return False
|