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 ;
+ }
+
+ let label = `${phaseLabel}`;
+ if (operation_total > 0) {
+ label = `${label} (${operation_processed.toLocaleString()} / ${operation_total.toLocaleString()})`;
+ }
+
+ return ;
+}
+
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 ;
+ // 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(