Feat: Moving files in file browser should trigger item update if exists and update sidecars as well.

This commit is contained in:
arabcoders 2025-11-14 20:05:58 +03:00
parent 1c7444e232
commit 4a69b0b6a1
4 changed files with 329 additions and 31 deletions

View file

@ -824,6 +824,79 @@ def rename_file(old_path: Path, new_name: str) -> tuple[Path, list[tuple[Path, P
raise
def move_file(old_path: Path, target_dir: Path) -> tuple[Path, list[tuple[Path, Path]]]:
"""
Move a file along with all its sidecar files to a target directory.
Args:
old_path (Path): The original file path.
target_dir (Path): The target directory to move the file to.
Returns:
tuple[Path, list[tuple[Path, Path]]]: Tuple of (new_path, list of (old_sidecar_path, new_sidecar_path) tuples).
Raises:
OSError: If moving fails.
ValueError: If target directory doesn't exist or if destination would cause collision.
"""
if not target_dir.exists():
msg: str = f"Target directory '{target_dir}' does not exist"
raise ValueError(msg)
if not target_dir.is_dir():
msg = f"Target '{target_dir}' is not a directory"
raise ValueError(msg)
new_path: Path = target_dir / old_path.name
if new_path.exists():
msg = f"Destination '{new_path}' already exists"
raise ValueError(msg)
sidecar_data: dict[str, dict] = get_file_sidecar(old_path)
sidecar_files: list[Path] = []
for category in sidecar_data.values():
sidecar_files.extend([item["file"] for item in category if "file" in item and isinstance(item["file"], Path)])
renamed_sidecars: list[tuple[Path, Path]] = []
move_operations: list[tuple[Path, Path]] = []
for sidecar in sidecar_files:
new_sidecar_path: Path = target_dir / sidecar.name
if new_sidecar_path.exists():
msg = f"Sidecar destination '{new_sidecar_path}' already exists"
raise ValueError(msg)
move_operations.append((sidecar, new_sidecar_path))
try:
moved_main: Path = old_path.rename(new_path)
for old_sidecar, new_sidecar in move_operations:
try:
moved_sidecar: Path = old_sidecar.rename(new_sidecar)
renamed_sidecars.append((old_sidecar, moved_sidecar))
except OSError as e:
LOG.error(f"Failed to move sidecar '{old_sidecar}': {e}")
try:
moved_main.rename(old_path)
# Rollback any sidecars that were already moved
for rolled_back_old, rolled_back_new in renamed_sidecars:
try:
rolled_back_new.rename(rolled_back_old)
except OSError:
LOG.error(f"Failed to rollback sidecar move from '{rolled_back_new}' to '{rolled_back_old}'")
except OSError:
LOG.error(f"Failed to rollback main file move from '{moved_main}' to '{old_path}'")
raise
return moved_main, renamed_sidecars
except OSError as e:
LOG.error(f"Failed to move '{old_path}' to '{new_path}': {e}")
raise
def get_possible_images(dir: str) -> list[dict]:
images: list = []

View file

@ -14,7 +14,7 @@ from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.ffprobe import ffprobe
from app.library.router import route
from app.library.Utils import delete_dir, get_file, get_file_sidecar, get_files, get_mime_type, rename_file
from app.library.Utils import delete_dir, get_file, get_file_sidecar, get_files, get_mime_type, move_file, rename_file
LOG: logging.Logger = logging.getLogger(__name__)
@ -364,15 +364,19 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
continue
try:
# Rename main file and all sidecar files
renamed, renamed_sidecars = rename_file(path, new_name)
sidecar_count: int = len(renamed_sidecars)
sidecar_info: str = (
f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})"
if sidecar_count > 0
else ""
)
sidecar_count: int = 0
sidecar_info: str = ""
sidecar_renamed: list[tuple[Path, Path]] = []
if path.is_dir():
renamed: Path = path.rename(new_path)
else:
renamed, sidecar_renamed = rename_file(path, new_name)
sidecar_count: int = len(sidecar_renamed)
sidecar_info: str = (
f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})"
if sidecar_count > 0
else ""
)
LOG.info(
f"Renamed '{path.relative_to(config.download_path)}' to '{renamed.relative_to(config.download_path)}'{sidecar_info}"
@ -386,21 +390,16 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
continue
else:
extra_info: dict[str, Path] = {"new_path": renamed}
if renamed_sidecars:
extra_info["sidecar_count"] = len(renamed_sidecars)
if sidecar_renamed:
extra_info["sidecar_count"] = len(sidecar_renamed)
record(path, ok=True, action=action, extra=extra_info)
for old_sidecar, new_sidecar in renamed_sidecars:
record(
old_sidecar,
ok=True,
action="rename_sidecar",
extra={"new_path": new_sidecar, "parent": str(path.relative_to(config.download_path))},
)
for old_sidecar, new_sidecar in sidecar_renamed:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(renamed.relative_to(config.download_path))
if len(renamed_sidecars) > 0:
if sidecar_renamed:
item.info.get_file_sidecar()
queue.done.put(item)
@ -432,7 +431,7 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
continue
raw_new: str = unquote_plus(str(new_path)).strip()
target_dir = (
target_dir: Path = (
Path(config.download_path)
if not raw_new or raw_new in ("/", ".")
else Path(config.download_path).joinpath(raw_new.lstrip("/"))
@ -446,7 +445,7 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
)
continue
root_real = Path(config.download_path).resolve()
root_real: Path = Path(config.download_path).resolve()
if not target_dir.exists() or not target_dir.is_dir():
record(
path, ok=False, error="Destination path is invalid.", action=action, extra={"new_path": target_dir}
@ -468,14 +467,49 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
continue
try:
dest = target_dir.joinpath(path.name)
path.rename(dest)
sidecar_count: int = 0
sidecar_moved: list[tuple[Path, Path]] = []
sidecar_info: str = ""
if path.is_dir():
dest: Path = target_dir.joinpath(path.name)
moved: Path = path.rename(dest)
else:
moved, sidecar_moved = move_file(path, target_dir)
sidecar_count: int = len(sidecar_moved)
sidecar_info: str = (
f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})"
if sidecar_count > 0
else ""
)
LOG.info(
f"Moved '{path.relative_to(config.download_path)}' to '{moved.relative_to(config.download_path)}'{sidecar_info}"
)
except OSError as e:
LOG.exception(e)
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
except ValueError as e:
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
else:
record(path, ok=True, action=action, extra={"new_path": dest})
extra_info: dict[str, Path] = {"new_path": moved}
if sidecar_moved:
extra_info["sidecar_count"] = len(sidecar_moved)
record(path, ok=True, action=action, extra=extra_info)
for old_sidecar, new_sidecar in sidecar_moved:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(moved.relative_to(config.download_path))
if sidecar_moved:
item.info.get_file_sidecar()
queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(data=operations_status, status=web.HTTPOk.status_code)

View file

@ -39,6 +39,7 @@ from app.library.Utils import (
load_cookies,
load_modules,
merge_dict,
move_file,
parse_tags,
read_logfile,
rename_file,
@ -2189,3 +2190,189 @@ class TestRenameFile:
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "renamed.en-US.ass" in sidecar_names
assert "renamed.thumb.jpg" in sidecar_names
class TestMoveFile:
"""Test move_file function."""
def test_move_single_file_no_sidecars(self, tmp_path: Path):
"""Test moving a single file without sidecar files."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 0 == len(sidecars)
def test_move_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test moving a file with subtitle sidecar."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_file = source_dir / "video.en.srt"
subtitle_file.write_text("test subtitle")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 1 == len(sidecars)
old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists()
assert "video.en.srt" == new_sidecar.name
assert new_sidecar.parent == target_dir
assert old_sidecar == subtitle_file
assert not subtitle_file.exists()
def test_move_file_with_multiple_sidecars(self, tmp_path: Path):
"""Test moving a file with multiple sidecar files."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_en = source_dir / "video.en.srt"
subtitle_en.write_text("english subtitle")
subtitle_fr = source_dir / "video.fr.srt"
subtitle_fr.write_text("french subtitle")
info_file = source_dir / "video.info.json"
info_file.write_text('{"title": "test"}')
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 3 == len(sidecars)
# Check all sidecars were moved
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "video.en.srt" in sidecar_names
assert "video.fr.srt" in sidecar_names
assert "video.info.json" in sidecar_names
# Check all are in target directory
for _old_sidecar, new_sidecar in sidecars:
assert new_sidecar.parent == target_dir
# Check old files don't exist
assert not subtitle_en.exists()
assert not subtitle_fr.exists()
assert not info_file.exists()
def test_move_file_destination_exists(self, tmp_path: Path):
"""Test moving a file when destination already exists."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "target"
target_dir.mkdir()
existing_file = target_dir / "video.mp4"
existing_file.write_text("existing content")
# Should raise ValueError
with pytest.raises(ValueError, match="already exists"):
move_file(test_file, target_dir)
# Original files should still exist
assert test_file.exists()
assert existing_file.exists()
def test_move_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test moving when sidecar destination already exists."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_file = source_dir / "video.en.srt"
subtitle_file.write_text("test subtitle")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Create conflicting sidecar destination
conflicting_sidecar = target_dir / "video.en.srt"
conflicting_sidecar.write_text("existing subtitle")
# Should raise ValueError
with pytest.raises(ValueError, match=r"Sidecar destination.*already exists"):
move_file(test_file, target_dir)
# Original files should still exist
assert test_file.exists()
assert subtitle_file.exists()
assert conflicting_sidecar.exists()
def test_move_file_target_not_directory(self, tmp_path: Path):
"""Test moving when target is not a directory."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
# Create a file (not directory) as target
target_file = tmp_path / "target.txt"
target_file.write_text("not a directory")
# Should raise ValueError
with pytest.raises(ValueError, match="not a directory"):
move_file(test_file, target_file)
# Original file should still exist
assert test_file.exists()
def test_move_file_target_does_not_exist(self, tmp_path: Path):
"""Test moving when target directory doesn't exist."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "nonexistent"
# Should raise ValueError
with pytest.raises(ValueError, match="does not exist"):
move_file(test_file, target_dir)
# Original file should still exist
assert test_file.exists()

View file

@ -726,9 +726,10 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
}
if ('rename' === action) {
const moveSideCars = 'file' === item.type ? ' (and its sidecars)' : ''
const { status, value: newName } = await dialog.promptDialog({
title: 'Rename Item',
message: `Enter new name for '${item.name}' Sidecars will be renamed as well:`,
message: `Enter new name for '${item.name}'${moveSideCars}:`,
initial: item.name,
confirmText: 'Rename',
cancelText: 'Cancel',
@ -775,9 +776,10 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
}
if ('move' === action) {
const moveSideCars = 'file' === item.type ? ' (and its sidecars)' : ''
const { status, value: newPath } = await dialog.promptDialog({
title: 'Move Item',
message: `Enter new path for '${item.name}':`,
message: `Enter new path for '${item.name}'${moveSideCars}:`,
initial: item.path.replace(/[^/]+$/, '') || '/',
confirmText: 'Move',
cancelText: 'Cancel',
@ -792,10 +794,12 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
await actionRequest(item, 'move', { new_path: new_path }, (item, _, data) => {
items.value = items.value.filter(i => i.path !== item.path)
toast.success(`Moved '${item.name}' to '${data.new_path}'.`)
})
await actionRequest(item, 'move', { new_path: new_path }, (item, _, data, source) => {
if (item.path === source.path) {
toast.success(`Moved '${item.name}' to '${data.new_path}'.`)
}
items.value = items.value.filter(i => i.path !== data.path)
}, true)
return
}