diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 2420904..089cfd8 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -409,6 +409,7 @@ async def _scan_all_source_roots_async(): still hit Settings → Re-detect duplicates to force a fresh pass. """ from app.tasks.thumbs import regroup_duplicates_task + from app.tasks.vision import backfill_vision, recluster_faces async with AsyncSessionLocal() as session: result = await session.execute( @@ -433,6 +434,21 @@ async def _scan_all_source_roots_async(): except Exception as e: logger.warning(f"Could not queue post-scan regroup: {e}") + # 90s lets thumbnails finish so photos reach processing_status + # 'completed', which backfill_vision uses as its filter. + try: + backfill_vision.apply_async(countdown=90) + except Exception as e: + logger.warning(f"Could not queue post-scan vision backfill: {e}") + + # 300s gives face extraction time to run before reclustering. + # Fires even if some faces are still in-flight — the task is + # idempotent and the user can re-trigger from Settings. + try: + recluster_faces.apply_async(countdown=300) + except Exception as e: + logger.warning(f"Could not queue post-scan face recluster: {e}") + @shared_task(name='watch_folders') def watch_folders(): diff --git a/backend/app/tasks/vision.py b/backend/app/tasks/vision.py index 384091a..c44a1de 100644 --- a/backend/app/tasks/vision.py +++ b/backend/app/tasks/vision.py @@ -348,13 +348,40 @@ def _save_faces(photo_id: str, faces) -> dict: if faces: logger.info("Extracted %d verified face(s) from photo %s", len(faces), photo_id) + _schedule_recluster_debounced() return {'status': 'success', 'photo_id': photo_id, 'faces': len(faces)} +RECLUSTER_DEBOUNCE_KEY = "mule:recluster_faces:pending" +RECLUSTER_DELAY = 120 # seconds after last face extraction + + +def _schedule_recluster_debounced(): + """Schedule a recluster_faces run, debounced so rapid-fire face + extractions don't spawn hundreds of redundant cluster jobs.""" + try: + import redis as _redis + r = _redis.from_url(settings.redis_url) + already_pending = r.set(RECLUSTER_DEBOUNCE_KEY, "1", + ex=RECLUSTER_DELAY, nx=True) + if already_pending: + recluster_faces.apply_async(countdown=RECLUSTER_DELAY) + logger.info("Scheduled debounced recluster_faces in %ds", RECLUSTER_DELAY) + except Exception as e: + logger.debug("recluster debounce check failed: %s", e) + + @shared_task(name='recluster_faces', queue='vision') def recluster_faces(): """Run DBSCAN clustering over all face embeddings and assign/create Tag(kind=face_cluster) entries.""" + # Clear debounce key so new face extractions can schedule another round. + try: + import redis as _redis + _redis.from_url(settings.redis_url).delete(RECLUSTER_DEBOUNCE_KEY) + except Exception: + pass + if not settings.vision.enabled or not settings.vision.faces.enabled: return {'status': 'skipped', 'reason': 'faces disabled'} diff --git a/frontend/src/components/ScanProgress.tsx b/frontend/src/components/ScanProgress.tsx index bb5b3ee..043322f 100644 --- a/frontend/src/components/ScanProgress.tsx +++ b/frontend/src/components/ScanProgress.tsx @@ -1,7 +1,7 @@ import { useEffect, useRef, useState } from 'react' -import { FolderOpen, Loader2, Check, AlertCircle, X } from 'lucide-react' +import { FolderOpen, Loader2, Check, AlertCircle, X, Brain, Sparkles } from 'lucide-react' import { useQuery, useQueryClient } from '@tanstack/react-query' -import { library } from '../services/api' +import { library, WorkerStatus } from '../services/api' import clsx from 'clsx' interface ScanStatus { @@ -12,65 +12,101 @@ interface ScanStatus { errors: string[] } +type Phase = 'idle' | 'scanning' | 'processing' | 'done' + export function ScanProgress() { const [isVisible, setIsVisible] = useState(false) const [isMinimized, setIsMinimized] = useState(false) const queryClient = useQueryClient() const wasScanningRef = useRef(false) + const wasProcessingRef = useRef(false) - // Poll scan status every 2 seconds when scanning const { data: scanStatus } = useQuery({ queryKey: ['scan-status'], - queryFn: async () => { - const response = await library.scanStatus() - return response - }, + queryFn: () => library.scanStatus(), + refetchInterval: (query) => + query.state.data?.is_scanning ? 2000 : 10000, + enabled: true, + }) + + const isScanning = scanStatus?.is_scanning ?? false + + // Poll worker status to track vision queue activity. + // Fast polling (3s) while processing, slow (15s) otherwise. + const { data: workerStatus } = useQuery({ + queryKey: ['worker-status-progress'], + queryFn: () => library.maintenance.workerStatus(), refetchInterval: (query) => { - // Poll every 2 seconds if scanning, otherwise every 10 seconds - return query.state.data?.is_scanning ? 2000 : 10000 + const q = totalQueued(query.state.data) + return q > 0 ? 3000 : 15000 }, enabled: true, }) - useEffect(() => { - const isScanning = scanStatus?.is_scanning ?? false + const visionActive = visionQueued(workerStatus) + const totalActive = totalQueued(workerStatus) - if (isScanning) { + const phase: Phase = isScanning + ? 'scanning' + : totalActive > 0 + ? 'processing' + : 'idle' + + useEffect(() => { + if (phase === 'scanning') { setIsVisible(true) setIsMinimized(false) wasScanningRef.current = true - } else if (wasScanningRef.current) { - // Just transitioned from scanning → done. THIS is the right moment - // to invalidate caches that might have new data: the photos query - // (new files indexed), the folder tree (new folders walked), the - // heap counts (in case a heap photo got reattached). - wasScanningRef.current = false - queryClient.invalidateQueries({ queryKey: ['photos'] }) - queryClient.invalidateQueries({ queryKey: ['folders'] }) - queryClient.invalidateQueries({ queryKey: ['folders', 'tree'] }) - queryClient.invalidateQueries({ queryKey: ['heaps'] }) - queryClient.invalidateQueries({ queryKey: ['tags'] }) - queryClient.invalidateQueries({ queryKey: ['library', 'stats'] }) + wasProcessingRef.current = false + } else if (phase === 'processing') { + // Show widget when processing starts (even without a prior scan, + // e.g. backfill triggered from Settings). + if (!isVisible) setIsVisible(true) + wasProcessingRef.current = true - if (isVisible && (scanStatus?.processed_files ?? 0) > 0) { - // Keep showing for 3 seconds after scan completes - setTimeout(() => { - if (!scanStatus?.is_scanning) { - setIsVisible(false) - } - }, 3000) + if (wasScanningRef.current) { + // Scan just finished — invalidate data caches. + wasScanningRef.current = false + queryClient.invalidateQueries({ queryKey: ['photos'] }) + queryClient.invalidateQueries({ queryKey: ['folders'] }) + queryClient.invalidateQueries({ queryKey: ['folders', 'tree'] }) + queryClient.invalidateQueries({ queryKey: ['heaps'] }) + queryClient.invalidateQueries({ queryKey: ['tags'] }) + queryClient.invalidateQueries({ queryKey: ['library', 'stats'] }) + } + } else if (phase === 'idle') { + if (wasScanningRef.current) { + // Scan finished with no queued processing (small import). + wasScanningRef.current = false + queryClient.invalidateQueries({ queryKey: ['photos'] }) + queryClient.invalidateQueries({ queryKey: ['folders'] }) + queryClient.invalidateQueries({ queryKey: ['folders', 'tree'] }) + queryClient.invalidateQueries({ queryKey: ['heaps'] }) + queryClient.invalidateQueries({ queryKey: ['tags'] }) + queryClient.invalidateQueries({ queryKey: ['library', 'stats'] }) + } + if (wasProcessingRef.current) { + // Processing just drained — refresh tags (new clusters/objects). + wasProcessingRef.current = false + queryClient.invalidateQueries({ queryKey: ['tags'] }) + queryClient.invalidateQueries({ queryKey: ['photos'] }) + queryClient.invalidateQueries({ queryKey: ['library', 'stats'] }) + } + if (isVisible) { + setTimeout(() => setIsVisible(false), 3000) } } - }, [scanStatus?.is_scanning, scanStatus?.processed_files, isVisible, queryClient]) + }, [phase, isVisible, queryClient]) - if (!isVisible || !scanStatus) return null + if (!isVisible) return null - const progress = scanStatus.total_files > 0 + // Scan progress percentage + const scanProgress = scanStatus && scanStatus.total_files > 0 ? (scanStatus.processed_files / scanStatus.total_files) * 100 : 0 - const isComplete = !scanStatus.is_scanning && scanStatus.processed_files > 0 - const hasErrors = scanStatus.errors && scanStatus.errors.length > 0 + const isComplete = phase === 'idle' && (scanStatus?.processed_files ?? 0) > 0 + const hasErrors = scanStatus?.errors && scanStatus.errors.length > 0 return (
setIsMinimized(!isMinimized)} >
- {scanStatus.is_scanning ? ( + {phase === 'scanning' ? ( + ) : phase === 'processing' ? ( + ) : isComplete && !hasErrors ? ( ) : hasErrors ? ( @@ -96,11 +134,13 @@ export function ScanProgress() { )} {!isMinimized && ( - {scanStatus.is_scanning + {phase === 'scanning' ? 'Scanning Folders' + : phase === 'processing' + ? 'Processing Photos' : isComplete - ? 'Scan Complete' - : 'Scan Status'} + ? 'Complete' + : 'Status'} )}
@@ -120,59 +160,80 @@ export function ScanProgress() { {/* Content */} {!isMinimized && (
- {/* Current folder */} - {scanStatus.current_folder && ( -
- {scanStatus.current_folder} -
+ {/* Scan phase */} + {phase === 'scanning' && scanStatus && ( + <> + {scanStatus.current_folder && ( +
+ {scanStatus.current_folder} +
+ )} +
+
+
+
+
+
+ + {scanStatus.processed_files} / {scanStatus.total_files || '?'} files + + + {Math.round(scanProgress)}% + +
+ )} - {/* Progress bar */} -
-
-
+
+ + Analyzing photos… +
+
+ {visionActive > 0 && ( +
+ Vision pipeline + {visionActive} queued +
)} - style={{ width: `${progress}%` }} - /> -
-
+ {(totalActive - visionActive) > 0 && ( +
+ Other tasks + {totalActive - visionActive} queued +
+ )} +
+
+ Thumbnails, embeddings, faces, tags — runs in background +
+ + )} - {/* Stats */} -
- - {scanStatus.processed_files} / {scanStatus.total_files || '?'} files - - - {scanStatus.is_scanning - ? `${Math.round(progress)}%` - : isComplete - ? 'Done' - : 'Idle'} - -
+ {/* Done phase */} + {phase === 'idle' && isComplete && ( +
+ + All processing complete +
+ )} {/* Errors */} {hasErrors && (
- {scanStatus.errors.slice(0, 3).map((error, i) => ( + {scanStatus!.errors.slice(0, 3).map((error, i) => (
• {error}
))} - {scanStatus.errors.length > 3 && ( + {scanStatus!.errors.length > 3 && (
- +{scanStatus.errors.length - 3} more errors + +{scanStatus!.errors.length - 3} more errors
)}
@@ -182,4 +243,22 @@ export function ScanProgress() { )}
) -} \ No newline at end of file +} + +function visionQueued(ws: WorkerStatus | undefined): number { + if (!ws) return 0 + const queued = ws.queues?.['vision'] ?? 0 + const active = ws.workers + ?.filter(w => w.queues?.includes('vision')) + .reduce((sum, w) => sum + (w.active ?? 0) + (w.reserved ?? 0), 0) ?? 0 + return queued + active +} + +function totalQueued(ws: WorkerStatus | undefined): number { + if (!ws) return 0 + const queued = Object.values(ws.queues ?? {}).reduce((a, b) => a + b, 0) + const active = ws.workers?.reduce( + (sum, w) => sum + (w.active ?? 0) + (w.reserved ?? 0), 0 + ) ?? 0 + return queued + active +}