From b51cd41ba355705be132a7441ebb255e03c3608d Mon Sep 17 00:00:00 2001 From: Roland Date: Wed, 23 Sep 2026 14:38:52 -0600 Subject: [PATCH] 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 | 598 +++++++++++++ wrolpi/files/reorganize_job.py | 537 ++++++++++++ wrolpi/files/test/test_worker_move.py | 13 +- wrolpi/files/worker.py | 1105 +------------------------ 4 files changed, 1155 insertions(+), 1098 deletions(-) create mode 100644 wrolpi/files/move.py create mode 100644 wrolpi/files/reorganize_job.py diff --git a/wrolpi/files/move.py b/wrolpi/files/move.py new file mode 100644 index 00000000..da104f9a --- /dev/null +++ b/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() diff --git a/wrolpi/files/reorganize_job.py b/wrolpi/files/reorganize_job.py new file mode 100644 index 00000000..7101e3c6 --- /dev/null +++ b/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) + diff --git a/wrolpi/files/test/test_worker_move.py b/wrolpi/files/test/test_worker_move.py index 2dcc7d9e..2ee4e4e8 100644 --- a/wrolpi/files/test/test_worker_move.py +++ b/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() diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 2ce4f4ec..99890a5c 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -1,46 +1,42 @@ """File worker queue: refresh, move, and reorganize jobs. -Filesystem compare/discover lives in `wrolpi.files.refresh`. This module owns the -job queue, status, and the FileWorker methods that apply compare results (upsert, -delete, model) plus move/reorganize. +Filesystem compare/discover lives in `wrolpi.files.refresh`. Move and reorganize +handlers live in `wrolpi.files.move` and `wrolpi.files.reorganize_job`. This module +owns the job queue, status, and applying compare results (upsert, delete, model). """ import asyncio import json import os import pathlib import queue -import shutil -import stat as stat_module import time -import uuid -from collections import OrderedDict, defaultdict +from collections import defaultdict from dataclasses import dataclass, field -from datetime import datetime, timezone from enum import Enum, auto from multiprocessing.queues import Queue from pathlib import Path -from typing import AsyncGenerator, Callable, Set, Dict, List, Tuple +from typing import Callable, Set, Dict, List, Tuple from sqlalchemy import text, or_ from wrolpi import flags from wrolpi.common import apply_modelers, apply_refresh_cleanup -from wrolpi.common import get_media_directory, logger, walk, chunks, unique_by_predicate +from wrolpi.common import get_media_directory, logger from wrolpi.db import get_db_session, get_db_curs -from wrolpi.errors import NoPrimaryFile from wrolpi.events import Events from wrolpi.vars import PYTEST from wrolpi.files.lib import ( - split_path_stem_and_suffix, _upsert_files, get_unique_files_by_stem, glob_shared_stem, - group_files_by_stem, get_primary_file, choose_primary_file, delete_directory, apply_indexers, - _move_file_group_files, _bulk_update_file_groups_db, MOVE_CHUNK_SIZE, - _bulk_update_file_groups_reorganize, get_normalized_ignored_directories, remove_files_in_ignored_directories, + split_path_stem_and_suffix, _upsert_files, glob_shared_stem, + group_files_by_stem, choose_primary_file, apply_indexers, + get_normalized_ignored_directories, remove_files_in_ignored_directories, ) from wrolpi.files.refresh import ( FileGroupDiff, FileComparisonResult, classify_file_group_diffs, compare_file_groups, count_files, count_files_with_progress, find_directories, GNU_FIND_PRINTF, _await_db, _diff_to_paths, _filenames_from_files_json, _stream_filesystem_entries, ) +from wrolpi.files.move import FileMoveMixin, build_move_plan_bulk +from wrolpi.files.reorganize_job import FileReorganizeMixin logger = logger.getChild(__name__) @@ -62,6 +58,7 @@ __all__ = [ 'count_files', 'file_worker', 'GNU_FIND_PRINTF', + 'build_move_plan_bulk', ] @@ -69,256 +66,6 @@ class FileWorkerJobFailed(RuntimeError): """A tracked file-worker job finished unsuccessfully.""" -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 FileTaskType(str, Enum): count = auto() # Simply count the files. refresh = auto() # Update the DB to match the files that exist on disk. @@ -356,7 +103,7 @@ class FileTask: pending_file_format: str = None -class FileWorker: +class FileWorker(FileMoveMixin, FileReorganizeMixin): def __init__(self): # Use last-in, first out so a task that inserts another task is handled next. @@ -1181,180 +928,6 @@ class FileWorker: ''', params) curs.execute('DELETE FROM doc WHERE id IN (SELECT value FROM json_each(:ids))', params) - 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)}) - async def _refresh_files_directly( self, file_paths: list[pathlib.Path], @@ -1520,658 +1093,6 @@ class FileWorker: await apply_refresh_cleanup() save_tags_config.activate_switch() - 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() - - 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' - 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) - - raise TimeoutError(f'Collection {collection_name} reorganization timed out') - # Each Sanic worker constructs a FileWorker. The perpetual signal is started # once, but request handlers (and wait_for_job in tests) can also drain the