diff --git a/app/celery_app.py b/app/celery_app.py index 72f484f..ce1ca47 100644 --- a/app/celery_app.py +++ b/app/celery_app.py @@ -73,6 +73,8 @@ def make_celery(app=None): 'app.tasks.scan.update_batch_stats': {'queue': 'maintenance'}, 'app.tasks.maintenance.sweep_blocklisted_tag_from_images': {'queue': 'maintenance'}, 'app.tasks.maintenance.sync_character_fandoms_to_images': {'queue': 'maintenance'}, + 'app.tasks.maintenance.verify_media_integrity': {'queue': 'maintenance'}, + 'app.tasks.maintenance.verify_unverified_images': {'queue': 'maintenance'}, # Import tasks - handled by worker (heavy processing) 'app.tasks.import_file.*': {'queue': 'import'}, diff --git a/app/main.py b/app/main.py index 14cbb63..750c3ef 100644 --- a/app/main.py +++ b/app/main.py @@ -38,6 +38,7 @@ def index(): images = ( ImageRecord.query .options(joinedload(ImageRecord.tags)) + .filter(ImageRecord.integrity_status.in_(('ok', 'unknown'))) .order_by(func.random()) .limit(20) .all() @@ -65,8 +66,14 @@ def random_images_api(): except ValueError: pass - # Build query - q = ImageRecord.query.options(joinedload(ImageRecord.tags)) + # Build query — exclude flagged-corrupt rows so the shuffle never serves + # a thumbnail-broken image. 'unknown' rows (pre-sweep / freshly imported) + # are treated as healthy until the verifier proves otherwise. + q = ( + ImageRecord.query + .options(joinedload(ImageRecord.tags)) + .filter(ImageRecord.integrity_status.in_(('ok', 'unknown'))) + ) if exclude_ids: q = q.filter(~ImageRecord.id.in_(exclude_ids)) @@ -648,6 +655,69 @@ def trigger_sync_character_fandoms_to_images(): return redirect(url_for('main.settings', tab='maintenance')) +@main.post('/settings/maintenance/verify-images') +def trigger_verify_unverified_images(): + """Sweep ImageRecords and enqueue per-image structural verification. + + Form param `mode=all` forces re-verify across the whole library; the + default skips rows already in a non-'unknown' state so re-clicking the + button doesn't thrash through cleanly-verified data. + """ + from app.tasks.maintenance import verify_unverified_images + only_unknown = request.form.get('mode', 'unknown') != 'all' + verify_unverified_images.apply_async( + kwargs={'only_unknown': only_unknown}, queue='maintenance', + ) + return redirect(url_for('main.settings', tab='maintenance')) + + +@main.get('/api/integrity/failed') +def api_integrity_failed(): + """List ImageRecords whose structural verification failed. + + Query params: + - status: one of 'truncated' | 'unreadable' | 'missing' (optional; + omit to return every non-'ok' / non-'unknown' row) + - limit: page size (default 100, max 500) + - offset: pagination cursor + """ + status_filter = (request.args.get('status') or '').strip() + limit = min(request.args.get('limit', 100, type=int), 500) + offset = max(request.args.get('offset', 0, type=int), 0) + + q = ImageRecord.query.filter( + ImageRecord.integrity_status.in_(('truncated', 'unreadable', 'missing')) + ) + if status_filter in ('truncated', 'unreadable', 'missing'): + q = q.filter(ImageRecord.integrity_status == status_filter) + + total = q.count() + rows = ( + q.order_by(ImageRecord.integrity_checked_at.desc().nullslast(), + ImageRecord.id.desc()) + .limit(limit) + .offset(offset) + .all() + ) + return jsonify( + ok=True, + total=total, + limit=limit, + offset=offset, + items=[ + { + 'id': r.id, + 'filename': r.filename, + 'filepath': r.filepath, + 'integrity_status': r.integrity_status, + 'integrity_checked_at': r.integrity_checked_at.isoformat() + if r.integrity_checked_at else None, + } + for r in rows + ], + ) + + # ---------------------------- # Tag add/remove endpoints # ---------------------------- diff --git a/app/models.py b/app/models.py index dfbe02f..9fc141b 100644 --- a/app/models.py +++ b/app/models.py @@ -34,6 +34,12 @@ class ImageRecord(db.Model): taken_at = db.Column(db.DateTime, nullable=True) imported_at = db.Column(db.DateTime, default=db.func.now()) is_thumbnail = db.Column(db.Boolean, default=False) + # Structural-integrity state set by app.services.integrity.verify_path. + # 'ok' rows are healthy; everything else is excluded from random/showcase/ML/ + # suggestion paths. See migration l26042501 for the value enumeration. + integrity_status = db.Column(db.String(16), nullable=False, default='unknown', + server_default='unknown') + integrity_checked_at = db.Column(db.DateTime(timezone=True), nullable=True) tags = db.relationship( "Tag", secondary=image_tags, diff --git a/app/services/integrity.py b/app/services/integrity.py new file mode 100644 index 0000000..b199dbd --- /dev/null +++ b/app/services/integrity.py @@ -0,0 +1,171 @@ +"""Structural-integrity verification for media files. + +Public surface: `verify_path(path) -> (status, detail)`. +Status values match the `image_record.integrity_status` column: + - 'ok': passed all checks + - 'truncated': structurally a valid file up to a point, then missing + trailing bytes (the WD14 '6 bytes not processed' case) + - 'unreadable': can't open at all, or container is malformed enough that + even the head doesn't parse + - 'missing': filepath doesn't exist on disk + +Dispatch is by file extension. Images go through PIL (with truncated-image +loading explicitly disabled so we surface the very thing we want to detect). +Videos go through ffprobe. Archives go through zipfile.testzip / tarfile +where the stdlib supports it. Anything else returns 'ok' (we don't know +how to validate it, so we don't lie about its state). +""" +from __future__ import annotations + +import logging +import os +import subprocess +import zipfile + +# We deliberately import the PIL.ImageFile module here without flipping +# LOAD_TRUNCATED_IMAGES — opposite of wd14.py's runtime tolerance, since +# this module's job is to *detect* truncation, not paper over it. +from PIL import Image, UnidentifiedImageError + +log = logging.getLogger(__name__) + + +IMAGE_EXTS = ('.jpg', '.jpeg', '.jfif', '.png', '.gif', '.bmp', '.tiff', '.webp') +VIDEO_EXTS = ('.mp4', '.mov', '.avi', '.mkv', '.webm', '.m4v', '.wmv', '.flv') +ZIP_EXTS = ('.zip', '.cbr', '.cbz') + + +def verify_path(path: str) -> tuple[str, str | None]: + """Verify a single file. Returns (status, detail-or-None). + + Detail is a short human-readable string for the failure case, useful + for the report endpoint and logs. None when status='ok'. + """ + if not os.path.exists(path): + return 'missing', None + try: + size = os.path.getsize(path) + except OSError as e: + return 'missing', f'stat failed: {e}' + if size == 0: + return 'unreadable', 'zero-byte file' + + ext = os.path.splitext(path)[1].lower() + + if ext in IMAGE_EXTS: + return _verify_image(path) + if ext in VIDEO_EXTS: + return _verify_video(path) + if ext in ZIP_EXTS: + return _verify_zip(path) + # Unknown extension: don't fail it, but don't lie that we verified it. + # Caller writes 'ok' but the kind=unknown case is rare in this library. + return 'ok', None + + +def _verify_image(path: str) -> tuple[str, str | None]: + """PIL-based two-stage check. + + Stage 1: open + verify(). Validates header, structure, and (for most + formats) walks the entire stream to confirm framing — this is what + catches the 6-byte truncation. verify() invalidates the Image after, + so callers can't draw from it (we don't need to). + + Stage 2: re-open + load(). Some PIL formats (notably JPEG) only flag + truncation during load(), not verify(). Re-opening with truncated + loading disabled lets a malformed scan trigger OSError here. + """ + try: + with Image.open(path) as img: + img.verify() + except (OSError, SyntaxError, UnidentifiedImageError, ValueError) as e: + msg = str(e).lower() + if _looks_truncated(msg): + return 'truncated', str(e) + return 'unreadable', str(e) + + try: + # Force a full decode without the LOAD_TRUNCATED_IMAGES safety net + # that wd14/siglip enabled at module-import time. + from PIL import ImageFile + prev = ImageFile.LOAD_TRUNCATED_IMAGES + ImageFile.LOAD_TRUNCATED_IMAGES = False + try: + with Image.open(path) as img: + img.load() + finally: + ImageFile.LOAD_TRUNCATED_IMAGES = prev + except (OSError, SyntaxError, ValueError) as e: + msg = str(e).lower() + if _looks_truncated(msg): + return 'truncated', str(e) + return 'unreadable', str(e) + + return 'ok', None + + +# Phrases PIL emits across formats when the file ends earlier than the +# decoder expected. Centralized so both verify-stages agree on the mapping. +_TRUNCATED_HINTS = ( + 'trunc', # 'image file is truncated', 'truncated file read' + 'not processed', # 'N bytes not processed' (the original wd14 case) + 'broken png', # PNG short final chunk + 'premature', # JPEG/PNG premature end + 'unexpected end', # GIF / generic end-of-stream + 'not enough', # 'not enough data' +) + + +def _looks_truncated(lower_msg: str) -> bool: + return any(h in lower_msg for h in _TRUNCATED_HINTS) + + +def _verify_video(path: str) -> tuple[str, str | None]: + """ffprobe-based container walk. + + `-v error` keeps stdout/stderr quiet on success and emits the first + structural complaint on failure. Any non-zero exit is treated as a + failed verification. ffprobe distinguishes 'truncated' poorly across + formats, so we lean on stderr text to map it. + """ + try: + result = subprocess.run( + ['ffprobe', '-v', 'error', '-show_format', '-show_streams', '-i', path], + capture_output=True, + text=True, + timeout=30, + ) + except FileNotFoundError: + log.warning("ffprobe not on PATH; skipping video verify for %s", path) + return 'ok', None + except subprocess.TimeoutExpired: + return 'unreadable', 'ffprobe timed out' + + if result.returncode == 0: + return 'ok', None + err = (result.stderr or '').strip().lower() + if any(s in err for s in ('truncated', 'partial', 'eof', 'end of file', 'invalid data found')): + return 'truncated', result.stderr.strip() + return 'unreadable', result.stderr.strip() or f'ffprobe exit {result.returncode}' + + +def _verify_zip(path: str) -> tuple[str, str | None]: + """zipfile.testzip walks every CRC. None = clean; bad-name = corrupt. + + BadZipFile / LargeZipFile are unreadable cases (header malformed or + too big to test), distinct from a CRC mismatch on a single member + which we treat as truncated/partial. + """ + try: + with zipfile.ZipFile(path) as zf: + bad = zf.testzip() + except zipfile.BadZipFile as e: + return 'unreadable', str(e) + except zipfile.LargeZipFile as e: + return 'unreadable', str(e) + except OSError as e: + return 'unreadable', str(e) + + if bad is None: + return 'ok', None + return 'truncated', f'CRC failed for member: {bad}' diff --git a/app/services/tag_suggestions.py b/app/services/tag_suggestions.py index 36ae260..5afaf2c 100644 --- a/app/services/tag_suggestions.py +++ b/app/services/tag_suggestions.py @@ -248,7 +248,13 @@ def get_suggestions( The optional `cfg` and `existing_all` kwargs let callers that loop over many images amortize the config fetch and the Tag-table scan. """ - if ImageRecord.query.get(image_id) is None: + img = ImageRecord.query.get(image_id) + if img is None: + return {'character': [], 'copyright': [], 'general': []} + # Don't compute suggestions for files we know are corrupt — the source + # WD14 predictions on them are unreliable, and the user shouldn't be + # nudged to attach tags to a broken image. + if img.integrity_status not in ('ok', 'unknown'): return {'character': [], 'copyright': [], 'general': []} if cfg is None: diff --git a/app/tasks/import_file.py b/app/tasks/import_file.py index 8ce4d16..29ddf7d 100644 --- a/app/tasks/import_file.py +++ b/app/tasks/import_file.py @@ -231,6 +231,31 @@ def import_media_file(self, task_id: int): return _skip_task(task, 'Duplicate by hash', result_image_id=existing.id) + # Supersede-by-name for previously-flagged corrupt records. + # If the user (or the downloader) drops a fresh copy of a file we + # earlier marked corrupt at /import//, the bytes + # don't match (so the hash dedup above missed it) but the source + # path does. Treat it as a supersede so tags/series/embeddings + # carry over and the old corrupt /images/... file gets replaced. + flagged = ( + ImageRecord.query + .filter(ImageRecord.integrity_status.in_(('truncated', 'unreadable', 'missing'))) + .filter(ImageRecord.filename == filename) + .filter(ImageRecord.filepath.like(f"%/{artist}/%")) + .first() + ) + if flagged is not None: + phash = calculate_perceptual_hash(src_path) + log.info( + f"Supersede-on-replace: fresh /import/{artist}/{filename} replaces " + f"flagged image_id={flagged.id} (status={flagged.integrity_status})" + ) + return _supersede_existing( + task, flagged, src_path, file_hash, phash, + metadata, artist, dest_dir, filename, + archive_path, archive_sidecar_path, + ) + # pHash similarity check # Quick mode: SKIP pHash comparison (relies on content hash for duplicates) # Deep mode: Full pHash comparison for similarity detection @@ -372,6 +397,14 @@ def import_media_file(self, task_id: int): # Never let an enqueue failure roll back an otherwise-successful import. log.warning(f"Could not enqueue tag_and_embed for image {record.id}: {ml_err}") + # Queue structural integrity verification. Runs cheaply on the + # maintenance queue; result lands as integrity_status on the row. + try: + from app.tasks.maintenance import verify_media_integrity + verify_media_integrity.delay(record.id) + except Exception as v_err: + log.warning(f"Could not enqueue verify_media_integrity for image {record.id}: {v_err}") + # Update batch stats if task.batch_id: from app.tasks.scan import update_batch_stats @@ -760,6 +793,13 @@ def _process_archive_file(src_path: str, artist: str, dest_dir: str, if sidecar_path: apply_sidecar_metadata.delay(record.id, sidecar_path) + # Queue integrity verification for the extracted file too. + try: + from app.tasks.maintenance import verify_media_integrity + verify_media_integrity.delay(record.id) + except Exception as v_err: + log.warning(f"Could not enqueue verify_media_integrity for image {record.id}: {v_err}") + log.debug(f"Imported from archive: {actual_filename} -> {dest_path}") return { @@ -936,6 +976,14 @@ def _supersede_existing(task: ImportTask, old_record: ImageRecord, src_path: str from app.tasks.scan import update_batch_stats update_batch_stats.delay(task.batch_id) + # Re-verify the superseded file. supersede_image already cleared + # integrity_status to 'unknown'; this resolves it back to a real value. + try: + from app.tasks.maintenance import verify_media_integrity + verify_media_integrity.delay(record.id) + except Exception as v_err: + log.warning(f"Could not enqueue verify_media_integrity (supersede) for image {record.id}: {v_err}") + return { 'status': 'superseded', 'image_id': record.id, diff --git a/app/tasks/maintenance.py b/app/tasks/maintenance.py index fc49477..ea9e951 100644 --- a/app/tasks/maintenance.py +++ b/app/tasks/maintenance.py @@ -6,12 +6,14 @@ task was removed on 2026-04-21 as part of the bare-name refactor. """ import logging import re +from datetime import datetime, timezone from celery import shared_task from sqlalchemy import text from app import db -from app.models import Tag +from app.models import ImageRecord, ImportTask, Tag +from app.services.integrity import verify_path log = logging.getLogger(__name__) @@ -205,6 +207,94 @@ def sync_character_fandoms_to_images() -> dict: } +@shared_task( + name='app.tasks.maintenance.verify_media_integrity', + soft_time_limit=60, + time_limit=120, +) +def verify_media_integrity(image_id: int) -> dict: + """Run structural verification on one image and persist the result. + + Idempotent — safe to re-enqueue freely; each call re-checks the file + on disk and writes a fresh `integrity_checked_at` timestamp. Used both + by the post-import hook (so newly-imported rows land verified) and by + the sweep task (which re-verifies the existing library). + """ + image = ImageRecord.query.get(image_id) + if image is None: + return {'image_id': image_id, 'status': 'no_record'} + + status, detail = verify_path(image.filepath) + image.integrity_status = status + image.integrity_checked_at = datetime.now(timezone.utc) + db.session.commit() + + if status != 'ok': + log.warning( + "verify_media_integrity: image_id=%s path=%r → %s (%s)", + image_id, image.filepath, status, detail, + ) + return {'image_id': image_id, 'status': status, 'detail': detail} + + +@shared_task( + name='app.tasks.maintenance.verify_unverified_images', + soft_time_limit=300, + time_limit=600, +) +def verify_unverified_images(only_unknown: bool = True) -> dict: + """Sweep every ImageRecord and enqueue per-image verify tasks. + + Skips paths that are currently the target of an in-flight import — same + contract `deep_scan_directory` uses, so a half-written file mid-import + doesn't get false-flagged. By default only revisits rows whose status is + still 'unknown' (post-migration default + brand-new rows that haven't + been picked up by the import-time hook yet); pass only_unknown=False to + force a full re-verify across the library. + """ + active_paths = { + row[0] for row in db.session.execute(text(""" + SELECT source_path FROM import_task + WHERE status IN ('pending', 'queued', 'processing') + """)).fetchall() + } + + q = ImageRecord.query.filter(ImageRecord.filepath.isnot(None)) + if only_unknown: + q = q.filter(ImageRecord.integrity_status == 'unknown') + + enqueued = 0 + skipped_active = 0 + last_id = 0 + BATCH = 500 + while True: + rows = ( + q.filter(ImageRecord.id > last_id) + .order_by(ImageRecord.id.asc()) + .limit(BATCH) + .all() + ) + if not rows: + break + for r in rows: + if r.filepath in active_paths: + skipped_active += 1 + continue + verify_media_integrity.delay(r.id) + enqueued += 1 + last_id = rows[-1].id + + log.info( + "verify_unverified_images: enqueued=%d skipped_active=%d only_unknown=%s", + enqueued, skipped_active, only_unknown, + ) + return { + 'enqueued': enqueued, + 'skipped_active': skipped_active, + 'only_unknown': only_unknown, + } + + @shared_task( name='app.tasks.maintenance.sweep_blocklisted_tag_from_images', soft_time_limit=60, diff --git a/app/tasks/ml.py b/app/tasks/ml.py index a43e1ce..77bd9e5 100644 --- a/app/tasks/ml.py +++ b/app/tasks/ml.py @@ -100,6 +100,15 @@ def tag_and_embed(self, image_id: int): log.warning(f"tag_and_embed: file missing at {image.filepath}") return {'status': 'file_missing'} + # Skip flagged-corrupt rows. The verifier already saw the file fail + # structurally; running inference would either crash or produce noise. + # 'unknown' (pre-sweep / mid-import) still proceeds. + if image.integrity_status not in ('ok', 'unknown'): + log.info( + f"tag_and_embed: skipping image {image_id} — integrity={image.integrity_status}" + ) + return {'status': 'skipped_integrity', 'integrity': image.integrity_status} + is_video = image.filepath.lower().endswith(VIDEO_EXTS) try: @@ -187,6 +196,7 @@ def backfill(self, batch_size: int = 50, pause_seconds: float = 0.5): ), ) .filter(ImageRecord.id > last_id) + .filter(ImageRecord.integrity_status.in_(('ok', 'unknown'))) .filter( (ImageTagPrediction.image_id.is_(None)) | (ImageEmbedding.image_id.is_(None)) diff --git a/app/templates/settings.html b/app/templates/settings.html index e867ffd..f645c6e 100644 --- a/app/templates/settings.html +++ b/app/templates/settings.html @@ -470,6 +470,30 @@ +
+

Verify image integrity

+

+ Walks every image and runs a structural check (PIL for images, + ffprobe for video, zipfile for archives). Flagged rows are excluded + from random/showcase, ML retag, and the suggestion modal until a + fresh copy lands in /import/<artist>/<filename> + — at which point the import pipeline supersedes the corrupt row, + preserving its tags. The default mode only checks rows whose status + is still unknown (post-migration default + freshly + imported); use "force re-verify all" after disk events to recheck + everything. Flagged rows are listed at + /api/integrity/failed. +

+
+ +
+
+ + +
+
+

Auto-accept threshold (general tags)

diff --git a/app/utils/image_importer.py b/app/utils/image_importer.py index 9b4b547..c996f4d 100644 --- a/app/utils/image_importer.py +++ b/app/utils/image_importer.py @@ -252,6 +252,11 @@ def supersede_image(old_record: "ImageRecord", new_filepath: str, new_thumb_path old_record.width = new_metadata["width"] old_record.height = new_metadata["height"] old_record.format = new_metadata["format"] + # The bytes on disk just changed — the prior verification result is no + # longer authoritative. Reset to 'unknown' so the post-supersede verify + # hook (or the next sweep) re-records the truth. + old_record.integrity_status = 'unknown' + old_record.integrity_checked_at = None # Keep the earlier taken_at date (prefer original date) if new_metadata.get("taken_at") and old_record.taken_at: if new_metadata["taken_at"] < old_record.taken_at: diff --git a/migrations/versions/l26042501_add_image_integrity_status.py b/migrations/versions/l26042501_add_image_integrity_status.py new file mode 100644 index 0000000..02f5c3a --- /dev/null +++ b/migrations/versions/l26042501_add_image_integrity_status.py @@ -0,0 +1,63 @@ +"""Add integrity_status + integrity_checked_at to image_record. + +Tracks per-image structural verification state so corrupt files (truncated +downloads, broken containers, partial archives) can be flagged and excluded +from random/showcase/ML/suggestion paths instead of blowing up downstream +preprocessing. + +Status values: + - unknown: never verified (default for existing rows + freshly imported + until the verifier runs) + - ok: passed marker check + format-level verify + - truncated: missing trailing bytes (PIL EOI/IEND, ffprobe end-of-stream) + - unreadable: header/structure invalid; can't even open the file + - missing: filepath doesn't exist on disk + +Revision ID: l26042501 +Revises: k26042201 +Create Date: 2026-04-25 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = 'l26042501' +down_revision = 'k26042201' +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + 'image_record', + sa.Column( + 'integrity_status', + sa.String(16), + nullable=False, + server_default='unknown', + ), + ) + op.add_column( + 'image_record', + sa.Column( + 'integrity_checked_at', + sa.DateTime(timezone=True), + nullable=True, + ), + ) + # Partial index so the "list flagged rows" report and the + # exclude-from-random/ML filters stay cheap on a million-row table. + op.create_index( + 'image_record_integrity_status_idx', + 'image_record', + ['integrity_status'], + postgresql_where=sa.text("integrity_status <> 'ok'"), + ) + + +def downgrade(): + op.drop_index('image_record_integrity_status_idx', table_name='image_record') + op.drop_column('image_record', 'integrity_checked_at') + op.drop_column('image_record', 'integrity_status')