files: split FileWorker move and reorganize into mixins

FileWorker stays the queue, status, and refresh apply path.  Move planning
and execute/revert live in wrolpi.files.move; collection reorganize jobs
live in wrolpi.files.reorganize_job.  Perpetual-process dispatch is
unchanged.

Existing `from wrolpi.files.worker import build_move_plan_bulk` still works.

wrolpi/files/move.py

@@ -0,0 +1,598 @@
+"""FileWorker move jobs: plan, execute, revert.
+
+Extracted from worker.py so the queue/status class is not mixed with move logic.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import pathlib
+import shutil
+import stat as stat_module
+from collections import OrderedDict
+from pathlib import Path
+from typing import Callable, Dict, List, Set, Tuple
+
+from sqlalchemy import text, or_
+
+from wrolpi import flags
+from wrolpi.common import get_media_directory, logger, walk, unique_by_predicate, chunks
+from wrolpi.db import get_db_session, get_db_curs
+from wrolpi.errors import NoPrimaryFile
+from wrolpi.events import Events
+from wrolpi.files.lib import (
+    split_path_stem_and_suffix, get_unique_files_by_stem, glob_shared_stem,
+    group_files_by_stem, get_primary_file, delete_directory,
+    _move_file_group_files, _bulk_update_file_groups_db, MOVE_CHUNK_SIZE,
+    get_normalized_ignored_directories, remove_files_in_ignored_directories,
+)
+
+logger = logger.getChild(__name__)
+
+
+async def build_move_plan_bulk(
+        sources: list[pathlib.Path],
+        destination: pathlib.Path,
+        progress_callback: Callable[[int, int], None] = None,
+) -> Tuple[Dict[pathlib.Path, pathlib.Path], list[pathlib.Path]]:
+    """
+    Build move plan using bulk SQL operations instead of per-source queries.
+
+    Uses a temp table pattern similar to compare_file_groups for O(1) query complexity.
+
+    Args:
+        sources: List of file/directory paths to move
+        destination: Target directory
+        progress_callback: Optional callback(processed, total) called during file collection
+
+    Returns:
+        Tuple of (plan dict mapping old_primary_path -> new_primary_path,
+                  list of old_directories for cleanup)
+    """
+    from wrolpi.files.models import FileGroup
+
+    plan: Dict[pathlib.Path, pathlib.Path] = dict()
+    old_directories: list[pathlib.Path] = list(unique_by_predicate(
+        i if i.is_dir() else i.parent for i in sources
+    ))
+
+    # Deduplicate sources by stem
+    sources = list(get_unique_files_by_stem(sources))
+    logger.info(f'build_move_plan_bulk: processing {len(sources)} sources')
+
+    # Partition sources into files, directories, and deleted (pre-moved)
+    # Optimized: single stat() call per path instead of up to 3 (is_file, is_dir, exists)
+    file_sources: list[pathlib.Path] = []
+    dir_sources: list[pathlib.Path] = []
+    deleted_sources: list[pathlib.Path] = []
+    total_sources = len(sources)
+
+    for i, source in enumerate(sources, 1):
+        try:
+            st = source.stat()
+            if stat_module.S_ISREG(st.st_mode):
+                file_sources.append(source)
+            elif stat_module.S_ISDIR(st.st_mode):
+                dir_sources.append(source)
+            # Other types (symlinks, etc.) are silently skipped
+        except FileNotFoundError:
+            deleted_sources.append(source)
+
+        # Report progress during source partitioning (first 50% of planning)
+        if progress_callback and i % 100 == 0:
+            progress_callback(i, total_sources * 2)  # *2 because file collection is second half
+
+    # Collect all paths to insert: file paths + expanded directory contents.  This walks the whole
+    # source tree, which is slow on Pi storage, so it must finish before the DB session opens - the
+    # session holds the SQLite write lock for its lifetime and would starve every other writer.
+    all_file_paths: list[pathlib.Path] = []
+    files_collected = 0
+    # Use total_sources as the denominator for the second half of progress
+    total_items = len(file_sources) + len(dir_sources)
+
+    # For files, collect the file and its shared-stem siblings
+    file_source_set: Set[str] = set()
+    for i, source in enumerate(file_sources, 1):
+        files = glob_shared_stem(source)
+        all_file_paths.extend(files)
+        file_source_set.update(str(f) for f in files)
+        files_collected += 1
+        # Report progress (second half: 50-100%)
+        if progress_callback and files_collected % 50 == 0:
+            progress = total_sources + (files_collected * total_sources // max(total_items, 1))
+            progress_callback(progress, total_sources * 2)
+
+    # For directories, walk and collect all files
+    dir_file_mapping: Dict[pathlib.Path, pathlib.Path] = {}  # file -> source_dir
+    for i, source_dir in enumerate(dir_sources, 1):
+        for f in walk(source_dir):
+            if f.is_file():
+                all_file_paths.append(f)
+                dir_file_mapping[f] = source_dir
+        files_collected += 1
+        # Report progress during directory walking (every 10 directories)
+        if progress_callback and files_collected % 10 == 0:
+            progress = total_sources + (files_collected * total_sources // max(total_items, 1))
+            progress_callback(progress, total_sources * 2)
+
+    # Final progress update before deduplication
+    if progress_callback:
+        progress_callback(total_sources * 2, total_sources * 2)
+
+    # Deduplicate
+    all_file_paths = list(set(all_file_paths))
+
+    # Planning reads FileGroups and then INSERTs the ones missing from the DB, so the transaction
+    # must begin as a writer.  A deferred transaction upgrading to a writer mid-way gets "database
+    # is locked" *immediately* (busy_timeout is skipped for lock upgrades), so any concurrent
+    # writer - a refresh, a download, a config save - would abort the whole move.
+    with get_db_session(commit=True) as session:
+        try:
+            # Create temp table for source paths.  SQLite has no ON COMMIT DROP; the table is
+            # explicitly dropped below on success/failure (and IF NOT EXISTS + DELETE guard
+            # against any leftover from a previous failed run on the same connection).
+            session.execute(text("""
+                                 CREATE TEMP TABLE IF NOT EXISTS move_sources
+                                 (
+                                     source_path      TEXT PRIMARY KEY,
+                                     source_type      TEXT NOT NULL,
+                                     source_directory TEXT
+                                 )
+                                 """))
+            session.execute(text("DELETE FROM move_sources"))
+
+            # Bulk insert file paths into temp table
+            if all_file_paths:
+                session.execute(
+                    text("""
+                         INSERT INTO move_sources (source_path, source_type, source_directory)
+                         VALUES (:path, :type, :dir)
+                         ON CONFLICT DO NOTHING
+                         """),
+                    [{"path": str(p), "type": 'file', "dir": str(p.parent)} for p in all_file_paths],
+                )
+
+            # Also insert deleted sources (for pre-moved files lookup)
+            if deleted_sources:
+                session.execute(
+                    text("""
+                         INSERT INTO move_sources (source_path, source_type, source_directory)
+                         VALUES (:path, :type, :dir)
+                         ON CONFLICT DO NOTHING
+                         """),
+                    [{"path": str(p), "type": 'deleted', "dir": str(p.parent)} for p in deleted_sources],
+                )
+
+            # Single bulk query to get all FileGroups matching our sources
+            result = session.execute(text("""
+                                          SELECT fg.id, fg.primary_path, fg.directory, ms.source_path, ms.source_type
+                                          FROM file_group fg
+                                                   JOIN move_sources ms ON fg.primary_path = ms.source_path
+                                          """))
+
+            # Build lookup of found FileGroups
+            fg_by_path: Dict[str, Tuple[int, str, str]] = {}  # path -> (id, primary_path, directory)
+            for row in result:
+                fg_by_path[row.source_path] = (row.id, row.primary_path, row.directory)
+
+            logger.info(f'build_move_plan_bulk: found {len(fg_by_path)} FileGroups for {len(all_file_paths)} files')
+
+            # Process files from direct file sources (`file_source_set` was collected above)
+            for path_str, (fg_id, primary_path, directory) in fg_by_path.items():
+                path = pathlib.Path(path_str)
+                primary_path = pathlib.Path(primary_path)
+
+                # Is this from a file source or a directory source?
+                if path_str in file_source_set:
+                    # Direct file source -> destination / filename
+                    new_path = destination / primary_path.name
+                elif path in dir_file_mapping:
+                    # From directory source -> destination / source_dir.name / relative_path
+                    source_dir = dir_file_mapping[path]
+                    new_path = destination / source_dir.name / primary_path.relative_to(source_dir)
+                else:
+                    # Deleted source (pre-moved) -> destination / filename
+                    new_path = destination / primary_path.name
+
+                # Check for conflicts (destination exists)
+                if new_path.exists():
+                    if not primary_path.exists():
+                        # Pre-moved - OK to proceed
+                        plan[primary_path] = new_path
+                    else:
+                        raise FileExistsError(f'Cannot move: {new_path} already exists')
+                else:
+                    plan[primary_path] = new_path
+
+            # Handle files not in DB - create FileGroups
+            found_paths = set(fg_by_path.keys())
+            missing_db_files = [p for p in all_file_paths if str(p) not in found_paths and p.exists()]
+
+            if missing_db_files:
+                logger.info(f'build_move_plan_bulk: creating FileGroups for {len(missing_db_files)} files not in DB')
+                for paths in group_files_by_stem(missing_db_files):
+                    try:
+                        get_primary_file(paths)
+                        fg = FileGroup.from_paths(session, *paths)
+                        session.flush()
+
+                        # Determine destination path
+                        if any(str(p) in file_source_set for p in paths):
+                            new_path = destination / fg.primary_path.name
+                        else:
+                            # Find which directory source this came from
+                            source_dir = None
+                            for p in paths:
+                                if p in dir_file_mapping:
+                                    source_dir = dir_file_mapping[p]
+                                    break
+                            if source_dir:
+                                new_path = destination / source_dir.name / fg.primary_path.relative_to(source_dir)
+                            else:
+                                new_path = destination / fg.primary_path.name
+
+                        if new_path.exists() and fg.primary_path.exists():
+                            raise FileExistsError(f'Cannot move: {new_path} already exists')
+                        plan[fg.primary_path] = new_path
+
+                    except NoPrimaryFile:
+                        for file in paths:
+                            fg = FileGroup.from_paths(session, file)
+                            session.flush()
+                            if any(str(p) in file_source_set for p in paths):
+                                new_path = destination / fg.primary_path.name
+                            elif file in dir_file_mapping:
+                                source_dir = dir_file_mapping[file]
+                                new_path = destination / source_dir.name / fg.primary_path.relative_to(source_dir)
+                            else:
+                                new_path = destination / fg.primary_path.name
+                            if new_path.exists() and fg.primary_path.exists():
+                                raise FileExistsError(f'Cannot move: {new_path} already exists')
+                            plan[fg.primary_path] = new_path
+
+            # Add directories to plan for cleanup
+            for source_dir in dir_sources:
+                for directory in (i for i in walk(source_dir) if i.is_dir()):
+                    new_directory = destination / source_dir.name / directory.relative_to(source_dir)
+                    plan[directory] = new_directory
+
+        except BaseException:
+            # SQLite has no ON COMMIT DROP; attempt to drop the temp table, but do NOT
+            # commit — partial plan work must not be persisted on failure.  (Outside of
+            # tests, get_db_session rolls back, which also undoes the CREATE TEMP TABLE;
+            # the IF NOT EXISTS + DELETE guard above handles any leftover.)
+            try:
+                session.execute(text("DROP TABLE IF EXISTS move_sources"))
+            except Exception:
+                logger.exception('Failed to drop move_sources temp table')
+            raise
+        else:
+            # Drop the temp table, then commit the plan work (SQLite has no ON COMMIT DROP).
+            session.execute(text("DROP TABLE IF EXISTS move_sources"))
+            session.commit()
+
+    # Sort plan by depth (deepest first) for safe move ordering
+    plan = OrderedDict(
+        sorted(plan.items(), key=lambda i: (len(i[0].parents), i[0].name), reverse=True)
+    )
+
+    logger.info(f'build_move_plan_bulk: plan has {len(plan)} items')
+    return plan, old_directories
+
+
+class FileMoveMixin:
+    def _validate_move_paths(
+            self,
+            sources: List[pathlib.Path],
+            destination: pathlib.Path,
+            media_directory: pathlib.Path,
+    ) -> bool:
+        """Validate that all source and destination paths are within the media directory.
+
+        Returns True if valid, False otherwise.
+        """
+        for source in sources:
+            if not str(source).startswith(str(media_directory)):
+                logger.error(f'{source} is not within the media directory')
+                self.reset_status()
+                return False
+        if not str(destination).startswith(str(media_directory)):
+            logger.error(f'{destination} is not within the media directory')
+            self.reset_status()
+            return False
+        return True
+
+    async def _execute_move_chunks(
+            self,
+            plan: Dict[pathlib.Path, pathlib.Path],
+            session,
+            created_directories: Set[pathlib.Path],
+            revert_plan: Dict[pathlib.Path, pathlib.Path],
+    ) -> Set[pathlib.Path]:
+        """Execute the move plan in chunks, updating progress.
+
+        Returns the set of new directories created.
+        """
+        from wrolpi.files.models import FileGroup, Directory
+
+        total_items = len(plan)
+        processed = 0
+        new_directories: Set[pathlib.Path] = set()
+
+        # Query existing directories once
+        existing_directories = {
+            pathlib.Path(d[0]) for d in session.query(Directory.path).all()
+        }
+        inserted_directories: Set[pathlib.Path] = set()
+
+        for chunk in chunks(list(plan.items()), MOVE_CHUNK_SIZE):
+            chunk_plan = {}
+            # Include files that exist OR don't exist (pre-moved)
+            # but exclude directories
+            old_files = [old for old, new in chunk if old.is_file() or not old.is_dir()]
+            old_dirs = [old for old, new in chunk if old.is_dir()]
+
+            # Get FileGroups for files in this chunk
+            file_groups = session.query(FileGroup).filter(
+                FileGroup.primary_path.in_([str(f) for f in old_files])
+            ).all()
+
+            # Build lookup
+            fg_by_path = {fg.primary_path: fg for fg in file_groups}
+
+            # Move each FileGroup's files
+            for old_file in old_files:
+                fg = fg_by_path.get(old_file)
+                if fg:
+                    new_path = plan[old_file]
+                    parent = new_path.parent
+                    parent.mkdir(parents=True, exist_ok=True)
+                    if parent not in new_directories:
+                        new_directories.add(parent)
+                        created_directories.add(parent)
+                    # Only move physical files if source exists
+                    # (files may have been pre-moved by user)
+                    if old_file.exists():
+                        _move_file_group_files(fg, new_path)
+                        # Track for rollback (new -> old)
+                        revert_plan[new_path] = old_file
+                    chunk_plan[old_file] = new_path
+
+            # Delete old directories.  `delete_directory` is not used here because it opens its own
+            # session; a second connection deadlocks against the write lock this transaction holds.
+            for old_dir in old_dirs:
+                if old_dir.is_dir():
+                    try:
+                        old_dir.rmdir()
+                    except OSError:
+                        continue  # Directory not empty yet
+                    session.query(Directory).filter_by(path=str(old_dir)).delete(synchronize_session=False)
+
+            # Insert new Directory records
+            missing_dirs = new_directories - existing_directories - inserted_directories
+            if missing_dirs:
+                session.add_all([
+                    Directory(path=str(d), name=d.name) for d in missing_dirs
+                ])
+                inserted_directories.update(missing_dirs)
+
+            # Bulk update FileGroups
+            _bulk_update_file_groups_db(session, chunk_plan)
+            session.flush()
+
+            # Update progress
+            processed += len(chunk)
+            percent = int((processed / total_items) * 100) if total_items > 0 else 100
+            self.update_status(
+                operation_processed=processed,
+                operation_percent=percent,
+            )
+            await asyncio.sleep(0)  # Yield for cancellation
+
+        return new_directories
+
+    def _cleanup_old_directories(
+            self,
+            sources: List[pathlib.Path],
+            old_directories: List[pathlib.Path],
+    ) -> None:
+        """Clean up old directories after a move operation.
+
+        Deletes empty subdirectories within the source tree, but preserves
+        the root source directories themselves (even if empty).
+
+        Optimized: uses single stat() calls and caches directory status to reduce
+        redundant filesystem operations on slow RPi storage.
+        """
+        from wrolpi.files.lib import delete_directory
+
+        # Identify directory sources that were moved (not file parents)
+        # Only these directories should have their subdirectories cleaned up
+        # Optimized: single stat() call per source instead of up to 2 (is_dir + potential exists)
+        moved_directories = set()
+        for source in sources:
+            try:
+                st = source.stat()
+                if stat_module.S_ISDIR(st.st_mode):
+                    moved_directories.add(source)
+            except FileNotFoundError:
+                pass  # Source doesn't exist anymore
+
+        # Collect subdirectories within moved directories for cleanup
+        # These are nested subdirs that may be empty after files were moved out
+        dirs_to_check = set()
+        for root in moved_directories:
+            try:
+                # os.walk efficiently yields directories without redundant stat calls
+                for dirpath, dirnames, _ in os.walk(root):
+                    for dirname in dirnames:
+                        dirs_to_check.add(pathlib.Path(dirpath) / dirname)
+            except OSError:
+                pass  # Root was deleted during move
+
+        # Sort deepest first to delete children before parents
+        dirs_to_check = sorted(dirs_to_check, key=lambda p: len(p.parts), reverse=True)
+
+        # Batch collect directories that no longer exist for DB cleanup
+        deleted_dirs_for_db = []
+
+        for directory in dirs_to_check:
+            if directory in moved_directories:
+                continue  # Never delete the moved directory roots themselves
+            try:
+                # Try to delete if empty - rmdir fails on non-empty dirs
+                # delete_directory handles the DB cleanup too
+                delete_directory(directory)
+            except FileNotFoundError:
+                # Directory was already deleted, collect for batch DB cleanup
+                deleted_dirs_for_db.append(str(directory))
+            except OSError:
+                pass  # Directory not empty, leave it
+
+        # Batch cleanup DB records for directories that were already deleted
+        if deleted_dirs_for_db:
+            with get_db_curs(commit=True) as curs:
+                curs.execute('DELETE FROM directory WHERE path IN (SELECT value FROM json_each(:paths))',
+                             {'paths': json.dumps(deleted_dirs_for_db)})
+
+    def _revert_move(
+            self,
+            revert_plan: Dict[pathlib.Path, pathlib.Path],
+            created_directories: Set[pathlib.Path],
+            destination: pathlib.Path,
+            destination_existed: bool,
+    ) -> None:
+        """Revert a failed move operation by moving files back to original locations."""
+        # Rollback: move files back to original locations
+        # Use shutil.move directly (not _move_file_group_files) to avoid any
+        # issues with the function that may have caused the original failure
+        if revert_plan:
+            for new_path, old_path in revert_plan.items():
+                if new_path.exists():
+                    try:
+                        old_path.parent.mkdir(parents=True, exist_ok=True)
+                        shutil.move(str(new_path), str(old_path))
+                        logger.debug(f'Reverted: {new_path} -> {old_path}')
+                    except Exception as revert_error:
+                        logger.error(f'Failed to revert {new_path}: {revert_error}')
+
+        # Clean up created directories (deepest first)
+        dirs_to_clean = sorted(created_directories, key=lambda p: len(p.parents), reverse=True)
+        for directory in dirs_to_clean:
+            if directory.is_dir():
+                try:
+                    # Only delete if empty
+                    if not any(directory.iterdir()):
+                        delete_directory(directory)
+                except (OSError, StopIteration):
+                    pass
+
+        # Clean up destination if we created it
+        if not destination_existed and destination.is_dir():
+            try:
+                # Walk destination and delete empty dirs (deepest first)
+                for path in sorted(walk(destination), key=lambda p: len(p.parents), reverse=True):
+                    if path.is_dir() and not any(path.iterdir()):
+                        delete_directory(path)
+                # Finally delete destination if empty
+                if not any(destination.iterdir()):
+                    delete_directory(destination)
+            except (OSError, StopIteration):
+                pass
+
+    async def handle_move(self, task: FileTask):
+        """Handle move task with progress tracking.
+
+        Moves files/directories to the destination using bulk operations.
+        Updates status throughout for frontend progress display.
+        """
+        from wrolpi.events import Events
+
+        destination = task.destination
+        sources = task.paths
+
+        if not destination or not sources:
+            logger.error('Invalid move task: missing destination or sources')
+            self._fail_job(task.job_id, 'Invalid move task: missing destination or sources')
+            self.reset_status()
+            return
+
+        media_directory = get_media_directory()
+
+        if not self._validate_move_paths(sources, destination, media_directory):
+            self._fail_job(task.job_id, 'Move paths are not within the media directory')
+            return
+
+        # Estimate total for planning phase based on number of sources
+        planning_total = len(sources) * 2  # Approximation for progress tracking
+
+        self.update_status(
+            status='planning',
+            task_type='move',
+            paths=[str(p) for p in sources],
+            destination=str(destination),
+            operation_total=planning_total,
+            operation_processed=0,
+            operation_percent=0,
+            error=None,
+        )
+
+        # Track if destination existed before we create it (for cleanup on failure)
+        destination_existed = destination.is_dir()
+        # Track moved files for rollback (new_path -> old_path)
+        revert_plan: Dict[pathlib.Path, pathlib.Path] = {}
+        # Track directories created during move (for cleanup on failure)
+        created_directories: Set[pathlib.Path] = set()
+        plan = {}
+
+        # Progress callback for planning phase
+        def on_planning_progress(processed: int, total: int):
+            percent = int((processed / total) * 100) if total > 0 else 0
+            self.update_status(operation_processed=processed, operation_total=total, operation_percent=percent)
+
+        try:
+            destination.mkdir(parents=True, exist_ok=True)
+
+            # Build the move plan using bulk SQL operations
+            plan, old_directories = await build_move_plan_bulk(sources, destination, on_planning_progress)
+
+            # Update status with total
+            self.update_status(
+                status='moving',
+                operation_total=len(plan),
+            )
+
+            # Execute the plan in chunks.  This reads FileGroups/Directories then updates them, so
+            # it must begin as a writer; a deferred transaction's lock upgrade fails instantly when
+            # anything else is writing (see `build_move_plan_bulk`).
+            with flags.file_worker_discovery:
+                with get_db_session(commit=True) as session:
+                    await self._execute_move_chunks(
+                        plan, session, created_directories, revert_plan
+                    )
+
+            self._cleanup_old_directories(sources, old_directories)
+            await self._apply_post_processing()
+
+            logger.info(f'Move completed: {len(plan)} items moved to {destination}')
+            Events.send_file_move_completed(f'Moved {len(sources)} items to {destination}')
+            self._complete_job(task.job_id)
+
+        except asyncio.CancelledError:
+            logger.warning('Move task was cancelled')
+            try:
+                if revert_plan:
+                    self._revert_move(revert_plan, created_directories, destination, destination_existed)
+            except Exception as revert_error:
+                logger.error(f'Failed to revert cancelled move: {revert_error}')
+            self._fail_job(task.job_id, 'Move task was cancelled')
+            raise
+        except Exception as e:
+            logger.error(f'Move failed: {e}, reverting {len(revert_plan)} items', exc_info=e)
+            self.update_status(status='reverting', error=str(e))
+            self._revert_move(revert_plan, created_directories, destination, destination_existed)
+            self._fail_job(task.job_id, e)
+            Events.send_file_move_failed(f'Move to {destination} failed: {e}')
+        finally:
+            self.reset_status()

wrolpi/files/reorganize_job.py

@@ -0,0 +1,537 @@
+"""FileWorker reorganize jobs: per-collection and batch.
+
+Extracted from worker.py.  Collection-level name-format logic lives in
+wrolpi.collections.reorganize; this module applies the resulting move mappings.
+"""
+from __future__ import annotations
+
+import asyncio
+import pathlib
+import shutil
+from pathlib import Path
+
+from wrolpi import flags
+from wrolpi.common import logger
+from wrolpi.db import get_db_session
+from wrolpi.events import Events
+from wrolpi.files.lib import (
+    split_path_stem_and_suffix, _bulk_update_file_groups_reorganize, delete_directory,
+    MOVE_CHUNK_SIZE,
+)
+
+logger = logger.getChild(__name__)
+
+PROGRESS_UPDATE_INTERVAL = 100
+
+
+class FileReorganizeMixin:
+    async def handle_reorganize(self, task: FileTask):
+        """Handle reorganize task for collection file reorganization.
+
+        Unlike handle_move which moves all files to a single destination,
+        handle_reorganize handles arbitrary source->destination mappings
+        where each file (or FileGroup) can move to a different location.
+
+        This is used when a user changes their file_name_format configuration
+        and wants to reorganize existing files to match the new format.
+
+        Optimized for bulk operations:
+        - Single query to pre-fetch all FileGroups
+        - Uses FileGroup.files instead of filesystem globs
+        - Dict lookup for O(n) path matching
+        - Batched database updates
+        """
+        from wrolpi.events import Events
+        from wrolpi.files.models import FileGroup
+
+        move_mappings = task.move_mappings
+
+        if not move_mappings:
+            logger.info('Reorganize task has no move mappings, nothing to do')
+            self._complete_job(task.job_id)
+            self.reset_status()
+            return
+
+        total_moves = len(move_mappings)
+        logger.info(f'Starting reorganize task with {total_moves} file moves')
+
+        # Transition job from 'pending' to 'running' so progress can be tracked
+        self._set_job_status(task.job_id, 'running')
+
+        self.update_status(
+            status='reorganizing',
+            task_type='reorganize',
+            paths=[],
+            destination=None,
+            operation_total=total_moves,
+            operation_processed=0,
+            operation_percent=0,
+            error=None,
+            collection_id=task.collection_id,
+            collection_kind=task.collection_kind,
+        )
+
+        # Track successful moves for rollback
+        completed_moves: list[tuple[pathlib.Path, pathlib.Path]] = []
+        # Track created directories for cleanup
+        created_directories: set[pathlib.Path] = set()
+        # Track source directories to clean up after moves
+        source_directories: set[pathlib.Path] = set()
+
+        try:
+            with flags.file_worker_discovery:
+                # Phase 1: Bulk pre-fetch all FileGroups
+                source_paths = [str(s) for s, d in move_mappings]
+                with get_db_session() as session:
+                    file_groups = session.query(FileGroup).filter(
+                        FileGroup.primary_path.in_(source_paths)
+                    ).all()
+                    # Build lookup dict and detach from session
+                    # Note: fg.primary_path is a pathlib.Path (from MediaPathType), but
+                    # source_paths are strings, so we use str() for the dictionary keys
+                    fg_by_path = {}
+                    for fg in file_groups:
+                        fg_by_path[str(fg.primary_path)] = {
+                            'id': fg.id,
+                            'directory': pathlib.Path(fg.directory),
+                            'files': list(fg.files),  # Copy the list
+                            'data': dict(fg.data) if fg.data else None,
+                        }
+
+                processed = 0
+                pending_updates = []
+
+                for source_path, dest_path in move_mappings:
+                    source_path = pathlib.Path(source_path)
+                    dest_path = pathlib.Path(dest_path)
+
+                    source_exists = source_path.exists()
+                    dest_exists = dest_path.exists()
+
+                    # Look up the pre-fetched FileGroup
+                    fg_info = fg_by_path.get(str(source_path))
+                    if not fg_info:
+                        logger.debug(f'FileGroup not found by primary_path: {source_path}')
+                        processed += 1
+                        continue
+
+                    if source_exists and dest_exists:
+                        # Conflict: both exist - skip with warning
+                        logger.warning(f'Reorganize: both source and destination exist: {source_path} -> {dest_path}')
+                        processed += 1
+                        continue
+
+                    if not source_exists and not dest_exists:
+                        # Neither exists - skip with warning
+                        logger.warning(f'Reorganize: neither source nor destination exist: {source_path}')
+                        processed += 1
+                        continue
+
+                    if not source_exists and dest_exists:
+                        # Recovery case: file already moved, just update DB to match destination
+                        logger.info(f'Reorganize recovery: updating DB for already-moved file: {dest_path}')
+
+                        dest_stem, _ = split_path_stem_and_suffix(dest_path)
+
+                        # Find existing files at destination that match the FileGroup's file extensions
+                        moved_files = []
+                        for file_info in fg_info['files']:
+                            _, suffix = split_path_stem_and_suffix(pathlib.Path(file_info['path']))
+                            dest_file = dest_path.parent / f'{dest_stem}{suffix}'
+                            if dest_file.exists():
+                                # Map original source path to destination for DB update
+                                src_file = fg_info['directory'] / file_info['path']
+                                moved_files.append((src_file, dest_file))
+
+                        if moved_files:
+                            # Build DB update using existing destination files
+                            moved_lookup = {src: dst for src, dst in moved_files}
+                            new_files = []
+                            for file_info in fg_info['files']:
+                                old_file_path = pathlib.Path(file_info['path'])
+                                if not old_file_path.is_absolute():
+                                    old_file_path = fg_info['directory'] / old_file_path
+                                if old_file_path in moved_lookup:
+                                    file_info = dict(file_info)
+                                    file_info['path'] = moved_lookup[old_file_path].name
+                                new_files.append(file_info)
+
+                            new_data = fg_info['data']
+                            if new_data:
+                                new_data = dict(new_data)
+                                filename_lookup = {src.name: dst.name for src, dst in moved_files}
+                                for key, value in new_data.items():
+                                    if isinstance(value, str) and value in filename_lookup:
+                                        new_data[key] = filename_lookup[value]
+                                    elif isinstance(value, list):
+                                        new_data[key] = [filename_lookup.get(v, v) for v in value]
+
+                            pending_updates.append({
+                                'id': fg_info['id'],
+                                'directory': str(dest_path.parent),
+                                'primary_path': str(dest_path),
+                                'files': new_files,
+                                'data': new_data,
+                            })
+                            completed_moves.append((source_path, dest_path))
+
+                        processed += 1
+                        if processed % PROGRESS_UPDATE_INTERVAL == 0:
+                            percent = int((processed / total_moves) * 100)
+                            self.update_status(operation_processed=processed, operation_percent=percent)
+                        continue
+
+                    # Normal case: source exists, dest doesn't - move files
+                    # Track source directory for cleanup
+                    source_directories.add(source_path.parent)
+
+                    # Create destination directory if needed
+                    if not dest_path.parent.exists():
+                        dest_path.parent.mkdir(parents=True, exist_ok=True)
+                        created_directories.add(dest_path.parent)
+
+                    # Phase 2: Use FileGroup.files instead of glob_shared_stem()
+                    source_stem_files = []
+                    for file_info in fg_info['files']:
+                        file_path = pathlib.Path(file_info['path'])
+                        if not file_path.is_absolute():
+                            file_path = fg_info['directory'] / file_path
+                        source_stem_files.append(file_path)
+
+                    # Move all files in the FileGroup
+                    dest_stem, _ = split_path_stem_and_suffix(dest_path)
+                    moved_files = []
+                    for src_file in source_stem_files:
+                        if not src_file.exists():
+                            continue
+                        # Compute destination for this file based on its suffix
+                        _, src_suffix = split_path_stem_and_suffix(src_file)
+                        dest_file = dest_path.parent / f'{dest_stem}{src_suffix}'
+
+                        try:
+                            shutil.move(str(src_file), str(dest_file))
+                            moved_files.append((src_file, dest_file))
+                            logger.debug(f'Reorganize: moved {src_file} -> {dest_file}')
+                        except Exception as e:
+                            logger.error(f'Failed to move {src_file} -> {dest_file}: {e}')
+                            raise
+
+                    # Prepare FileGroup update
+                    if moved_files:
+                        # Phase 3: Use dict lookup instead of nested loop
+                        moved_lookup = {src: dst for src, dst in moved_files}
+
+                        # Update files list with new paths
+                        new_files = []
+                        for file_info in fg_info['files']:
+                            old_file_path = pathlib.Path(file_info['path'])
+                            if not old_file_path.is_absolute():
+                                old_file_path = fg_info['directory'] / old_file_path
+
+                            # Find the corresponding new path using dict lookup
+                            if old_file_path in moved_lookup:
+                                file_info = dict(file_info)
+                                file_info['path'] = moved_lookup[old_file_path].name
+                            new_files.append(file_info)
+
+                        # Update data dict if it has path references
+                        # Data fields store filenames only (no '/'), so use filename-to-filename lookup
+                        new_data = fg_info['data']
+                        if new_data:
+                            new_data = dict(new_data)
+                            # Build filename-to-filename lookup (not full paths)
+                            filename_lookup = {src.name: dst.name for src, dst in moved_files}
+                            for key, value in new_data.items():
+                                if isinstance(value, str) and value in filename_lookup:
+                                    new_data[key] = filename_lookup[value]
+                                elif isinstance(value, list):
+                                    # Handle list fields like caption_paths
+                                    new_data[key] = [filename_lookup.get(v, v) for v in value]
+
+                        # Phase 4: Collect update for batch processing
+                        pending_updates.append({
+                            'id': fg_info['id'],
+                            'directory': str(dest_path.parent),
+                            'primary_path': str(dest_path),
+                            'files': new_files,
+                            'data': new_data,
+                        })
+                        completed_moves.append((source_path, dest_path))
+
+                        # Flush every MOVE_CHUNK_SIZE items
+                        if len(pending_updates) >= MOVE_CHUNK_SIZE:
+                            _bulk_update_file_groups_reorganize(pending_updates)
+                            pending_updates = []
+
+                    processed += 1
+
+                    # Update progress
+                    if processed % PROGRESS_UPDATE_INTERVAL == 0 or processed == total_moves:
+                        percent = int((processed / total_moves) * 100)
+                        logger.info(f'Reorganize progress: {processed}/{total_moves} files ({percent}%)')
+                        self.update_status(
+                            operation_processed=processed,
+                            operation_percent=percent,
+                        )
+
+                # Final flush of pending updates
+                if pending_updates:
+                    _bulk_update_file_groups_reorganize(pending_updates)
+
+                # Clean up empty source directories
+                for directory in sorted(source_directories, key=lambda p: len(p.parents), reverse=True):
+                    try:
+                        if directory.is_dir() and not any(directory.iterdir()):
+                            delete_directory(directory)
+                            logger.debug(f'Cleaned up empty directory: {directory}')
+                    except (OSError, StopIteration):
+                        pass
+
+            # Skip post-processing - reorganization only moves files, nothing to re-index
+
+            # Update file_format after successful completion (deferred update for resumability)
+            if task.collection_id and task.pending_file_format:
+                from wrolpi.collections.models import Collection
+                with get_db_session(commit=True) as session:
+                    collection = session.query(Collection).filter_by(id=task.collection_id).one_or_none()
+                    if collection:
+                        collection.file_format = task.pending_file_format
+                        logger.info(f'Updated file_format for collection {collection.name}')
+                        # Trigger config save
+                        if collection.kind == 'domain':
+                            from modules.archive.lib import save_domains_config
+                            save_domains_config.activate_switch()
+                        elif collection.kind == 'channel':
+                            from modules.videos.lib import save_channels_config
+                            save_channels_config.activate_switch()
+
+            logger.info(f'Reorganize completed: {len(completed_moves)} files moved')
+            Events.send_file_move_completed(f'Reorganized {len(completed_moves)} files')
+            self._complete_job(task.job_id)
+
+        except asyncio.CancelledError:
+            logger.warning('Reorganize task was cancelled')
+            self._fail_job(task.job_id, 'Reorganize task was cancelled')
+            raise
+        except Exception as e:
+            logger.error(f'Reorganize failed: {e}', exc_info=e)
+            self.update_status(status='error', error=str(e))
+            # Note: We don't attempt automatic rollback for reorganize tasks
+            # as the partial state may be complex. User should refresh files.
+            self._fail_job(task.job_id, e)
+            Events.send_file_move_failed(f'Reorganize failed: {e}')
+        finally:
+            self.reset_status()
+
+    async def handle_batch_reorganize(self, task: FileTask):
+        """Handle batch reorganization of multiple collections.
+
+        Processes collections sequentially:
+        1. For each collection, execute reorganization
+        2. Track overall progress and per-collection progress
+        3. Stop on first failure, report which collection failed
+
+        Progress tracking:
+        - overall_percent = (completed_collections + current_collection_%) / total_collections
+        - batch_status dict tracks completed list, current collection, and failure info
+        """
+        from wrolpi.collections.reorganize import execute_reorganization
+        from wrolpi.collections.models import Collection
+        from wrolpi.events import Events
+
+        collection_ids = task.collection_ids or []
+        kind = task.batch_kind
+
+        if not collection_ids:
+            logger.info('Batch reorganize task has no collections, nothing to do')
+            self._complete_job(task.job_id)
+            self.reset_status()
+            return
+
+        total_collections = len(collection_ids)
+        logger.info(f'Starting batch reorganization of {total_collections} {kind} collections')
+
+        # Transition job from 'pending' to 'running'
+        self._set_job_status(task.job_id, 'running')
+
+        # Initialize batch status tracking
+        batch_status = {
+            'batch_job_id': task.job_id,  # Include job ID so frontend can resume polling
+            'total_collections': total_collections,
+            'completed': [],
+            'current_collection': None,
+            'failed_collection': None,
+            'error': None,
+            'overall_percent': 0,
+        }
+
+        self.update_status(
+            status='batch_reorganizing',
+            task_type='batch_reorganize',
+            paths=[],
+            destination=None,
+            operation_total=total_collections,
+            operation_processed=0,
+            operation_percent=0,
+            error=None,
+            batch_status=batch_status,
+            collection_kind=kind,
+        )
+
+        try:
+            for idx, collection_id in enumerate(collection_ids):
+                # Fetch collection info
+                with get_db_session() as session:
+                    collection = session.query(Collection).filter_by(id=collection_id).one_or_none()
+                    if not collection:
+                        logger.warning(f'Batch reorganize: collection {collection_id} not found, skipping')
+                        continue
+
+                    collection_name = collection.name
+
+                    # Update current collection in status
+                    batch_status['current_collection'] = {
+                        'id': collection_id,
+                        'name': collection_name,
+                        'status': 'running',
+                        'total': 0,
+                        'completed': 0,
+                        'percent': 0,
+                    }
+                    self.update_status(batch_status=batch_status)
+
+                    logger.info(f'Batch reorganize: processing {collection_name} ({idx + 1}/{total_collections})')
+
+                    try:
+                        # Execute reorganization for this collection
+                        # This queues a sub-job that we need to wait for
+                        job_id = execute_reorganization(collection_id, session)
+
+                        if job_id:
+                            # Wait for the reorganization job to complete
+                            # Poll for completion while updating batch status
+                            await self._wait_for_sub_job_with_progress(
+                                job_id, batch_status, collection_id, collection_name, idx, total_collections
+                            )
+
+                        # Mark this collection as completed
+                        batch_status['completed'].append({
+                            'id': collection_id,
+                            'name': collection_name,
+                            'status': 'complete',
+                        })
+
+                        # Update overall progress
+                        completed_count = len(batch_status['completed'])
+                        batch_status['overall_percent'] = int((completed_count / total_collections) * 100)
+                        batch_status['current_collection'] = None
+                        self.update_status(
+                            operation_processed=completed_count,
+                            operation_percent=batch_status['overall_percent'],
+                            batch_status=batch_status,
+                        )
+
+                        logger.info(f'Batch reorganize: completed {collection_name}')
+
+                    except Exception as e:
+                        # Collection failed - stop batch processing
+                        logger.error(f'Batch reorganize: {collection_name} failed: {e}')
+
+                        # Get the actual channel ID for the UI link (collection_id != channel_id)
+                        channel_id = None
+                        if kind == 'channel':
+                            from modules.videos.models import Channel
+                            channel = session.query(Channel).filter_by(collection_id=collection_id).one_or_none()
+                            channel_id = channel.id if channel else None
+
+                        batch_status['failed_collection'] = {
+                            'id': collection_id,
+                            'name': collection_name,
+                            'channel_id': channel_id,
+                        }
+                        batch_status['error'] = str(e)
+                        batch_status['current_collection'] = None
+                        self.update_status(
+                            status='error',
+                            error=str(e),
+                            batch_status=batch_status,
+                        )
+                        self._fail_job(task.job_id, e)
+                        Events.send_file_move_failed(
+                            f'Batch reorganization failed on {collection_name}: {e}'
+                        )
+                        return
+
+            # All collections completed successfully
+            logger.info(f'Batch reorganization completed: {len(batch_status["completed"])} collections processed')
+            Events.send_file_move_completed(
+                f'Batch reorganization completed: {len(batch_status["completed"])} {kind}s reorganized'
+            )
+            self._complete_job(task.job_id)
+
+        except asyncio.CancelledError:
+            logger.warning('Batch reorganize task was cancelled')
+            self._fail_job(task.job_id, 'Batch reorganize task was cancelled')
+            raise
+        except Exception as e:
+            logger.error(f'Batch reorganize failed: {e}', exc_info=e)
+            batch_status['error'] = str(e)
+            self.update_status(status='error', error=str(e), batch_status=batch_status)
+            self._fail_job(task.job_id, e)
+            Events.send_file_move_failed(f'Batch reorganization failed: {e}')
+        finally:
+            self.reset_status()
+
+    async def _wait_for_sub_job_with_progress(
+            self,
+            job_id: str,
+            batch_status: dict,
+            collection_id: int,
+            collection_name: str,
+            collection_idx: int,
+            total_collections: int,
+    ):
+        """Wait for a sub-job to complete while updating batch progress.
+
+        This is used by handle_batch_reorganize to wait for each collection's
+        reorganization job while keeping the batch status updated.
+        """
+        import time
+        start = time.time()
+        timeout = 3600  # 1 hour max per collection
+
+        while time.time() - start < timeout:
+            status = self.get_job_status(job_id)
+            if status == 'complete':
+                return
+            if status == 'failed':
+                error = self.get_job_error(job_id) or 'unknown error'
+                from wrolpi.files.worker import FileWorkerJobFailed
+                raise FileWorkerJobFailed(
+                    f'Collection {collection_name} reorganization failed: {error}'
+                )
+
+            # Nested under process_queue (batch reorganize); must drain the
+            # sub-job ourselves or this wait deadlocks.
+            self.transfer_queue()
+            await self.process_queue()
+
+            # Update current collection progress based on worker status (now fresh)
+            worker_status = self.status
+            current = batch_status.get('current_collection', {})
+            if current:
+                current['total'] = worker_status.get('operation_total', 0)
+                current['completed'] = worker_status.get('operation_processed', 0)
+                current['percent'] = worker_status.get('operation_percent', 0)
+                batch_status['current_collection'] = current
+
+                # Calculate overall percent including current progress
+                completed_count = len(batch_status.get('completed', []))
+                current_progress = current['percent'] / 100
+                batch_status['overall_percent'] = int(
+                    ((completed_count + current_progress) / total_collections) * 100
+                )
+                self.update_status(batch_status=batch_status)
+
+            await asyncio.sleep(0.05)
+

wrolpi/files/test/test_worker_move.py

@@ -380,7 +380,7 @@ async def test_file_worker_move_rollback_on_failure(
             raise IOError("Simulated disk failure")
         return original_move(fg, new_path)
 
-    monkeypatch.setattr('wrolpi.files.worker._move_file_group_files', failing_move)
+    monkeypatch.setattr('wrolpi.files.move._move_file_group_files', failing_move)
 
     task = FileTask(FileTaskType.move, [test_directory / 'source'], destination=dest)
     file_worker.private_queue.put_nowait(task)
@@ -417,7 +417,7 @@ async def test_handle_move_marks_job_failed_on_exception(
     def failing_move(fg, new_path):
         raise IOError('Simulated disk failure')
 
-    monkeypatch.setattr('wrolpi.files.worker._move_file_group_files', failing_move)
+    monkeypatch.setattr('wrolpi.files.move._move_file_group_files', failing_move)
 
     job_id = 'move-will-fail'
     task = FileTask(FileTaskType.move, [source_file], destination=dest, job_id=job_id)
@@ -518,7 +518,7 @@ async def test_file_worker_move_preserves_existing_destination_on_failure(
     def failing_move(fg, new_path):
         raise IOError("Simulated failure")
 
-    monkeypatch.setattr('wrolpi.files.worker._move_file_group_files', failing_move)
+    monkeypatch.setattr('wrolpi.files.move._move_file_group_files', failing_move)
 
     task = FileTask(FileTaskType.move, [source_file], destination=dest)
     file_worker.private_queue.put_nowait(task)
@@ -547,7 +547,7 @@ async def test_file_worker_move_removes_created_destination_on_failure(
     def failing_move(fg, new_path):
         raise IOError("Simulated failure")
 
-    monkeypatch.setattr('wrolpi.files.worker._move_file_group_files', failing_move)
+    monkeypatch.setattr('wrolpi.files.move._move_file_group_files', failing_move)
 
     task = FileTask(FileTaskType.move, [source_file], destination=dest)
     file_worker.private_queue.put_nowait(task)
@@ -656,7 +656,8 @@ async def test_file_worker_move_survives_concurrent_writer(
     with contextlib.ExitStack() as stack:
         maker = stack.enter_context(production_like_sessions(test_session))
 
-        real_build_move_plan_bulk = worker_module.build_move_plan_bulk
+        from wrolpi.files import move as move_module
+        real_build_move_plan_bulk = move_module.build_move_plan_bulk
 
         async def build_plan_then_hold_write_lock(*args, **kwargs):
             plan = await real_build_move_plan_bulk(*args, **kwargs)
@@ -664,7 +665,7 @@ async def test_file_worker_move_survives_concurrent_writer(
             stack.enter_context(write_lock_held_briefly(db_file))
             return plan
 
-        monkeypatch.setattr(worker_module, 'build_move_plan_bulk', build_plan_then_hold_write_lock)
+        monkeypatch.setattr(move_module, 'build_move_plan_bulk', build_plan_then_hold_write_lock)
 
         await file_worker.process_queue()