f290784bf3
Two copies of IMG_1234.jpg sitting in different folders looked identical on the duplicates grid — same filename, same dimensions, same Best heuristic. The user had no way to pick which copy to keep without opening each in the preview overlay. Backend: include filepath in the per-member payload from GET /api/v1/library/duplicates/groups (was filename-only). Frontend: a black 65% strip at the bottom of every duplicate thumbnail showing the parent folder name (the actual discriminator when filenames match), with the full filepath surfaced via the native title tooltip on hover. The dimensions chip moves from bottom-left to top-left so the bottom strip can run edge-to-edge. memberToPhoto stops faking filepath=filename (a years-old workaround that broke any code path needing the real path); the synthetic Photo the grid hands to PhotoThumbnail now carries the real filepath. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
853 lines
31 KiB
Python
853 lines
31 KiB
Python
"""
|
||
Library API router for stats, scanning, and maintenance.
|
||
|
||
The /maintenance/* endpoints are surfaced through the frontend Settings
|
||
panel. They're intentionally idempotent and operate by re-queueing the
|
||
existing Celery tasks rather than doing any heavy lifting in the
|
||
request thread.
|
||
"""
|
||
import logging
|
||
import os
|
||
import shutil
|
||
from typing import List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import select, func, update, true as sa_true
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.database import get_db
|
||
from app.models import Photo
|
||
from app.models.folders import SourceRoot
|
||
from app.models.user import User
|
||
from app.dependencies import get_current_user
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _owner_filter(user: User, scope: str | None):
|
||
"""Return a column expression scoping photos to the current user,
|
||
or a pass-through true() when an admin requests global scope."""
|
||
if scope == "global" and user.role == "admin":
|
||
return sa_true()
|
||
return Photo.user_id == user.id
|
||
|
||
# Media types we accept in the regenerate-thumbnails request body. Mirrors
|
||
# the values produced by `app.tasks.scan.get_media_type`.
|
||
_VALID_MEDIA_TYPES = {'photo', 'raw', 'heic', 'video'}
|
||
|
||
@router.get("/stats")
|
||
async def get_library_stats(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Get library statistics. Pass ?scope=global (admin only) for
|
||
cross-user totals (used by the Settings page)."""
|
||
owner = _owner_filter(current_user, scope)
|
||
visible = owner & (Photo.is_discarded.is_(False)) & (Photo.is_hidden.is_(False))
|
||
|
||
all_photos_count = (
|
||
await db.execute(select(func.count(Photo.id)).where(visible))
|
||
).scalar() or 0
|
||
|
||
rated_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(visible, Photo.rating >= 1)
|
||
)
|
||
).scalar() or 0
|
||
|
||
colored_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(
|
||
visible, Photo.color_label.is_not(None)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
|
||
with_gps_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(
|
||
visible, Photo.latitude.is_not(None)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
|
||
duplicates_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(
|
||
visible, Photo.is_duplicate.is_(True)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
|
||
discarded_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(owner, Photo.is_discarded.is_(True))
|
||
)
|
||
).scalar() or 0
|
||
|
||
needs_review_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(visible, Photo.needs_review.is_(True))
|
||
)
|
||
).scalar() or 0
|
||
|
||
# Legacy split (kept for the existing /stats consumers).
|
||
photo_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(
|
||
owner,
|
||
Photo.media_type.in_(['photo', 'heic', 'raw'])
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
video_count = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(owner, Photo.media_type == 'video')
|
||
)
|
||
).scalar() or 0
|
||
|
||
size = (await db.execute(select(func.sum(Photo.file_size)).where(owner))).scalar() or 0
|
||
|
||
# Source root directories (active ones only). Scoped to the
|
||
# requesting user unless they're an admin asking for global view —
|
||
# otherwise the Settings panel would leak other users' NC paths.
|
||
sr_query = select(SourceRoot.path).where(SourceRoot.is_active.is_(True))
|
||
if not (scope == "global" and current_user.role == "admin"):
|
||
sr_query = sr_query.where(SourceRoot.user_id == current_user.id)
|
||
roots = (
|
||
await db.execute(sr_query.order_by(SourceRoot.path))
|
||
).scalars().all()
|
||
|
||
return {
|
||
"all_photos": all_photos_count,
|
||
"rated": rated_count,
|
||
"colored": colored_count,
|
||
"with_gps": with_gps_count,
|
||
"duplicates": duplicates_count,
|
||
"discarded": discarded_count,
|
||
"needs_review": needs_review_count,
|
||
"total_photos": photo_count,
|
||
"total_videos": video_count,
|
||
"total_size": size,
|
||
"total_size_gb": round(size / (1024**3), 2) if size else 0,
|
||
"source_dirs": roots,
|
||
}
|
||
|
||
@router.post("/scan")
|
||
async def trigger_scan(current_user: User = Depends(get_current_user)):
|
||
"""Trigger full library re-scan"""
|
||
from app.tasks.scan import scan_all_source_roots
|
||
|
||
scan_all_source_roots.delay()
|
||
|
||
return {"status": "success", "message": "Library scan started"}
|
||
|
||
|
||
@router.post("/maintenance/recover-stuck")
|
||
async def recover_stuck_photos(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""Reset photos stuck in 'processing' for more than 30 minutes back to
|
||
'pending' so the pipeline can retry them. Returns the count of recovered
|
||
photos."""
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
cutoff = datetime.now(timezone.utc) - timedelta(minutes=30)
|
||
result = await db.execute(
|
||
update(Photo)
|
||
.where(
|
||
Photo.processing_status == 'processing',
|
||
Photo.updated_at < cutoff,
|
||
)
|
||
.values(
|
||
processing_status='pending',
|
||
processing_error='Auto-recovered from stuck processing state',
|
||
)
|
||
)
|
||
await db.commit()
|
||
count = result.rowcount
|
||
if count:
|
||
logger.info("Recovered %d stuck photos back to pending", count)
|
||
return {"status": "success", "recovered": count}
|
||
|
||
|
||
@router.post("/backfill-gps")
|
||
async def trigger_backfill_gps(current_user: User = Depends(get_current_user)):
|
||
"""Re-run EXIF metadata extraction on every photo that's still missing
|
||
GPS coordinates. Useful after fixing the EXIF parser, or any time the
|
||
Map view looks emptier than expected. Returns immediately — work runs
|
||
on the Celery worker."""
|
||
from app.tasks.scan import backfill_gps
|
||
|
||
backfill_gps.delay()
|
||
return {"status": "success", "message": "GPS backfill queued"}
|
||
|
||
|
||
@router.post("/maintenance/backfill-taken-at")
|
||
async def trigger_backfill_taken_at(current_user: User = Depends(get_current_user)):
|
||
"""Re-run extract_metadata on every non-manual photo to recompute
|
||
taken_at with the current EXIF-priority list and path-based fallback.
|
||
Useful after the date-extraction logic changes (e.g. dropping the
|
||
ModifyDate fallback). Manual edits are preserved."""
|
||
from app.services.metadata import backfill_taken_at
|
||
|
||
backfill_taken_at.delay()
|
||
return {"status": "success", "message": "taken_at backfill queued"}
|
||
|
||
|
||
@router.get("/scan/status")
|
||
async def get_scan_status(db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
|
||
"""Get current scan status"""
|
||
import redis
|
||
from app.config import settings
|
||
|
||
# Connect to Redis to get scan status
|
||
r = redis.Redis.from_url(settings.redis_url)
|
||
|
||
# Get scan status from Redis (set by worker tasks)
|
||
is_scanning = r.get('scan:active') == b'true'
|
||
current_folder = r.get('scan:current_folder')
|
||
processed_files = int(r.get('scan:processed_files') or 0)
|
||
total_files = int(r.get('scan:total_files') or 0)
|
||
errors = r.lrange('scan:errors', 0, -1)
|
||
|
||
return {
|
||
"is_scanning": is_scanning,
|
||
"current_folder": current_folder.decode() if current_folder else None,
|
||
"processed_files": processed_files,
|
||
"total_files": total_files,
|
||
"errors": [e.decode() for e in errors] if errors else []
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Maintenance endpoints — surfaced via the Settings panel.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class RegenerateThumbnailsRequest(BaseModel):
|
||
"""Optional filters narrowing which photos get re-queued. With both
|
||
fields omitted the request resets every photo in the library."""
|
||
media_types: Optional[List[str]] = Field(
|
||
default=None,
|
||
description="Restrict to these media_type values (photo/raw/heic/video).",
|
||
)
|
||
only_failed: bool = Field(
|
||
default=False,
|
||
description="If true, only re-queue photos whose processing_status is 'failed'.",
|
||
)
|
||
only_pending: bool = Field(
|
||
default=False,
|
||
description="If true, only (re-)queue photos whose processing_status is 'pending'. "
|
||
"Useful for kicking rows that were created by a scan but never had "
|
||
"their thumbnail task picked up.",
|
||
)
|
||
|
||
|
||
@router.get("/maintenance/thumbnail-stats")
|
||
async def get_thumbnail_stats(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Counts of photos by processing_status, plus a media-type breakdown
|
||
so the Settings panel can show the user what's outstanding."""
|
||
owner = _owner_filter(current_user, scope)
|
||
status_rows = (
|
||
await db.execute(
|
||
select(Photo.processing_status, func.count(Photo.id))
|
||
.where(owner)
|
||
.group_by(Photo.processing_status)
|
||
)
|
||
).all()
|
||
|
||
media_rows = (
|
||
await db.execute(
|
||
select(Photo.media_type, func.count(Photo.id))
|
||
.where(owner)
|
||
.group_by(Photo.media_type)
|
||
)
|
||
).all()
|
||
|
||
by_status = {status or 'unknown': count for status, count in status_rows}
|
||
by_media_type = {media or 'unknown': count for media, count in media_rows}
|
||
total = sum(by_status.values())
|
||
|
||
return {
|
||
"total": total,
|
||
"pending": by_status.get('pending', 0),
|
||
"processing": by_status.get('processing', 0),
|
||
"completed": by_status.get('completed', 0),
|
||
"failed": by_status.get('failed', 0),
|
||
"by_media_type": by_media_type,
|
||
}
|
||
|
||
|
||
@router.post("/maintenance/regenerate-thumbnails")
|
||
async def regenerate_thumbnails(
|
||
body: RegenerateThumbnailsRequest,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Reset matching photos' on-disk thumbnail directories and re-queue
|
||
Celery thumbnail generation. Used by the Settings panel for the
|
||
'regenerate video thumbnails' / 'regenerate failed' buttons.
|
||
|
||
Files on disk are removed under /data/thumbs/<photo_id>/ so the next
|
||
request to /photos/{id}/thumb/{size} actually re-generates instead of
|
||
serving the stale placeholder.
|
||
"""
|
||
from app.tasks.thumbs import generate_thumbnails
|
||
|
||
owner = _owner_filter(current_user, scope)
|
||
|
||
# Validate media_types early so a typo can't silently match nothing.
|
||
media_types = body.media_types
|
||
if media_types is not None:
|
||
invalid = [m for m in media_types if m not in _VALID_MEDIA_TYPES]
|
||
if invalid:
|
||
return {
|
||
"status": "error",
|
||
"message": f"Invalid media_types: {invalid}. "
|
||
f"Allowed: {sorted(_VALID_MEDIA_TYPES)}",
|
||
}
|
||
|
||
query = select(Photo).where(owner)
|
||
if media_types:
|
||
query = query.where(Photo.media_type.in_(media_types))
|
||
if body.only_failed:
|
||
query = query.where(Photo.processing_status == 'failed')
|
||
if body.only_pending:
|
||
query = query.where(Photo.processing_status == 'pending')
|
||
|
||
photos = (await db.execute(query)).scalars().all()
|
||
|
||
cleared_dirs = 0
|
||
file_errors = 0
|
||
for photo in photos:
|
||
thumb_dir = f"/data/thumbs/{photo.id}"
|
||
if os.path.isdir(thumb_dir):
|
||
try:
|
||
shutil.rmtree(thumb_dir)
|
||
cleared_dirs += 1
|
||
except OSError as e:
|
||
file_errors += 1
|
||
logger.warning(f"Could not clear thumb dir {thumb_dir}: {e}")
|
||
photo.processing_status = 'pending'
|
||
photo.processing_error = None
|
||
photo.thumb_small = None
|
||
photo.thumb_medium = None
|
||
photo.thumb_large = None
|
||
|
||
await db.commit()
|
||
|
||
# Queue celery tasks AFTER the commit so the worker sees the reset
|
||
# state when it picks the job up.
|
||
queued = 0
|
||
for photo in photos:
|
||
try:
|
||
generate_thumbnails.delay(photo.id)
|
||
queued += 1
|
||
except Exception as e:
|
||
logger.warning(f"Could not queue thumbnail job for {photo.id}: {e}")
|
||
|
||
return {
|
||
"status": "success",
|
||
"matched": len(photos),
|
||
"queued": queued,
|
||
"cleared_dirs": cleared_dirs,
|
||
"file_errors": file_errors,
|
||
"filters": {
|
||
"media_types": media_types,
|
||
"only_failed": body.only_failed,
|
||
},
|
||
}
|
||
|
||
|
||
@router.get("/maintenance/worker-status")
|
||
async def get_worker_status(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Diagnostics for the Celery worker fleet + recent task failures.
|
||
|
||
Surfaced in the Settings panel so the user can spot a stuck queue or
|
||
a worker that's gone away without tailing container logs. Returns:
|
||
|
||
- workers: list of {name, status, active, concurrency, queues}
|
||
derived from celery_app.control.inspect(). `status` is 'online'
|
||
when ping succeeds, 'unreachable' otherwise. Empty list means no
|
||
workers are responding at all (broker down, container crashed,
|
||
wrong queue routing, etc.).
|
||
- queues: per-queue depth read from Redis (LLEN of each queue key
|
||
used by celery.kombu). Mirrors what tasks are waiting to be
|
||
picked up.
|
||
- failures: aggregate count of photos with processing_status='failed'
|
||
plus the most recent N error messages so the user can see *why*
|
||
things failed without opening the DB.
|
||
- broker_ok: bool — could we even reach Redis?
|
||
"""
|
||
owner = _owner_filter(current_user, scope)
|
||
from app.tasks.celery import celery_app
|
||
from app.config import settings
|
||
import redis as _redis
|
||
|
||
# ----- Celery inspect (workers + active tasks) -------------------------
|
||
# Each inspect.* call is a separate broadcast-and-wait with its own
|
||
# timeout, so running them serially multiplies the wait. Fan them out
|
||
# to threads and gather, collapsing 6 × timeout into ~1 × timeout.
|
||
# Timeout dropped to 0.5s — a responsive worker answers within a few
|
||
# ms; anything past that is effectively "not responding" for the
|
||
# purposes of a settings dashboard.
|
||
import asyncio
|
||
workers: list[dict] = []
|
||
inspect_error: Optional[str] = None
|
||
try:
|
||
inspect = celery_app.control.inspect(timeout=0.5)
|
||
ping, active, reserved, scheduled, stats, active_queues = await asyncio.gather(
|
||
asyncio.to_thread(inspect.ping),
|
||
asyncio.to_thread(inspect.active),
|
||
asyncio.to_thread(inspect.reserved),
|
||
asyncio.to_thread(inspect.scheduled),
|
||
asyncio.to_thread(inspect.stats),
|
||
asyncio.to_thread(inspect.active_queues),
|
||
)
|
||
ping = ping or {}
|
||
active = active or {}
|
||
reserved = reserved or {}
|
||
scheduled = scheduled or {}
|
||
stats = stats or {}
|
||
active_queues = active_queues or {}
|
||
|
||
worker_names = set(ping) | set(active) | set(stats)
|
||
for name in sorted(worker_names):
|
||
wstats = stats.get(name) or {}
|
||
pool = wstats.get('pool') or {}
|
||
workers.append({
|
||
"name": name,
|
||
"status": "online" if name in ping else "unreachable",
|
||
"active": len(active.get(name, []) or []),
|
||
"reserved": len(reserved.get(name, []) or []),
|
||
"scheduled": len(scheduled.get(name, []) or []),
|
||
"concurrency": pool.get('max-concurrency'),
|
||
"processed": (wstats.get('total') or {}),
|
||
"queues": [q.get('name') for q in (active_queues.get(name) or [])],
|
||
"active_tasks": [
|
||
{
|
||
"id": t.get('id'),
|
||
"name": t.get('name'),
|
||
"args": t.get('args'),
|
||
"time_start": t.get('time_start'),
|
||
}
|
||
for t in (active.get(name) or [])[:10]
|
||
],
|
||
})
|
||
except Exception as e:
|
||
inspect_error = str(e)
|
||
logger.warning(f"Celery inspect failed: {e}")
|
||
|
||
# ----- Broker / queue depth --------------------------------------------
|
||
broker_ok = False
|
||
queue_depths: dict[str, int] = {}
|
||
broker_error: Optional[str] = None
|
||
try:
|
||
r = _redis.Redis.from_url(settings.redis_url, socket_timeout=1.0)
|
||
r.ping()
|
||
broker_ok = True
|
||
# `vision` runs the content classifier — the only heavy queue.
|
||
for q in ('default', 'high', 'low', 'vision'):
|
||
try:
|
||
queue_depths[q] = int(r.llen(q) or 0)
|
||
except Exception:
|
||
queue_depths[q] = 0
|
||
except Exception as e:
|
||
broker_error = str(e)
|
||
logger.warning(f"Redis broker unreachable: {e}")
|
||
|
||
# ----- Recent task failures from the photos table ----------------------
|
||
failed_total = (
|
||
await db.execute(
|
||
select(func.count(Photo.id)).where(owner, Photo.processing_status == 'failed')
|
||
)
|
||
).scalar() or 0
|
||
|
||
recent_failed_rows = (
|
||
await db.execute(
|
||
select(
|
||
Photo.id,
|
||
Photo.filename,
|
||
Photo.media_type,
|
||
Photo.processing_error,
|
||
Photo.updated_at,
|
||
)
|
||
.where(owner, Photo.processing_status == 'failed')
|
||
.order_by(Photo.updated_at.desc().nullslast())
|
||
.limit(20)
|
||
)
|
||
).all()
|
||
|
||
recent_failures = [
|
||
{
|
||
"photo_id": row[0],
|
||
"filename": row[1],
|
||
"media_type": row[2],
|
||
"error": (row[3] or '')[:500],
|
||
"updated_at": row[4].isoformat() if row[4] else None,
|
||
}
|
||
for row in recent_failed_rows
|
||
]
|
||
|
||
# ----- Most recent scan errors (Redis list) ----------------------------
|
||
scan_errors: list[str] = []
|
||
try:
|
||
if broker_ok:
|
||
r = _redis.Redis.from_url(settings.redis_url, socket_timeout=1.0)
|
||
raw = r.lrange('scan:errors', 0, 19) or []
|
||
scan_errors = [e.decode(errors='replace') for e in raw]
|
||
except Exception as e:
|
||
logger.debug(f"Could not read scan:errors: {e}")
|
||
|
||
return {
|
||
"broker_ok": broker_ok,
|
||
"broker_error": broker_error,
|
||
"inspect_error": inspect_error,
|
||
"workers": workers,
|
||
"worker_count": len(workers),
|
||
"queues": queue_depths,
|
||
"failures": {
|
||
"total": failed_total,
|
||
"recent": recent_failures,
|
||
},
|
||
"scan_errors": scan_errors,
|
||
}
|
||
|
||
|
||
@router.get("/maintenance/pipeline-stats")
|
||
async def get_pipeline_stats(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Per-stage progress across the ingestion pipeline.
|
||
|
||
Returns a `{stage_key: {done, total, label}}` map so the Settings
|
||
panel can render one progress bar per stage. `total` is the number
|
||
of non-discarded photos the stage is *expected* to run on — which is
|
||
every non-discarded photo for most stages, or a narrower subset when
|
||
a stage is image-only (e.g. embeddings don't run on videos).
|
||
|
||
Keep the shape flat + serialisable; the frontend turns it straight
|
||
into a list of rows without needing to know about the models.
|
||
"""
|
||
from app.models.tags import photo_tags # association Table, not a model
|
||
|
||
owner = _owner_filter(current_user, scope)
|
||
not_discarded = owner & Photo.is_discarded.is_(False)
|
||
|
||
async def scalar_count(query):
|
||
return (await db.execute(query)).scalar() or 0
|
||
|
||
# Total non-discarded photos — the denominator for most stages.
|
||
total_photos = await scalar_count(
|
||
select(func.count(Photo.id)).where(not_discarded)
|
||
)
|
||
|
||
# Image-only denominator (embeddings, tags, faces, OCR, phash). We
|
||
# exclude videos because those stages either don't apply or run off
|
||
# the extracted video frame which is treated separately.
|
||
total_images = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded, Photo.media_type != 'video'
|
||
)
|
||
)
|
||
|
||
completed = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded, Photo.processing_status == 'completed'
|
||
)
|
||
)
|
||
with_exif = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded, Photo.exif_json.is_not(None)
|
||
)
|
||
)
|
||
with_gps = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded,
|
||
Photo.latitude.is_not(None),
|
||
Photo.longitude.is_not(None),
|
||
)
|
||
)
|
||
with_phash = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded, Photo.phash.is_not(None)
|
||
)
|
||
)
|
||
|
||
# Classified: distinct photos with a content_type tag.
|
||
classified_done = await scalar_count(
|
||
select(func.count(func.distinct(photo_tags.c.photo_id)))
|
||
.select_from(photo_tags)
|
||
.join(Photo, Photo.id == photo_tags.c.photo_id)
|
||
.where(not_discarded, photo_tags.c.source == 'vision:clip_classifier')
|
||
)
|
||
needs_review_count = await scalar_count(
|
||
select(func.count(Photo.id)).where(not_discarded, Photo.needs_review.is_(True))
|
||
)
|
||
|
||
duplicate_groups = await scalar_count(
|
||
select(func.count(func.distinct(Photo.duplicate_group_id))).where(
|
||
not_discarded, Photo.duplicate_group_id.is_not(None)
|
||
)
|
||
)
|
||
duplicate_members = await scalar_count(
|
||
select(func.count(Photo.id)).where(
|
||
not_discarded, Photo.duplicate_group_id.is_not(None)
|
||
)
|
||
)
|
||
|
||
# Ordered list so the frontend renders stages in pipeline order
|
||
# without needing to know the sequence itself.
|
||
stages = [
|
||
{
|
||
"key": "thumbnails",
|
||
"label": "Thumbnails & pHash",
|
||
"done": completed,
|
||
"total": total_photos,
|
||
"hint": "Generated on scan. Unlocks every downstream stage.",
|
||
},
|
||
{
|
||
"key": "exif",
|
||
"label": "EXIF metadata",
|
||
"done": with_exif,
|
||
"total": total_photos,
|
||
"hint": "Camera, lens, capture time. Required for GPS + taken_at.",
|
||
},
|
||
{
|
||
"key": "gps",
|
||
"label": "GPS coordinates",
|
||
"done": with_gps,
|
||
"total": total_photos,
|
||
"hint": "Subset of EXIF. Drives the map view; many photos legitimately have none.",
|
||
"partial": True, # not every photo is expected to have GPS
|
||
},
|
||
{
|
||
"key": "phash",
|
||
"label": "Perceptual hashes",
|
||
"done": with_phash,
|
||
"total": total_images,
|
||
"hint": "Feeds duplicate detection.",
|
||
},
|
||
{
|
||
"key": "classification",
|
||
"label": "Content classification (photo vs other)",
|
||
"done": classified_done,
|
||
"total": total_images,
|
||
"hint": f"{needs_review_count} photos flagged for review.",
|
||
},
|
||
{
|
||
"key": "duplicates",
|
||
"label": "Duplicate groups",
|
||
"done": duplicate_groups,
|
||
"total": duplicate_groups, # same — current count, not a progress ratio
|
||
"hint": f"{duplicate_members} photos in {duplicate_groups} groups. Run regroup_duplicates after new imports.",
|
||
"standalone": True,
|
||
},
|
||
]
|
||
|
||
return {
|
||
"total_photos": total_photos,
|
||
"total_images": total_images,
|
||
"stages": stages,
|
||
}
|
||
|
||
|
||
@router.get("/maintenance/missing-stats")
|
||
async def get_missing_stats(current_user: User = Depends(get_current_user)):
|
||
"""Count photos whose files no longer exist on disk under a mounted
|
||
source root. Surfaced in Settings so the user can see a number before
|
||
pulling the trigger on prune-missing. Cheap enough to call freely."""
|
||
from app.services.cleanup import prune_missing_photos
|
||
return await prune_missing_photos(dry_run=True)
|
||
|
||
|
||
@router.post("/maintenance/prune-missing")
|
||
async def run_prune_missing(current_user: User = Depends(get_current_user)):
|
||
"""Actually delete the orphaned photo rows reported by /missing-stats.
|
||
Common cause: PHOTO_DIRS in .env was repointed at a different library
|
||
leaving every old row dangling. Skips any photo whose source root
|
||
isn't currently mounted (almost always means an unmounted drive)."""
|
||
from app.services.cleanup import prune_missing_photos
|
||
try:
|
||
return {"status": "success", **(await prune_missing_photos(dry_run=False))}
|
||
except Exception as e:
|
||
logger.error(f"Prune missing failed: {e}")
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
@router.post("/maintenance/cleanup")
|
||
async def run_data_integrity_cleanup(current_user: User = Depends(get_current_user)):
|
||
"""Re-run the source-roots / folders / photos data-integrity cleanup
|
||
that normally only runs on backend startup. Idempotent."""
|
||
from app.services.cleanup import cleanup_data_integrity
|
||
|
||
try:
|
||
await cleanup_data_integrity()
|
||
return {"status": "success"}
|
||
except Exception as e:
|
||
logger.error(f"Manual cleanup failed: {e}")
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────
|
||
# Duplicate detection
|
||
# ─────────────────────────────────────────────────────────────────────────
|
||
|
||
@router.get("/duplicates/groups")
|
||
async def get_duplicate_groups(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
scope: str | None = Query(None),
|
||
):
|
||
"""Return every duplicate group with its members.
|
||
|
||
Drives the frontend grouped grid view in the Duplicates section. One
|
||
SQL query, bucketed in Python — no N+1, no per-member fetch. Groups
|
||
are sorted by member_count DESC then earliest taken_at DESC so the
|
||
biggest / most recent clusters bubble to the top.
|
||
|
||
Each group also carries a `reason` field:
|
||
* "exact" — every member shares the same SHA-256 (true byte
|
||
duplicates that the perceptual hash trivially caught)
|
||
* "similar" — members differ at the byte level but match perceptually
|
||
"""
|
||
owner = _owner_filter(current_user, scope)
|
||
rows = (
|
||
await db.execute(
|
||
select(
|
||
Photo.id,
|
||
Photo.filename,
|
||
Photo.filepath,
|
||
Photo.taken_at,
|
||
Photo.file_size,
|
||
Photo.width,
|
||
Photo.height,
|
||
Photo.thumb_small,
|
||
Photo.file_hash,
|
||
Photo.folder_id,
|
||
Photo.media_type,
|
||
Photo.duplicate_group_id,
|
||
)
|
||
.where(owner)
|
||
.where(Photo.duplicate_group_id.is_not(None))
|
||
.where(Photo.is_discarded.is_(False))
|
||
.where(Photo.is_hidden.is_(False))
|
||
.order_by(Photo.duplicate_group_id)
|
||
)
|
||
).all()
|
||
|
||
# Bucket members by group_id. filepath is included so the
|
||
# Duplicates view can show "which folder does this copy live in"
|
||
# — the discriminator the user needs to pick a winner.
|
||
groups: dict[str, list[dict]] = {}
|
||
for row in rows:
|
||
member = {
|
||
"id": row[0],
|
||
"filename": row[1],
|
||
"filepath": row[2],
|
||
"taken_at": row[3].isoformat() if row[3] else None,
|
||
"file_size": row[4],
|
||
"width": row[5],
|
||
"height": row[6],
|
||
"thumb_small": row[7],
|
||
"file_hash": row[8],
|
||
"folder_id": row[9],
|
||
"media_type": row[10],
|
||
}
|
||
groups.setdefault(row[11], []).append(member)
|
||
|
||
def earliest(g: list[dict]) -> str:
|
||
# Used as a secondary sort key. Photos with no taken_at sort last
|
||
# by returning a far-future sentinel.
|
||
taken = [m["taken_at"] for m in g if m["taken_at"]]
|
||
return min(taken) if taken else "9999"
|
||
|
||
out = []
|
||
for group_id, members in groups.items():
|
||
if len(members) < 2:
|
||
# Defensive: a regroup race could leave a singleton briefly.
|
||
# Skip it so the UI never shows a "group of 1".
|
||
continue
|
||
# exact iff every member shares the same non-null file_hash
|
||
# (true byte-identical copies that pHash also caught). Anything
|
||
# else — different hashes, missing hashes — counts as "similar".
|
||
all_hashes = [m["file_hash"] for m in members]
|
||
reason = (
|
||
"exact"
|
||
if len(set(all_hashes)) == 1 and all_hashes[0] is not None
|
||
else "similar"
|
||
)
|
||
out.append({
|
||
"group_id": group_id,
|
||
"member_count": len(members),
|
||
"reason": reason,
|
||
"members": members,
|
||
})
|
||
|
||
out.sort(key=lambda g: (-g["member_count"], earliest(g["members"])))
|
||
return {
|
||
"groups": out,
|
||
"total_groups": len(out),
|
||
"total_members": sum(g["member_count"] for g in out),
|
||
}
|
||
|
||
|
||
@router.post("/maintenance/regroup-duplicates")
|
||
async def trigger_regroup_duplicates(current_user: User = Depends(get_current_user)):
|
||
"""Recompute duplicate groups from current perceptual hashes.
|
||
|
||
Fires the celery `regroup_duplicates` task which walks every photo's
|
||
phash, clusters by Hamming distance, and rewrites duplicate_group_id /
|
||
is_duplicate columns. Idempotent."""
|
||
from app.tasks.thumbs import regroup_duplicates_task
|
||
try:
|
||
regroup_duplicates_task.delay()
|
||
return {"status": "queued"}
|
||
except Exception as e:
|
||
logger.error(f"Regroup queue failed: {e}")
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
@router.post("/maintenance/backfill-phashes")
|
||
async def trigger_backfill_phashes(current_user: User = Depends(get_current_user)):
|
||
"""Compute perceptual hashes for every photo currently missing one.
|
||
|
||
One-shot recovery path for libraries that existed before the phash
|
||
column was added — the thumbs worker computes phash for everything
|
||
new, but old rows need a backfill pass."""
|
||
from app.tasks.thumbs import backfill_phashes
|
||
try:
|
||
backfill_phashes.delay()
|
||
return {"status": "queued"}
|
||
except Exception as e:
|
||
logger.error(f"Backfill queue failed: {e}")
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
@router.post("/maintenance/start-watcher")
|
||
async def start_file_watcher(current_user: User = Depends(get_current_user)):
|
||
"""Start the filesystem watcher. Uses a Redis lock so only one
|
||
instance runs at a time — safe to call repeatedly."""
|
||
from app.tasks.scan import watch_folders
|
||
try:
|
||
watch_folders.apply_async(countdown=2)
|
||
return {"status": "queued"}
|
||
except Exception as e:
|
||
logger.error(f"Watcher queue failed: {e}")
|
||
return {"status": "error", "message": str(e)} |