dtoro/mule-image
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cleaner TopBar + live scan progress wired end-to-end

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-04-08 00:54:11 +02:00
parent 320107841b
commit a8750afef0
2 changed files with 148 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
backend/app/tasks/scan.py
View File

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