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

refactor: drop AI/vision pipeline + plain Postgres + full-refresh script

Browse Source 
Removes the OpenCLIP-on-ONNX classifier and everything that fed or
consumed it:
  - backend: app/services/vision/, app/tasks/vision.py,
    app/services/feature_flags.py, app/routers/features.py — all
    deleted; admin AI/feature-flag endpoints and the worker-vision
    bootstrap call gone. Photo.needs_review and its index dropped.
  - frontend: AI Settings tab, useFeaturesQuery hook, FeatureFlag
    types, "Needs Review" sidebar entry + filter, needs_review filter
    URL param all gone.
  - infra: worker-vision compose service + models_data volume deleted;
    worker-light command no longer runs bootstrap_models; the db
    image switches from pgvector/pgvector:pg16 to postgres:16; backend
    Dockerfile drops the dedicated torch RUN layer; requirements.txt
    drops torch/torchvision/open-clip-torch/onnxruntime.

Alembic 0019_drop_ai_remnants:
  - drops photos.needs_review + ix_photos_needs_review
  - DROP EXTENSION IF EXISTS vector (must run before the image swap;
    the new postgres:16 doesn't ship pgvector)

New scripts/full_refresh.py: one-shot DB ↔ filesystem reconciliation.
Runs cleanup_data_integrity, scans every active SourceRoot inline
(no celery dependency so the worker can be stopped), hard-prunes
photo + folder rows for files that are gone, removes orphan
/data/thumbs/{user}/{photo}/ directories. New helper
prune_orphan_thumbnails in cleanup.py.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
claudio
2026-05-14 00:20:38 +02:00
parent 6915c30911
commit a27267f7ad
39 changed files with 265 additions and 1573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
backend/app/tasks/celery.py
View File

@@ -2,15 +2,12 @@
Celery configuration and app initialization
"""
import logging
import os
from celery import Celery
from celery.signals import worker_process_init
from app.config import settings
logger = logging.getLogger(__name__)
# Create Celery app
celery_app = Celery(
'mulita',
broker=settings.celery_broker_url,
@@ -19,43 +16,31 @@ celery_app = Celery(
'app.tasks.scan',
'app.tasks.thumbs',
'app.tasks.video',
'app.tasks.vision',
'app.services.metadata', # extract_metadata lives here
'app.services.metadata',
]
)
# Configure Celery
celery_app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
# Robust acknowledgment: keep message in broker until task succeeds.
task_acks_late=True,
task_reject_on_worker_lost=True,
# Global time limits — individual tasks can override via decorator.
task_soft_time_limit=300, # 5 min — raises SoftTimeLimitExceeded
task_time_limit=600, # 10 min — SIGKILL
# Explicit routes for every task name. Wildcard patterns don't match
# short names produced by @shared_task(name='...').
task_soft_time_limit=300,
task_time_limit=600,
task_routes={
# Vision queue — CPU-bound binary classification
'classify_content': {'queue': 'vision'},
'vision_fanout': {'queue': 'vision'},
# High-priority queue — thumbnails & duplicates
'generate_thumbnails': {'queue': 'high'},
'regenerate_all_thumbnails': {'queue': 'high'},
'backfill_phashes': {'queue': 'high'},
'regroup_duplicates': {'queue': 'high'},
'incremental_regroup_duplicates': {'queue': 'high'},
# Low-priority queue — scans
'scan_folder': {'queue': 'low'},
'scan_all_source_roots': {'queue': 'low'},
'backfill_gps': {'queue': 'low'},
# Pre-transcode HEVC videos in the background so /playback is a
# cache hit on first user click. CPU-heavy but tolerant of the
# low-priority queue (it doesn't block any user-facing flow).
# CPU-heavy but tolerant of the low-priority queue (doesn't block
# any user-facing flow).
'pretranscode_video': {'queue': 'low'},
# `watch_folders` is retired (file events come from NC webhooks)
# but the task definition still exists as a no-op shim for any
@@ -79,21 +64,3 @@ celery_app.conf.update(
},
},
)
@worker_process_init.connect
def _warmup_vision_models(**kwargs):
"""Pre-load vision models in the worker process so the first task
doesn't pay cold-start latency. Only runs on the vision queue."""
# The worker name contains the queue — only warm up vision workers.
worker_queues = os.environ.get("CELERY_QUEUES", "")
if "vision" not in worker_queues:
# Heuristic: check the celery command line for -Q vision
import sys
if "vision" not in " ".join(sys.argv):
return
try:
from app.services.vision.registry import registry
registry.warmup()
except Exception:
logger.exception("Vision model warmup failed")

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

@@ -479,7 +479,6 @@ async def _scan_all_source_roots_async():
still hit Settings → Re-detect duplicates to force a fresh pass.
"""
from app.tasks.thumbs import incremental_regroup_duplicates_task
from app.tasks.vision import backfill_vision
async with AsyncSessionLocal() as session:
result = await session.execute(
@@ -511,13 +510,6 @@ 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}")
# NOTE: we used to auto-queue `backfill_gps` here so photos
# scanned before the GPS-extraction fix would eventually get
# their coordinates populated. That fix shipped a long time

17
backend/app/tasks/thumbs.py
View File

@@ -48,11 +48,10 @@ THUMB_SIZES = {
}
# Sizes the worker writes to /data/thumbs. Empty set since Phase 4 —
# the API serves all sizes via Nextcloud's /core/preview proxy, and
# the vision worker also fetches NC previews on demand instead of
# reading a local cache. generate_thumbnails still runs the decode-
# and-pHash side-effect (perceptual dedup is mule-only and needs the
# original-resolution pixels) but no longer touches the disk.
# the API serves all sizes via Nextcloud's /core/preview proxy.
# generate_thumbnails still runs the decode-and-pHash side-effect
# (perceptual dedup is mule-only and needs the original-resolution
# pixels) but no longer touches the disk.
WORKER_THUMB_SIZES: set[str] = set()
def get_thumb_path(photo_id: str, size: str, user_id: str = None) -> str:
@@ -396,14 +395,6 @@ async def _generate_thumbnails_async(photo_id: str, task):
logger.info(f"Thumbnails generated for photo {photo_id}")
# Dispatch vision pipeline only after thumbnails succeeded —
# vision tasks need the generated thumbnails to run inference.
try:
from app.tasks.vision import vision_fanout
vision_fanout.delay(photo_id)
except Exception as e:
logger.warning(f"Could not dispatch vision_fanout for {photo_id}: {e}")
return {'status': 'success', 'photo_id': photo_id}
except Exception as e:

263
backend/app/tasks/vision.py
View File

@@ -1,263 +0,0 @@
"""
Celery tasks for the vision pipeline.
A single binary classifier decides whether a photo is 'photography' or
'other'. Photos classified as 'other' get needs_review=true so the user
can triage screenshots / documents / memes in the UI.
"""
import logging
from pathlib import Path
import numpy as np
from celery import shared_task
from sqlalchemy import create_engine, text as sa_text, select, delete, update
from sqlalchemy.orm import Session, sessionmaker
from PIL import Image
from app.config import settings
from app.services.feature_flags import is_enabled, FLAG_VISION_ENABLED
logger = logging.getLogger(__name__)
VISION_READY_KEY = "mulita:vision:ready"
def _vision_worker_ready() -> bool:
try:
import redis as _redis
return bool(_redis.from_url(settings.redis_url).exists(VISION_READY_KEY))
except Exception:
return False
_sync_engine = None
def _get_sync_engine():
global _sync_engine
if _sync_engine is None:
sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "")
_sync_engine = create_engine(sync_url, pool_pre_ping=True, pool_size=3, max_overflow=5)
return _sync_engine
def _get_sync_session() -> Session:
return sessionmaker(bind=_get_sync_engine())()
# CLIP was trained against 640px medium thumbs that the on-disk
# pipeline used to produce. Now we ask Nextcloud's preview endpoint
# for the same edge size so the classifier sees the same input
# distribution.
_VISION_PREVIEW_PX = 640
def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None:
"""Load the photo's RGB pixels into a numpy array for inference.
Primary path: ask Nextcloud's `/index.php/core/preview` for a
640px preview via the sync helper. Replaces the disk read of
`/data/thumbs/{photo_id}/medium.webp` so the on-disk pipeline
can retire entirely.
Disk fallback (transitional): if NC has no preview or no
credentials, look for the medium.webp the old pipeline wrote.
Goes dead once `generate_thumbnails` stops writing files.
"""
from io import BytesIO
from app.models import Photo
from app.models.user import User
from app.services.nextcloud_dav import get_preview_bytes
user_id: str | None = None
fileid: int | None = None
session = _get_sync_session()
try:
row = session.execute(
select(Photo.user_id, Photo.nextcloud_fileid).where(
Photo.id == photo_id
)
).one_or_none()
if row:
user_id, fileid = row[0], row[1]
finally:
session.close()
if user_id and fileid:
session = _get_sync_session()
try:
owner = session.execute(
select(User).where(User.id == user_id)
).scalar_one_or_none()
finally:
session.close()
if owner is not None and owner.nextcloud_app_password_enc:
try:
body = get_preview_bytes(
owner, fileid, _VISION_PREVIEW_PX, _VISION_PREVIEW_PX,
)
except Exception as e:
logger.warning(
"NC preview fetch failed for %s: %s", photo_id, e
)
body = None
if body:
try:
img = Image.open(BytesIO(body)).convert("RGB")
img.load()
arr = np.array(img)
img.close()
return arr
except Exception as e:
logger.warning(
"NC preview decode failed for %s: %s", photo_id, e
)
# Legacy disk fallback — transitional, dead once thumbs.py stops
# writing /data/thumbs.
thumb_base = Path("/data/thumbs")
thumb_path = thumb_base / photo_id / f"{size}.webp"
if not thumb_path.exists():
matches = list(thumb_base.glob(f"*/{photo_id}/{size}.webp"))
if matches:
thumb_path = matches[0]
else:
logger.warning("Thumbnail not found anywhere for %s", photo_id)
return None
try:
img = Image.open(thumb_path).convert("RGB")
img.load()
arr = np.array(img)
img.close()
return arr
except Exception as e:
logger.warning("Corrupt or unreadable thumbnail for %s: %s", photo_id, e)
return None
@shared_task(name='vision_fanout', queue='vision')
def vision_fanout(photo_id: str):
"""Dispatch vision work for a photo. Today this is just the binary
classifier; the indirection stays so scanner/upload code keeps one
entrypoint."""
if not is_enabled(FLAG_VISION_ENABLED):
return {'status': 'skipped', 'reason': 'vision disabled'}
classify_content.delay(photo_id)
return {'status': 'dispatched', 'photo_id': photo_id}
@shared_task(name='classify_content', queue='vision', bind=True, max_retries=3)
def classify_content(self, photo_id: str):
"""Run the binary classifier and write:
- a Tag(kind='content_type', name IN ('photography','other'))
- Photo.needs_review = (label == 'other')
"""
if not is_enabled(FLAG_VISION_ENABLED):
return {'status': 'skipped', 'reason': 'vision disabled'}
image = _load_thumb(photo_id, "medium")
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
try:
from app.services.vision.registry import registry
classifier = registry.get_classifier()
result = classifier.classify(image)
except Exception as exc:
logger.exception("classify_content failed for %s", photo_id)
raise self.retry(exc=exc, countdown=60)
from app.models import Photo
from app.models.tags import Tag, photo_tags
source_name = "vision:clip_classifier"
label = result.label
confidence = result.confidence
session = _get_sync_session()
try:
photo = session.execute(
select(Photo).where(Photo.id == photo_id)
).scalar_one_or_none()
if photo is None:
return {'status': 'error', 'message': 'photo not found'}
owner_id = photo.user_id
# Drop any previous classification for this photo.
session.execute(
delete(photo_tags).where(
photo_tags.c.photo_id == photo_id,
photo_tags.c.source == source_name,
)
)
tag = session.execute(
select(Tag).where(
Tag.name == label, Tag.kind == 'content_type', Tag.user_id == owner_id
)
).scalar_one_or_none()
if not tag:
tag = Tag(name=label, kind='content_type', source=source_name, user_id=owner_id)
session.add(tag)
session.flush()
session.execute(
photo_tags.insert().values(
photo_id=photo_id,
tag_id=tag.id,
confidence=confidence,
source=source_name,
)
)
session.execute(
update(Photo)
.where(Photo.id == photo_id)
.values(needs_review=(label == 'other'))
)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
logger.info("[%s] Classified %s as %s (%.2f)", self.request.id, photo_id, label, confidence)
return {'status': 'success', 'photo_id': photo_id, 'label': label}
@shared_task(name='backfill_vision', bind=True, max_retries=10)
def backfill_vision(self, limit: int | None = None, **_ignored):
"""Queue classify_content for photos without a content_type tag."""
if not _vision_worker_ready():
logger.info("Vision worker not ready yet — retrying in 30s")
raise self.retry(countdown=30)
ordering = "ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST"
limit_clause = " LIMIT :lim" if limit else ""
params: dict = {}
if limit:
params["lim"] = int(limit)
session = _get_sync_session()
try:
sql = f"""
SELECT p.id FROM photos p
WHERE p.processing_status = 'completed'
AND NOT EXISTS (
SELECT 1 FROM photo_tags pt
WHERE pt.photo_id = p.id
AND pt.source = 'vision:clip_classifier'
)
{ordering}{limit_clause}
"""
ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()]
finally:
session.close()
for pid in ids:
classify_content.delay(pid)
logger.info("Backfill queued %d photos for classification", len(ids))
return {'status': 'queued', 'count': len(ids)}