Adding refresh steps to UI for better user feedback. Modifying file worker progress tracking for better accuracy.
Some checks are pending
CodeQL / Analyze (push) Waiting to run
Controller Tests / controller-tests (push) Waiting to run
Jest Tests / jest-tests (push) Waiting to run

This commit is contained in:
Roland 2026-02-07 19:39:55 -07:00
parent a6d6aaf9b4
commit 8a03ceec52
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
4 changed files with 396 additions and 79 deletions

View file

@ -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 (
<Step.Group size='mini' fluid>
{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 (
<Step
key={phase.key}
active={isActive}
completed={isCompleted}
disabled={isDisabled}
>
<Icon name={isCompleted ? 'check' : phase.icon}/>
<Step.Content>
<Step.Title>{phase.title}</Step.Title>
<Step.Description>{isActive || isCompleted ? description : ''}</Step.Description>
</Step.Content>
</Step>
);
})}
</Step.Group>
);
}
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 <Progress error percent={100}>{`Error: ${error}`}</Progress>;
}
let label = `${phaseLabel}`;
if (operation_total > 0) {
label = `${label} (${operation_processed.toLocaleString()} / ${operation_total.toLocaleString()})`;
}
return <Progress active color='violet' percent={operation_percent || 0} progress>{label}</Progress>;
}
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 <>
<RefreshSteps progress={progress}/>
<RefreshProgressBar
status={status}
operation_total={operation_total}
operation_processed={operation_processed}
operation_percent={operation_percent}
error={error}
/>
</>;
}
let label = `Refresh: ${phaseLabel}`;
if (operation_total > 0) {
label = `${label} (${operation_processed.toLocaleString()} / ${operation_total.toLocaleString()})`;
}
return <Progress active color='violet' percent={operation_percent || 0} progress>{label}</Progress>;
// Show just progress bar for other operations (moves, reorganize, targeted refresh)
return <RefreshProgressBar
status={status}
operation_total={operation_total}
operation_processed={operation_processed}
operation_percent={operation_percent}
error={error}
/>;
}
function FilesPage() {

View file

@ -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."""

View file

@ -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

View file

@ -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(