7b153f0d28
The taken_at extractor walked four EXIF fields in order: DateTimeOriginal,
CreateDate, MediaCreateDate, ModifyDate. The last one is set every time
a file is re-saved (Lightroom export, EXIF strip, batch resize), so any
photo whose original capture metadata was lost during editing ended up
labeled 'exif' with the *edit* date instead of the shoot date.
Changes:
- SubSecDateTimeOriginal at the top of the list (sub-second precision,
often carries OffsetTime).
- QuickTime:CreateDate added next to MediaCreateDate.
- ModifyDate dropped from the trusted list entirely.
- When no trusted EXIF date is present, fall back to guess_date_from_path
(already used for has_date_warning) and tag taken_at_source='path'.
Better than filesystem mtime, which on Nextcloud-mounted libraries
just reflects the upload time.
- Skip the date-write block entirely if photo.taken_at_source == 'manual'
so a rescan can't clobber a user correction.
- parse_exif_datetime: handle the all-zero placeholder some cameras
emit, accept tz-aware variants (%z), normalize to naive UTC.
Frontend: new 'PATH' badge in TakenAtEditor with a tooltip explaining
the date came from filename / folder rather than real EXIF.
Backfill: new backfill_taken_at celery task and
POST /api/v1/library/maintenance/backfill-taken-at endpoint that
re-enqueues extract_metadata for every non-manual photo. ~21k tasks
finish in ~15 min on the existing worker-light concurrency.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
849 lines
31 KiB
Python
849 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.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.
|
||
groups: dict[str, list[dict]] = {}
|
||
for row in rows:
|
||
member = {
|
||
"id": row[0],
|
||
"filename": row[1],
|
||
"taken_at": row[2].isoformat() if row[2] else None,
|
||
"file_size": row[3],
|
||
"width": row[4],
|
||
"height": row[5],
|
||
"thumb_small": row[6],
|
||
"file_hash": row[7],
|
||
"folder_id": row[8],
|
||
"media_type": row[9],
|
||
}
|
||
groups.setdefault(row[10], []).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)} |