From a8750afef03be8993aca16b38ad7157eeefa618b Mon Sep 17 00:00:00 2001 From: dtoro Date: Wed, 8 Apr 2026 00:54:11 +0200 Subject: [PATCH] feat: cleaner TopBar + live scan progress wired end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related polish items. 1. Drop dead TopBar buttons - Removed the hamburger menu (Tab already toggles the sidebar), the grid/list view-mode toggle (only Grid was ever implemented), and the FolderOpen / Upload / Settings action icons (no features behind them). - TopBar is now: logo + active heap pill | search | filter toggle. Removed the now-unused Grid/List/Menu/FolderOpen/ Upload/Settings icon imports and the dead viewMode local state. 2. Wire live scan progress - The frontend ScanProgress widget was already polling /api/v1/library/scan/status, but the worker never wrote the Redis keys that endpoint reads — it only updated celery's internal task state. So the progress UI was permanently idle. - Worker now writes scan:active / scan:current_folder / scan:processed_files / scan:total_files / scan:errors at every meaningful step. _get_redis() returns None on failure so a Redis outage degrades gracefully (scan still runs, progress just doesn't show). - Pre-walk computes total_files upfront — without it the progress bar jumped every time os.walk discovered a new subfolder because the running total was being updated as it went. - Errors are RPUSHed to a capped list (MAX_ERROR_ENTRIES=50) so a noisy scan can't blow up Redis. - finally: clause guarantees scan:active flips to false even on a crash, so the UI never sticks at "scanning" forever. - scan_all_source_roots clears scan:errors and resets counters before queuing the per-root tasks, so each top-level scan starts with a clean slate. Two latent bugs caught and fixed in passing: - watch_folders was still reading settings.source_roots which no longer exists since we moved source roots to the DB. Now it loads them from the DB via a synchronous one-shot async wrapper at task startup. - _scan_all_source_roots_async was missing entirely after the last refactor — defined inline now, reads active source roots from the DB and dispatches scan_folder per row. Co-Authored-By: Claude Opus 4.6 (1M context) --- backend/app/tasks/scan.py | 181 +++++++++++++++++----- frontend/src/components/layout/TopBar.tsx | 82 +--------- 2 files changed, 148 insertions(+), 115 deletions(-) diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 0a39c71..ce0df34 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -14,6 +14,7 @@ from celery import shared_task from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession import aiofiles +import redis from app.database import AsyncSessionLocal from app.models import Photo, Folder, SourceRoot @@ -23,6 +24,27 @@ from app.services.metadata import extract_metadata logger = logging.getLogger(__name__) + +# Redis keys read by GET /api/v1/library/scan/status. The frontend +# ScanProgress widget polls that endpoint, so anything we want to surface +# in the UI lives here. +REDIS_KEY_ACTIVE = 'scan:active' +REDIS_KEY_CURRENT_FOLDER = 'scan:current_folder' +REDIS_KEY_PROCESSED = 'scan:processed_files' +REDIS_KEY_TOTAL = 'scan:total_files' +REDIS_KEY_ERRORS = 'scan:errors' +MAX_ERROR_ENTRIES = 50 # cap the errors list so a noisy scan doesn't blow Redis + + +def _get_redis(): + """Connect to the broker for progress writes. Returns None on failure + so a Redis outage doesn't prevent the scan itself from running.""" + try: + return redis.Redis.from_url(settings.celery_broker_url) + except Exception as e: + logger.warning(f"Could not reach Redis for scan progress: {e}") + return None + # Supported file extensions PHOTO_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.tiff', '.tif', '.webp', '.bmp'} RAW_EXTENSIONS = {'.cr2', '.cr3', '.nef', '.arw', '.raf', '.dng', '.orf', '.rw2', '.pef', '.srw'} @@ -65,37 +87,74 @@ def scan_folder(self, folder_path: str, source_root_id: Optional[str] = None): return asyncio.run(_scan_folder_async(folder_path, source_root_id, self)) async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], task): - """Async implementation of folder scanning""" + """Async implementation of folder scanning. Writes progress to Redis so + GET /api/v1/library/scan/status can surface it to the frontend + ScanProgress widget.""" logger.info(f"Starting scan of folder: {folder_path}") - + + r = _get_redis() + + def progress_set(key: str, value) -> None: + if r is None: + return + try: + r.set(key, str(value)) + except Exception as e: + logger.debug(f"scan progress set failed: {e}") + + def progress_push_error(message: str) -> None: + if r is None: + return + try: + r.lpush(REDIS_KEY_ERRORS, message) + r.ltrim(REDIS_KEY_ERRORS, 0, MAX_ERROR_ENTRIES - 1) + except Exception as e: + logger.debug(f"scan progress push_error failed: {e}") + + # Mark scan active immediately so the UI starts polling fast. + progress_set(REDIS_KEY_ACTIVE, 'true') + progress_set(REDIS_KEY_CURRENT_FOLDER, folder_path) + async with AsyncSessionLocal() as session: try: # Get or create source root if not source_root_id: source_root = await get_or_create_source_root(session, folder_path) source_root_id = source_root.id - - # Walk the directory tree + + # Pre-walk to compute the total file count upfront. Without this + # the progress bar would jump every time a new subfolder is + # encountered because the running total kept growing. total_files = 0 + for _root, _dirs, files in os.walk(folder_path): + total_files += sum( + 1 for f in files if Path(f).suffix.lower() in SUPPORTED_EXTENSIONS + ) + progress_set(REDIS_KEY_TOTAL, total_files) + progress_set(REDIS_KEY_PROCESSED, 0) + processed_files = 0 errors = [] - + for root, dirs, files in os.walk(folder_path): # Get or create folder entry folder = await get_or_create_folder(session, root, source_root_id) - + progress_set(REDIS_KEY_CURRENT_FOLDER, root) + # Filter supported files supported_files = [f for f in files if Path(f).suffix.lower() in SUPPORTED_EXTENSIONS] - total_files += len(supported_files) - + # Process files in batches batch_size = settings.scanner.batch_size for i in range(0, len(supported_files), batch_size): batch = supported_files[i:i + batch_size] - + # Defer task dispatch until AFTER commit so workers don't + # query for rows that aren't visible to other sessions yet. + pending_dispatch: list[str] = [] + for filename in batch: filepath = os.path.join(root, filename) - + try: # Check if file already exists in database existing = await session.execute( @@ -104,19 +163,20 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta if existing.scalar_one_or_none(): logger.debug(f"File already indexed: {filepath}") processed_files += 1 + progress_set(REDIS_KEY_PROCESSED, processed_files) continue - + # Get file stats stat = os.stat(filepath) - + # Calculate file hash for duplicate detection file_hash = await calculate_file_hash(filepath) - + # Check for duplicate by hash duplicate = await session.execute( select(Photo).where(Photo.file_hash == file_hash) ) if file_hash else None - + # Create photo entry photo = Photo( filepath=filepath, @@ -131,55 +191,66 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta is_duplicate=bool(duplicate.scalar_one_or_none() if duplicate else False), processing_status='pending' ) - + session.add(photo) - await session.flush() # Get the photo ID - - # Queue thumbnail generation - generate_thumbnails.delay(photo.id) - - # Queue metadata extraction - extract_metadata.delay(photo.id) - + await session.flush() # Assign defaults / FK ids + + # Queue dispatch happens after the batch commit + # below; otherwise the worker can race the writer + # and see "Photo not found". + pending_dispatch.append(photo.id) + processed_files += 1 - - # Update progress + progress_set(REDIS_KEY_PROCESSED, processed_files) + + # Celery internal progress (used by celery tooling) if processed_files % 10 == 0: task.update_state( state='PROGRESS', meta={ 'current': processed_files, 'total': total_files, - 'folder': root + 'folder': root, } ) - + except Exception as e: logger.error(f"Error processing file {filepath}: {e}") errors.append({'file': filepath, 'error': str(e)}) + progress_push_error(f"{filepath}: {e}") continue - - # Commit batch + + # Commit batch, then queue worker tasks. Dispatch order + # matters: commit first so workers can find the rows. await session.commit() - + + for photo_id in pending_dispatch: + generate_thumbnails.delay(photo_id) + extract_metadata.delay(photo_id) + # Update folder scan timestamp folder.last_scanned = datetime.utcnow() folder.photo_count = processed_files await session.commit() - + logger.info(f"Scan complete. Processed {processed_files}/{total_files} files. Errors: {len(errors)}") - + return { 'status': 'completed', 'processed': processed_files, 'total': total_files, - 'errors': errors + 'errors': errors, } - + except Exception as e: logger.error(f"Scan failed: {e}") + progress_push_error(f"scan failed: {e}") await session.rollback() raise + finally: + # Always mark inactive on the way out so a crashed scan doesn't + # leave the UI thinking we're still scanning. + progress_set(REDIS_KEY_ACTIVE, 'false') def _normalize_path(path: str) -> str: """Canonicalise a filesystem path so we don't get duplicate DB rows for @@ -251,10 +322,24 @@ async def get_or_create_folder(session: AsyncSession, path: str, source_root_id: @shared_task(name='scan_all_source_roots') def scan_all_source_roots(): """Scan every active source root currently registered in the DB.""" + # Clear stale per-scan progress before queuing new work so the UI sees + # a clean slate even if a previous run crashed mid-flight. + r = _get_redis() + if r is not None: + try: + r.delete(REDIS_KEY_ERRORS) + r.set(REDIS_KEY_PROCESSED, 0) + r.set(REDIS_KEY_TOTAL, 0) + except Exception as e: + logger.debug(f"scan_all_source_roots redis reset failed: {e}") + return asyncio.run(_scan_all_source_roots_async()) async def _scan_all_source_roots_async(): + """Read every active SourceRoot from the DB and queue a scan_folder task + for each. Source roots whose path no longer exists on disk are skipped + with a warning (the cleanup service surfaces those at startup too).""" async with AsyncSessionLocal() as session: result = await session.execute( select(SourceRoot).where(SourceRoot.is_active == True) # noqa: E712 @@ -266,16 +351,34 @@ async def _scan_all_source_roots_async(): else: logger.warning(f"Source root path does not exist: {sr.path}") + @shared_task(name='watch_folders') def watch_folders(): """ - Watch folders for changes using watchfiles - This is a long-running task that monitors file system events + Watch folders for changes using watchfiles. Long-running task that + monitors filesystem events under every active source root. """ from watchfiles import watch - - paths = [sr.path for sr in settings.source_roots if os.path.exists(sr.path)] - + + # Read source roots from the DB instead of the (now-removed) YAML + # config. Synchronous lookup is fine here — this runs once at task + # start, not on every event. + paths: list[str] = [] + try: + async def _load_paths(): + async with AsyncSessionLocal() as session: + result = await session.execute( + select(SourceRoot).where(SourceRoot.is_active == True) # noqa: E712 + ) + return [ + sr.path for sr in result.scalars().all() + if os.path.exists(sr.path) + ] + paths = asyncio.run(_load_paths()) + except Exception as e: + logger.error(f"watch_folders could not load source roots: {e}") + return + if not paths: logger.warning("No valid source roots to watch") return diff --git a/frontend/src/components/layout/TopBar.tsx b/frontend/src/components/layout/TopBar.tsx index 5f44464..83ce6d8 100644 --- a/frontend/src/components/layout/TopBar.tsx +++ b/frontend/src/components/layout/TopBar.tsx @@ -1,13 +1,7 @@ import { useState, useEffect, useRef } from 'react' import { Search, - Grid, - List, SlidersHorizontal, - FolderOpen, - Upload, - Settings, - Menu, X, ShoppingBasket, } from 'lucide-react' @@ -49,8 +43,6 @@ export function TopBar() { } }, [searchQuery, storeQ, setStoreQ]) - const [viewMode, setViewMode] = useState<'grid' | 'list'>('grid') - // Currently active heap. Shown as a pill so the user always knows where // their next P-press will land. const { data: heapsList = [] } = useHeapsQuery() @@ -58,17 +50,8 @@ export function TopBar() { return (
- {/* Left Section - Menu and App Name */} + {/* Left — logo + active heap pill */}
-
Mulita

Mulita

@@ -83,8 +66,8 @@ export function TopBar() { )}
- - {/* Center Section - Search */} + + {/* Center — search */}
@@ -117,38 +100,9 @@ export function TopBar() { )}
- - {/* Right Section - View Controls and Actions */} + + {/* Right — filter toggle */}
- {/* View Mode Toggle */} -
- - -
- - {/* Filter Button */} - -
- - {/* Action Buttons */} - - - - -
) -} \ No newline at end of file +}