feat: feedback verbetering 01
This commit is contained in:
Binary file not shown.
@@ -1,6 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import uuid
|
||||
@@ -136,8 +135,11 @@ class MoveTaskService:
|
||||
"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
|
||||
],
|
||||
@@ -226,12 +228,12 @@ class MoveTaskService:
|
||||
details={"path": source, "destination": destination_relative},
|
||||
)
|
||||
|
||||
progress_label = resolved_source.absolute.name
|
||||
if source_is_directory:
|
||||
directories, files = self._build_directory_plan(
|
||||
resolved_source=resolved_source,
|
||||
destination_root=destination_absolute,
|
||||
include_root_prefix=include_root_prefix,
|
||||
)
|
||||
files = []
|
||||
directories = []
|
||||
if include_root_prefix:
|
||||
progress_label = resolved_source.absolute.name
|
||||
else:
|
||||
files = [
|
||||
{
|
||||
@@ -241,6 +243,7 @@ class MoveTaskService:
|
||||
}
|
||||
]
|
||||
directories = []
|
||||
progress_label = files[0]["label"]
|
||||
|
||||
return {
|
||||
"source_relative": resolved_source.relative,
|
||||
@@ -252,6 +255,8 @@ class MoveTaskService:
|
||||
"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:
|
||||
@@ -271,70 +276,6 @@ class MoveTaskService:
|
||||
def _join_destination_base(destination_base: str, name: str) -> str:
|
||||
return f"{destination_base.rstrip('/')}/{name}" if destination_base.rstrip("/") else f"/{name}"
|
||||
|
||||
def _build_directory_plan(
|
||||
self,
|
||||
*,
|
||||
resolved_source: ResolvedPath,
|
||||
destination_root: Path,
|
||||
include_root_prefix: bool,
|
||||
) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
|
||||
directories: list[dict[str, str]] = [
|
||||
{
|
||||
"source": str(resolved_source.absolute),
|
||||
"destination": str(destination_root),
|
||||
}
|
||||
]
|
||||
files: list[dict[str, str]] = []
|
||||
for root, dirnames, filenames in os.walk(resolved_source.absolute, followlinks=False):
|
||||
root_path = Path(root)
|
||||
dirnames.sort(key=str.lower)
|
||||
filenames.sort(key=str.lower)
|
||||
for name in dirnames:
|
||||
entry = root_path / name
|
||||
if entry.is_symlink():
|
||||
raise AppError(
|
||||
code="type_conflict",
|
||||
message="Source directory must not contain symlinks",
|
||||
status_code=409,
|
||||
details={"path": resolved_source.relative},
|
||||
)
|
||||
relative = entry.relative_to(resolved_source.absolute)
|
||||
directories.append(
|
||||
{
|
||||
"source": str(entry),
|
||||
"destination": str(destination_root / relative),
|
||||
}
|
||||
)
|
||||
for name in filenames:
|
||||
entry = root_path / name
|
||||
if entry.is_symlink():
|
||||
raise AppError(
|
||||
code="type_conflict",
|
||||
message="Source directory must not contain symlinks",
|
||||
status_code=409,
|
||||
details={"path": resolved_source.relative},
|
||||
)
|
||||
relative = entry.relative_to(resolved_source.absolute)
|
||||
files.append(
|
||||
{
|
||||
"source": str(entry),
|
||||
"destination": str(destination_root / relative),
|
||||
"label": self._progress_label(
|
||||
top_level_name=resolved_source.absolute.name,
|
||||
relative_path=relative,
|
||||
include_root_prefix=include_root_prefix,
|
||||
),
|
||||
}
|
||||
)
|
||||
return directories, files
|
||||
|
||||
@staticmethod
|
||||
def _progress_label(*, top_level_name: str, relative_path: Path, include_root_prefix: bool) -> str:
|
||||
relative_value = relative_path.as_posix()
|
||||
if not relative_value:
|
||||
return top_level_name
|
||||
return f"{top_level_name}/{relative_value}" if include_root_prefix else relative_value
|
||||
|
||||
@staticmethod
|
||||
def _is_nested_destination(source: Path, destination: Path) -> bool:
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user