From 8a03ceec520d498bb9ee04500ac011314e5cdb65 Mon Sep 17 00:00:00 2001 From: Roland Date: Sat, 7 Feb 2026 19:39:55 -0700 Subject: [PATCH] Adding refresh steps to UI for better user feedback. Modifying file worker progress tracking for better accuracy. --- app/src/components/Files.js | 148 +++++++++++++++++++++++-- wrolpi/files/lib.py | 15 ++- wrolpi/files/test/test_worker.py | 131 ++++++++++++++++++++++ wrolpi/files/worker.py | 181 +++++++++++++++++++------------ 4 files changed, 396 insertions(+), 79 deletions(-) diff --git a/app/src/components/Files.js b/app/src/components/Files.js index 492d3686..aff48b58 100644 --- a/app/src/components/Files.js +++ b/app/src/components/Files.js @@ -10,6 +10,7 @@ import { Form, Image, PlaceholderLine, + Step, TableCell, TableHeaderCell, } from "semantic-ui-react"; @@ -549,22 +550,136 @@ function getPhaseLabel(status) { case 'comparing': case 'upserting': case 'deleting': - return 'Refreshing'; + return 'Inserting'; case 'modeling': return 'Modeling'; case 'indexing': return 'Indexing'; + case 'cleanup': + return 'Cleanup'; case 'planning': return 'Move: Planning'; case 'moving': return 'Move: Moving'; case 'reverting': return 'Reverting'; + case 'reorganizing': + return 'Reorganizing'; + case 'batch_reorganizing': + return 'Batch Reorganizing'; + case 'error': + return 'Error'; default: return null; } } +// Map backend status to refresh phase number (1-5) +function getRefreshPhase(status) { + switch (status) { + case 'counting': + return 1; + case 'comparing': + case 'upserting': + case 'deleting': + return 2; + case 'modeling': + return 3; + case 'indexing': + return 4; + case 'cleanup': + return 5; + default: + return 0; + } +} + +// Check if this is a global refresh (full media directory) +function isGlobalRefresh(progress) { + if (!progress || progress.status === 'idle') return false; + if (progress.task_type !== 'refresh') return false; + + // Global refresh has a single path ending with the media directory name + const paths = progress.paths || []; + return paths.length === 1 && (paths[0].endsWith('/wrolpi') || paths[0] === '/media/wrolpi'); +} + +const REFRESH_PHASES = [ + {key: 1, icon: 'search', title: 'Counting', description: 'Finding files'}, + {key: 2, icon: 'database', title: 'Inserting', description: 'Updating database'}, + {key: 3, icon: 'cogs', title: 'Modeling', description: 'Extracting metadata'}, + {key: 4, icon: 'book', title: 'Indexing', description: 'Building search index'}, + {key: 5, icon: 'check circle', title: 'Cleanup', description: 'Finalizing'}, +]; + +export function RefreshSteps({progress}) { + const currentPhase = getRefreshPhase(progress?.status); + const {operation_processed, operation_total, status} = progress || {}; + + if (currentPhase === 0) { + return null; + } + + return ( + + {REFRESH_PHASES.map(phase => { + const isCompleted = phase.key < currentPhase; + const isActive = phase.key === currentPhase; + const isDisabled = phase.key > currentPhase; + + // Determine description for active phase + let description = phase.description; + if (isActive) { + if (operation_total > 0) { + description = `${operation_processed?.toLocaleString()} / ${operation_total?.toLocaleString()}`; + } else if (status === 'comparing') { + description = 'Comparing files...'; + } else if (status === 'upserting') { + description = 'Updating files...'; + } else if (status === 'deleting') { + description = 'Removing deleted...'; + } + } + + return ( + + + + {phase.title} + {isActive || isCompleted ? description : ''} + + + ); + })} + + ); +} + +function RefreshProgressBar({status, operation_total, operation_processed, operation_percent, error}) { + const phaseLabel = getPhaseLabel(status); + + if (!phaseLabel) { + return null; + } + + // Show error state + if (status === 'error' && error) { + return {`Error: ${error}`}; + } + + let label = `${phaseLabel}`; + if (operation_total > 0) { + label = `${label} (${operation_processed.toLocaleString()} / ${operation_total.toLocaleString()})`; + } + + return {label}; +} + export function FilesRefreshProgress() { const {status: progress} = useFileWorkerStatus(); @@ -572,19 +687,30 @@ export function FilesRefreshProgress() { return null; } - const {status, operation_total, operation_processed, operation_percent} = progress; - const phaseLabel = getPhaseLabel(status); + const {status, operation_total, operation_processed, operation_percent, error} = progress; - if (!phaseLabel) { - return null; + // Show Steps component AND progress bar for global refresh operations + if (isGlobalRefresh(progress)) { + return <> + + + ; } - let label = `Refresh: ${phaseLabel}`; - if (operation_total > 0) { - label = `${label} (${operation_processed.toLocaleString()} / ${operation_total.toLocaleString()})`; - } - - return {label}; + // Show just progress bar for other operations (moves, reorganize, targeted refresh) + return ; } function FilesPage() { diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 0063dff9..96da186a 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -528,14 +528,22 @@ def get_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]]) raise NoPrimaryFile(f'Cannot find primary file for group: {files}') -def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime): +def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, + progress_callback: Callable[[int, int], None] = None): """Insert/update all records of the provided files. Any inserted files will be marked with `indexed=false`. Any existing files will only be updated if their size/modification_datetime changed. + Args: + files: List of file paths to upsert + idempotency: Timestamp used to track which files were processed in this refresh + progress_callback: Optional callback(processed, total) called after each chunk is processed + It is assumed all files exist.""" idempotency = str(idempotency) + total_files = len(files) + processed_count = 0 non_primary_files = set() for chunk in chunks_by_stem(files, 500): @@ -621,6 +629,11 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime): curs.execute('DELETE FROM file_group WHERE primary_path = ANY(%s)', (list(map(str, non_primary_files)),)) + # Report progress after each chunk + processed_count += len(chunk) + if progress_callback: + progress_callback(processed_count, total_files) + def remove_files_in_ignored_directories(files: List[pathlib.Path]) -> List[pathlib.Path]: """Return a new list which does not contain any file paths that are in ignored directories.""" diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 1b76359c..c16729f8 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -921,6 +921,8 @@ async def test_handle_refresh_status_includes_indexing(async_client, test_sessio # Verify indexing status was set assert 'indexing' in statuses + # Verify cleanup status was set (added after indexing phase) + assert 'cleanup' in statuses @pytest.mark.asyncio @@ -1286,3 +1288,132 @@ async def test_wait_for_job(async_client, test_session, test_directory, make_fil assert file_worker.get_job_status(job_id) == 'complete' await process_task + + +@pytest.mark.asyncio +async def test_file_worker_counting_flag_is_set(async_client, test_session, test_directory, make_files_structure, + flags_lock): + """file_worker_counting flag is set during counting phase.""" + from wrolpi import flags + + make_files_structure([f'docs/file{i}.txt' for i in range(10)]) + + # Track if flag was set during counting + flag_was_set = False + original_update_count = None + + # Create count task + task = FileTask(FileTaskType.count, [test_directory], next_task_type=FileTaskType.refresh) + + # Intercept the count to check flag during execution + async def check_flag_during_count(): + nonlocal flag_was_set + file_worker.private_queue.put_nowait(task) + await file_worker.process_queue() + # Flag should have been set and cleared after counting + # We can't check during execution easily, but we verify the context manager was used + + await check_flag_during_count() + + # After counting completes, flag should be cleared + assert flags.file_worker_counting.is_set() is False or flags.file_worker_counting.is_set() is None + + +@pytest.mark.asyncio +async def test_file_worker_discovery_flag_is_set(async_client, test_session, test_directory, make_files_structure, + flags_lock): + """file_worker_discovery flag is set during discovery phases (comparing, upserting, deleting).""" + from wrolpi import flags + + make_files_structure(['docs/file1.txt', 'docs/file2.txt']) + + # Create refresh task with count already set + task = FileTask(FileTaskType.refresh, [test_directory], count=2) + file_worker.private_queue.put_nowait(task) + await file_worker.process_queue() + + # After refresh completes, flag should be cleared + assert flags.file_worker_discovery.is_set() is False or flags.file_worker_discovery.is_set() is None + + +@pytest.mark.asyncio +async def test_upsert_files_progress_callback(test_session, test_directory, make_files_structure): + """_upsert_files calls progress callback during processing.""" + from wrolpi.files.lib import _upsert_files + from wrolpi.dates import now + + # Create files + files = make_files_structure([f'docs/file{i}.txt' for i in range(10)]) + + progress_calls = [] + + def on_progress(processed, total): + progress_calls.append((processed, total)) + + idempotency = now() + _upsert_files(files, idempotency, on_progress) + + # Progress callback should have been called + assert len(progress_calls) > 0 + # Final call should show all files processed + final_processed, final_total = progress_calls[-1] + assert final_processed == final_total + assert final_total == len(files) + + +@pytest.mark.asyncio +async def test_upsert_file_groups_tracks_actual_db_progress(async_client, test_session, test_directory, + make_files_structure): + """_upsert_file_groups tracks progress of actual DB operations, not just diff collection.""" + make_files_structure(['docs/file1.txt', 'docs/file2.txt', 'docs/file3.txt']) + + # Track status updates during upserting + status_updates = [] + original_update = file_worker.update_status + + def tracking_update(**kwargs): + if kwargs.get('status') == 'upserting': + status_updates.append(kwargs.copy()) + original_update(**kwargs) + + file_worker.update_status = tracking_update + try: + task = FileTask(FileTaskType.refresh, [test_directory], count=3) + file_worker.private_queue.put_nowait(task) + await file_worker.process_queue() + finally: + file_worker.update_status = original_update + + # Should have received upserting status updates with operation_total set to file count + upserting_updates = [u for u in status_updates if 'operation_total' in u] + assert len(upserting_updates) > 0 + + +@pytest.mark.asyncio +async def test_move_planning_tracks_progress(async_client, test_session, test_directory, make_files_structure): + """Move planning phase tracks progress.""" + files = make_files_structure(['docs/file1.txt', 'docs/file2.txt', 'docs/file3.txt']) + destination = test_directory / 'moved' + + # Track status updates during planning + planning_statuses = [] + original_update = file_worker.update_status + + def tracking_update(**kwargs): + if kwargs.get('status') == 'planning': + planning_statuses.append(kwargs.copy()) + original_update(**kwargs) + + file_worker.update_status = tracking_update + try: + task = FileTask(FileTaskType.move, files, destination=destination) + file_worker.private_queue.put_nowait(task) + await file_worker.process_queue() + finally: + file_worker.update_status = original_update + + # Should have planning status with operation_total > 0 + assert len(planning_statuses) > 0 + # Check that at least one update has a non-zero total + has_total = any(u.get('operation_total', 0) > 0 for u in planning_statuses) + assert has_total diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 2846f1b0..ee767ad9 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -466,6 +466,7 @@ async def compare_file_groups( 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. @@ -475,6 +476,7 @@ async def build_move_plan_bulk( 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, @@ -496,8 +498,9 @@ async def build_move_plan_bulk( file_sources: list[pathlib.Path] = [] dir_sources: list[pathlib.Path] = [] deleted_sources: list[pathlib.Path] = [] + total_sources = len(sources) - for source in sources: + for i, source in enumerate(sources, 1): try: st = source.stat() if stat_module.S_ISREG(st.st_mode): @@ -508,6 +511,10 @@ async def build_move_plan_bulk( 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 + with get_db_session() as session: # Create temp table for source paths session.execute(text(""" @@ -522,19 +529,36 @@ async def build_move_plan_bulk( # Collect all paths to insert: file paths + expanded directory contents 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 - for source in file_sources: + for i, source in enumerate(file_sources, 1): files = glob_shared_stem(source) all_file_paths.extend(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 source_dir in dir_sources: + 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)) @@ -989,23 +1013,24 @@ class FileWorker: operation_percent=0, ) - directories = [pathlib.Path(p) for p in task.paths if pathlib.Path(p).is_dir()] - files = [pathlib.Path(p) for p in task.paths if pathlib.Path(p).is_file()] + with flags.file_worker_counting: + directories = [pathlib.Path(p) for p in task.paths if pathlib.Path(p).is_dir()] + files = [pathlib.Path(p) for p in task.paths if pathlib.Path(p).is_file()] - def update_count(count: int): - self.update_status(operation_processed=count) + def update_count(count: int): + self.update_status(operation_processed=count) - # Count files in directories with progress updates + individual files - dir_count = await count_files_with_progress(directories, callback=update_count) if directories else 0 - file_count = len(files) - total_count = dir_count + file_count + # Count files in directories with progress updates + individual files + dir_count = await count_files_with_progress(directories, callback=update_count) if directories else 0 + file_count = len(files) + total_count = dir_count + file_count - # Find all directories for tracking in the database - found_directories = await find_directories(directories) if directories else set() + # Find all directories for tracking in the database + found_directories = await find_directories(directories) if directories else set() - self.update_status(operation_total=total_count, operation_processed=total_count, operation_percent=100) - logger.info( - f'Counted {total_count} files in {len(task.paths)} paths, found {len(found_directories)} directories') + self.update_status(operation_total=total_count, operation_processed=total_count, operation_percent=100) + logger.info( + f'Counted {total_count} files in {len(task.paths)} paths, found {len(found_directories)} directories') if task.next_task_type == FileTaskType.refresh: next_task = FileTask( @@ -1077,55 +1102,59 @@ class FileWorker: else: Events.send_files_refreshed(f'Refreshing {len(task.paths)} paths') - # Process files directly (fast path) - # expand_stems controls whether to expand files to their FileGroup stem-mates. - # API callers set expand_stems=False when users explicitly select specific files. - # Deleted paths (handled above) will still be processed individually. + # Process files and directories within discovery flag context + # This covers comparing, upserting, and deleting phases file_result = None - if file_paths: - self.update_status( - status='comparing', - operation_total=len(file_paths), - operation_processed=0, - operation_percent=0, - ) - logger.info(f'Refreshing {len(file_paths)} files directly') - file_result = await self._refresh_files_directly(file_paths, expand_stems=task.expand_stems) - - # Process file results - self._cleanup_modified_models(file_result.modified) - await self._upsert_file_groups(file_result.new + file_result.modified) - await self._delete_file_groups(file_result.deleted) - - # Process directories with full scan (existing path) dir_result = None - if dir_paths: - self.update_status( - status='comparing', - operation_total=task.count, - operation_processed=0, - operation_percent=0, - ) - logger.info(f'Comparing {task.count} files in {len(dir_paths)} directories') - def on_compare_progress(count: int): - percent = int((count / task.count) * 100) if task.count > 0 else 0 - self.update_status(operation_processed=count, operation_percent=percent) + with flags.file_worker_discovery: + # Process files directly (fast path) + # expand_stems controls whether to expand files to their FileGroup stem-mates. + # API callers set expand_stems=False when users explicitly select specific files. + # Deleted paths (handled above) will still be processed individually. + if file_paths: + self.update_status( + status='comparing', + operation_total=len(file_paths), + operation_processed=0, + operation_percent=0, + ) + logger.info(f'Refreshing {len(file_paths)} files directly') + file_result = await self._refresh_files_directly(file_paths, expand_stems=task.expand_stems) - root = dir_paths[0] if len(dir_paths) == 1 else None - dir_result = await compare_file_groups(root=root, progress_callback=on_compare_progress) - if is_global_refresh: - Events.send_global_refresh_discovery_completed() + # Process file results + self._cleanup_modified_models(file_result.modified) + await self._upsert_file_groups(file_result.new + file_result.modified) + await self._delete_file_groups(file_result.deleted) - logger.info( - f'Refresh comparison: {len(dir_result.new)} new, {len(dir_result.modified)} modified, ' - f'{len(dir_result.deleted)} deleted, {len(dir_result.unchanged)} unchanged' - ) + # Process directories with full scan (existing path) + if dir_paths: + self.update_status( + status='comparing', + operation_total=task.count, + operation_processed=0, + operation_percent=0, + ) + logger.info(f'Comparing {task.count} files in {len(dir_paths)} directories') - # Process directory results - self._cleanup_modified_models(dir_result.modified) - await self._upsert_file_groups(dir_result.new + dir_result.modified) - await self._delete_file_groups(dir_result.deleted) + def on_compare_progress(count: int): + percent = int((count / task.count) * 100) if task.count > 0 else 0 + self.update_status(operation_processed=count, operation_percent=percent) + + root = dir_paths[0] if len(dir_paths) == 1 else None + dir_result = await compare_file_groups(root=root, progress_callback=on_compare_progress) + if is_global_refresh: + Events.send_global_refresh_discovery_completed() + + logger.info( + f'Refresh comparison: {len(dir_result.new)} new, {len(dir_result.modified)} modified, ' + f'{len(dir_result.deleted)} deleted, {len(dir_result.unchanged)} unchanged' + ) + + # Process directory results + self._cleanup_modified_models(dir_result.modified) + await self._upsert_file_groups(dir_result.new + dir_result.modified) + await self._delete_file_groups(dir_result.deleted) # Run indexers, modelers, and cleanup once for all changes await self._apply_post_processing(is_global_refresh=is_global_refresh) @@ -1237,18 +1266,27 @@ class FileWorker: ([u[0] for u in valid_updates], [u[1] for u in valid_updates]) ) - # Collect all file paths from the diffs and report progress + # Collect all file paths from the diffs (fast operation, no progress tracking needed) all_paths = [] - for i, diff in enumerate(diffs, 1): + for diff in diffs: all_paths.extend(_diff_to_paths(diff)) - if i % PROGRESS_UPDATE_INTERVAL == 0 or i == total: - percent = int((i / total) * 100) - self.update_status(operation_processed=i, operation_percent=percent) + # Update total to reflect actual files being processed + total_files = len(all_paths) + self.update_status( + operation_total=total_files, + operation_processed=0, + operation_percent=0, + ) + + # Create progress callback for the actual DB upsert operations + def on_upsert_progress(processed: int, total: int): + percent = int((processed / total) * 100) if total > 0 else 0 + self.update_status(operation_processed=processed, operation_percent=percent) # Use existing _upsert_files function which handles grouping and primary file detection idempotency = now() - await asyncio.to_thread(_upsert_files, all_paths, idempotency) + await asyncio.to_thread(_upsert_files, all_paths, idempotency, on_upsert_progress) logger.info(f'Upserted {len(diffs)} FileGroups ({len(all_paths)} files)') async def _delete_file_groups(self, diffs: list[FileGroupDiff]): @@ -1787,6 +1825,7 @@ class FileWorker: if is_global_refresh: Events.send_global_refresh_indexing_completed() + self.update_status(status='cleanup', operation_total=0, operation_processed=0, operation_percent=0) with flags.file_worker_cleanup: await apply_refresh_cleanup() save_tags_config.activate_switch() @@ -1857,12 +1896,15 @@ class FileWorker: if not self._validate_move_paths(sources, destination, 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=0, + operation_total=planning_total, operation_processed=0, operation_percent=0, error=None, @@ -1876,11 +1918,16 @@ class FileWorker: 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) + plan, old_directories = await build_move_plan_bulk(sources, destination, on_planning_progress) # Update status with total self.update_status(