kodi
297 lines
11 KiB
Python
297 lines
11 KiB
Python
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))
|
|
has_directory = any(resolved_source.absolute.is_dir() for resolved_source in resolved_sources)
|
|
if root_alias != resolved_destination_base.alias and has_directory:
|
|
raise AppError(
|
|
code="invalid_request",
|
|
message="Cross-root batch move with directories 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")
|