diff --git a/app/__init__.py b/app/__init__.py index 2e3152f..58d3bf0 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -13,6 +13,9 @@ import sqlite3 db = SQLAlchemy() migrate = Migrate() +# Celery instance - will be configured with app context in create_app +celery = None + # ============================================================================= # CLI Commands @@ -25,6 +28,34 @@ def version_check_command(): from app.utils.version_migration import check_and_run_migrations check_and_run_migrations() + +@click.command("celery-worker") +@click.option('--queues', '-Q', default='scan,import,thumbnail,sidecar,default', + help='Comma-separated list of queues to consume') +@click.option('--concurrency', '-c', default=2, type=int, + help='Number of worker processes') +@with_appcontext +def celery_worker_command(queues, concurrency): + """Start a Celery worker for processing import tasks.""" + from app.celery_app import celery as celery_app + celery_app.worker_main([ + 'worker', + f'--queues={queues}', + f'--concurrency={concurrency}', + '--loglevel=info' + ]) + + +@click.command("celery-beat") +@with_appcontext +def celery_beat_command(): + """Start Celery Beat scheduler for periodic tasks.""" + from app.celery_app import celery as celery_app + celery_app.worker_main([ + 'beat', + '--loglevel=info' + ]) + @event.listens_for(Engine, "connect") def set_sqlite_pragma(dbapi_connection, connection_record): # Only apply to SQLite connections @@ -54,5 +85,13 @@ def create_app(config_class='config.Config'): # Register CLI commands app.cli.add_command(version_check_command) + app.cli.add_command(celery_worker_command) + app.cli.add_command(celery_beat_command) + + # Initialize Celery with Flask app context + from app.celery_app import make_celery + global celery + celery = make_celery(app) + app.celery = celery return app diff --git a/app/celery_app.py b/app/celery_app.py new file mode 100644 index 0000000..ede860f --- /dev/null +++ b/app/celery_app.py @@ -0,0 +1,123 @@ +# app/celery_app.py +""" +Celery application factory and configuration. +Provides task queue functionality for the ImageRepo import pipeline. +""" +import os +from celery import Celery + +# Default broker/backend URLs +DEFAULT_BROKER = 'redis://redis:6379/0' +DEFAULT_BACKEND = 'redis://redis:6379/0' + + +def make_celery(app=None): + """ + Create Celery instance with Flask app context support. + + Args: + app: Flask application instance (optional) + + Returns: + Configured Celery instance + """ + # Get broker/backend from app config or environment + if app: + broker = app.config.get('CELERY_BROKER_URL', DEFAULT_BROKER) + backend = app.config.get('CELERY_RESULT_BACKEND', DEFAULT_BACKEND) + else: + broker = os.environ.get('CELERY_BROKER_URL', DEFAULT_BROKER) + backend = os.environ.get('CELERY_RESULT_BACKEND', DEFAULT_BACKEND) + + celery = Celery( + 'imagerepo', + broker=broker, + backend=backend, + include=[ + 'app.tasks.scan', + 'app.tasks.import_file', + 'app.tasks.thumbnail', + 'app.tasks.sidecar' + ] + ) + + # Worker concurrency from environment + concurrency = int(os.environ.get('CELERY_WORKER_CONCURRENCY', '2')) + + celery.conf.update( + # Serialization + task_serializer='json', + accept_content=['json'], + result_serializer='json', + timezone='UTC', + enable_utc=True, + + # Concurrency - intentionally low for steady background processing + worker_concurrency=concurrency, + worker_prefetch_multiplier=1, # One task at a time per worker process + + # Task routing - separate queues for different task types + task_routes={ + 'app.tasks.scan.*': {'queue': 'scan'}, + 'app.tasks.import_file.*': {'queue': 'import'}, + 'app.tasks.thumbnail.*': {'queue': 'thumbnail'}, + 'app.tasks.sidecar.*': {'queue': 'sidecar'}, + }, + + # Task default queue for unrouted tasks + task_default_queue='default', + + # Result backend settings + result_expires=86400, # 24 hours + + # Task execution settings for resume capability + task_acks_late=True, # Acknowledge after task completes + task_reject_on_worker_lost=True, # Requeue if worker dies + + # Time limits + task_soft_time_limit=300, # 5 minutes soft limit + task_time_limit=600, # 10 minutes hard limit (archives may need more) + + # Periodic task schedule (Celery Beat) + beat_schedule={ + 'periodic-import-scan': { + 'task': 'app.tasks.scan.scan_directory', + 'schedule': int(os.environ.get('IMPORT_EVERY_SECONDS', '28800')), # 8 hours default + 'args': ('/import', '/images'), + }, + 'recover-interrupted-tasks': { + 'task': 'app.tasks.scan.recover_interrupted_tasks', + 'schedule': 300, # Every 5 minutes + }, + }, + ) + + if app: + # Don't pass Flask config directly to Celery - it contains old-style keys + # that conflict with Celery's new lowercase format. + # The broker/backend are already set above from app.config. + + # Create a task base class that runs within Flask app context + class ContextTask(celery.Task): + def __call__(self, *args, **kwargs): + with app.app_context(): + return self.run(*args, **kwargs) + + celery.Task = ContextTask + + return celery + + +def create_celery_with_app(): + """ + Create Celery instance with Flask app context for standalone workers. + This is used when running `celery -A app.celery_app:celery worker`. + """ + from app import create_app + flask_app = create_app() + return make_celery(flask_app) + + +# Create celery instance with Flask app context +# This ensures workers have access to the database and Flask extensions +celery = create_celery_with_app() diff --git a/app/main.py b/app/main.py index 3a48c0b..ccae610 100644 --- a/app/main.py +++ b/app/main.py @@ -7,7 +7,7 @@ from sqlalchemy.orm import joinedload from collections import OrderedDict from datetime import datetime -from app.models import ImageRecord, Tag, ArchiveRecord, image_tags +from app.models import ImageRecord, Tag, ArchiveRecord, image_tags, ImportTask, ImportBatch from app import db import os import shutil @@ -987,3 +987,299 @@ def delete_filtered_images(): deleted_count=deleted_count, errors=errors if errors else None ) + + +# ---------------------------- +# Import Task Queue API (Celery) +# ---------------------------- + +@main.route('/api/import/trigger', methods=['POST']) +def trigger_import_celery(): + """ + Trigger a directory scan and import via Celery. + Replaces the old flag-file mechanism. + """ + try: + from app.tasks.scan import scan_directory + + source_dir = request.form.get('source_dir', '/import') + dest_dir = request.form.get('dest_dir', '/images') + + result = scan_directory.delay(source_dir, dest_dir) + + return jsonify({ + 'ok': True, + 'task_id': result.id, + 'message': 'Import scan started' + }) + except Exception as e: + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/import/status') +def import_queue_status(): + """ + Get current import queue status. + Returns counts by task status and recent task info. + """ + try: + # Count tasks by status + status_counts = dict( + db.session.query( + ImportTask.status, + func.count(ImportTask.id) + ).group_by(ImportTask.status).all() + ) + + # Get active batch info + active_batch = ImportBatch.query.filter_by(status='running').order_by( + ImportBatch.started_at.desc() + ).first() + + # Recent tasks (last 20) + recent_tasks = ImportTask.query.order_by( + ImportTask.created_at.desc() + ).limit(20).all() + + return jsonify({ + 'ok': True, + 'status_counts': { + 'pending': status_counts.get('pending', 0), + 'queued': status_counts.get('queued', 0), + 'processing': status_counts.get('processing', 0), + 'complete': status_counts.get('complete', 0), + 'failed': status_counts.get('failed', 0), + 'skipped': status_counts.get('skipped', 0), + }, + 'active_batch': { + 'id': active_batch.id, + 'source_directory': active_batch.source_directory, + 'total_files': active_batch.total_files, + 'processed_files': active_batch.processed_files, + 'imported_count': active_batch.imported_count, + 'skipped_count': active_batch.skipped_count, + 'failed_count': active_batch.failed_count, + 'started_at': active_batch.started_at.isoformat() if active_batch.started_at else None + } if active_batch else None, + 'recent_tasks': [ + { + 'id': t.id, + 'source_path': os.path.basename(t.source_path) if t.source_path else None, + 'task_type': t.task_type, + 'status': t.status, + 'error_message': t.error_message[:100] if t.error_message else None, + 'created_at': t.created_at.isoformat() if t.created_at else None + } + for t in recent_tasks + ] + }) + except Exception as e: + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/import/task/') +def get_import_task(task_id): + """ + Get detailed status for a specific import task. + """ + task = ImportTask.query.get_or_404(task_id) + + # Check Celery task status if still processing + celery_status = None + if task.celery_task_id: + try: + from app.celery_app import celery + result = celery.AsyncResult(task.celery_task_id) + celery_status = result.status + except Exception: + pass + + return jsonify({ + 'ok': True, + 'task': { + 'id': task.id, + 'source_path': task.source_path, + 'file_hash': task.file_hash, + 'file_size': task.file_size, + 'task_type': task.task_type, + 'status': task.status, + 'celery_status': celery_status, + 'celery_task_id': task.celery_task_id, + 'error_message': task.error_message, + 'retry_count': task.retry_count, + 'context': task.context, + 'batch_id': task.batch_id, + 'result_image_id': task.result_image_id, + 'created_at': task.created_at.isoformat() if task.created_at else None, + 'started_at': task.started_at.isoformat() if task.started_at else None, + 'completed_at': task.completed_at.isoformat() if task.completed_at else None + } + }) + + +@main.route('/api/import/retry-failed', methods=['POST']) +def retry_failed_import_tasks(): + """ + Re-queue all failed import tasks. + """ + try: + from app.tasks.import_file import import_media_file, import_archive + + failed_tasks = ImportTask.query.filter_by(status='failed').all() + retried = 0 + + for task in failed_tasks: + task.status = 'pending' + task.error_message = None + task.retry_count = 0 + task.started_at = None + task.completed_at = None + + if task.task_type == 'import_image': + result = import_media_file.delay(task.id) + elif task.task_type == 'import_archive': + result = import_archive.delay(task.id) + else: + continue + + task.celery_task_id = result.id + task.status = 'queued' + retried += 1 + + db.session.commit() + + return jsonify({ + 'ok': True, + 'retried': retried + }) + except Exception as e: + db.session.rollback() + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/import/clear-completed', methods=['POST']) +def clear_completed_tasks(): + """ + Clear completed and skipped import tasks from the database. + Keeps failed tasks for review. + """ + try: + deleted = ImportTask.query.filter( + ImportTask.status.in_(['complete', 'skipped']) + ).delete(synchronize_session=False) + + db.session.commit() + + return jsonify({ + 'ok': True, + 'deleted': deleted + }) + except Exception as e: + db.session.rollback() + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/thumbnails/regenerate', methods=['POST']) +def regenerate_thumbnails_celery(): + """ + Trigger thumbnail regeneration for all images via Celery. + """ + try: + from app.tasks.thumbnail import regenerate_all_thumbnails + + overwrite = request.form.get('overwrite', 'true').lower() == 'true' + result = regenerate_all_thumbnails.delay(overwrite=overwrite) + + return jsonify({ + 'ok': True, + 'task_id': result.id, + 'message': 'Thumbnail regeneration started' + }) + except Exception as e: + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/thumbnails/regenerate-missing', methods=['POST']) +def regenerate_missing_thumbnails_celery(): + """ + Generate thumbnails only for images that don't have them. + """ + try: + from app.tasks.thumbnail import regenerate_missing_thumbnails + + result = regenerate_missing_thumbnails.delay() + + return jsonify({ + 'ok': True, + 'task_id': result.id, + 'message': 'Missing thumbnail generation started' + }) + except Exception as e: + return jsonify({'ok': False, 'error': str(e)}), 500 + + +@main.route('/api/celery/status') +def celery_cluster_status(): + """ + Get Celery cluster status (workers, queues, etc.). + """ + try: + from app.celery_app import celery + + inspect = celery.control.inspect() + + # Get worker stats + active = inspect.active() or {} + scheduled = inspect.scheduled() or {} + reserved = inspect.reserved() or {} + stats = inspect.stats() or {} + + # Count active tasks + active_count = sum(len(tasks) for tasks in active.values()) + + return jsonify({ + 'ok': True, + 'workers': list(stats.keys()), + 'worker_count': len(stats), + 'active_tasks': active_count, + 'active': active, + 'scheduled': scheduled, + 'reserved': reserved + }) + except Exception as e: + return jsonify({ + 'ok': False, + 'error': str(e), + 'message': 'Celery may not be running' + }), 503 + + +@main.route('/api/import/batches') +def list_import_batches(): + """ + List recent import batches with statistics. + """ + limit = request.args.get('limit', 10, type=int) + + batches = ImportBatch.query.order_by( + ImportBatch.started_at.desc() + ).limit(limit).all() + + return jsonify({ + 'ok': True, + 'batches': [ + { + 'id': b.id, + 'source_directory': b.source_directory, + 'status': b.status, + 'total_files': b.total_files, + 'processed_files': b.processed_files, + 'imported_count': b.imported_count, + 'skipped_count': b.skipped_count, + 'failed_count': b.failed_count, + 'started_at': b.started_at.isoformat() if b.started_at else None, + 'completed_at': b.completed_at.isoformat() if b.completed_at else None + } + for b in batches + ] + }) diff --git a/app/models.py b/app/models.py index 03ea892..5943e1c 100644 --- a/app/models.py +++ b/app/models.py @@ -76,3 +76,80 @@ class AppSettings(db.Model): key = db.Column(db.String(64), unique=True, nullable=False) value = db.Column(db.Text, nullable=True) updated_at = db.Column(db.DateTime, default=db.func.now(), onupdate=db.func.now()) + + +class ImportBatch(db.Model): + """ + Groups related import tasks for batch tracking and reporting. + One batch per scan operation. + """ + __tablename__ = "import_batch" + + id = db.Column(db.Integer, primary_key=True) + source_directory = db.Column(db.String(1024), nullable=False) + status = db.Column(db.String(32), default='running', index=True) # running, complete, failed, cancelled + + # Statistics + total_files = db.Column(db.Integer, default=0) + processed_files = db.Column(db.Integer, default=0) + imported_count = db.Column(db.Integer, default=0) + skipped_count = db.Column(db.Integer, default=0) + failed_count = db.Column(db.Integer, default=0) + + # Timestamps + started_at = db.Column(db.DateTime, default=db.func.now()) + completed_at = db.Column(db.DateTime, nullable=True) + + +class ImportTask(db.Model): + """ + Tracks the state of each file being processed through the import pipeline. + Enables resume capability and parallel processing. + """ + __tablename__ = "import_task" + + id = db.Column(db.Integer, primary_key=True) + + # File identification + source_path = db.Column(db.String(1024), nullable=False, index=True) + file_hash = db.Column(db.String(64), nullable=True, index=True) # SHA256, computed during processing + file_size = db.Column(db.BigInteger, nullable=True) + + # Classification + # Values: 'scan', 'import_image', 'import_archive', 'thumbnail', 'sidecar' + task_type = db.Column(db.String(32), nullable=False, index=True) + + # State tracking + # Values: 'pending', 'queued', 'processing', 'complete', 'failed', 'skipped' + status = db.Column(db.String(32), nullable=False, default='pending', index=True) + + # Celery task tracking + celery_task_id = db.Column(db.String(64), nullable=True, index=True) + + # Context/metadata (JSON) + # For images: {"artist": "...", "dest_dir": "..."} + # For archives: {"artist": "...", "dest_dir": "...", "archive_tag": "..."} + # For sidecars: {"image_id": 123, "sidecar_path": "..."} + context = db.Column(db.JSON, nullable=True) + + # Batch relationship + batch_id = db.Column(db.Integer, db.ForeignKey('import_batch.id', ondelete='SET NULL'), nullable=True) + batch = db.relationship('ImportBatch', backref=db.backref('tasks', lazy='dynamic')) + + # Result tracking + result_image_id = db.Column(db.Integer, db.ForeignKey('image_record.id', ondelete='SET NULL'), nullable=True) + result_image = db.relationship('ImageRecord', foreign_keys=[result_image_id]) + error_message = db.Column(db.Text, nullable=True) + retry_count = db.Column(db.Integer, default=0) + + # Timestamps + created_at = db.Column(db.DateTime, default=db.func.now(), index=True) + started_at = db.Column(db.DateTime, nullable=True) + completed_at = db.Column(db.DateTime, nullable=True) + + # Unique constraint: prevent duplicate tasks for same file+type combination + # Note: file_hash can be NULL initially (computed during processing) + __table_args__ = ( + db.Index('ix_import_task_status_type', 'status', 'task_type'), + db.Index('ix_import_task_batch_status', 'batch_id', 'status'), + ) diff --git a/app/static/js/modal-pagination.js b/app/static/js/modal-pagination.js index 5ed081d..b013179 100644 --- a/app/static/js/modal-pagination.js +++ b/app/static/js/modal-pagination.js @@ -53,7 +53,9 @@ document.addEventListener('DOMContentLoaded', () => { archive: '🗜️', character: '👤', series: '📺', - rating: '⚠️' + rating: '⚠️', + source: '🌐', + post: '📌' }; return icons[kind] || '#'; } @@ -303,12 +305,17 @@ document.addEventListener('DOMContentLoaded', () => { modal.classList.add('active'); updateImage(index); history.replaceState({ modalIndex: index }, '', window.location.href); - // Focus the tag input after modal opens - if (tagInput) { + // Focus the tag input after modal opens (desktop only - avoids keyboard popup on mobile) + if (tagInput && !isTouchDevice()) { setTimeout(() => tagInput.focus(), 100); } } + // Detect touch devices to avoid auto-focus keyboard popup + function isTouchDevice() { + return ('ontouchstart' in window) || (navigator.maxTouchPoints > 0); + } + function closeModal() { modal.classList.remove('active'); modalImageWrapper.innerHTML = ''; diff --git a/app/static/js/showcase.js b/app/static/js/showcase.js index e3ebc87..23313e4 100644 --- a/app/static/js/showcase.js +++ b/app/static/js/showcase.js @@ -286,7 +286,9 @@ archive: '🗜️', character: '👤', series: '📺', - rating: '⚠️' + rating: '⚠️', + source: '🌐', + post: '📌' }; const icon = icons[tag.kind]; diff --git a/app/static/style.css b/app/static/style.css index 3ae27d3..d57930b 100644 --- a/app/static/style.css +++ b/app/static/style.css @@ -369,6 +369,28 @@ header { .modal-body { flex-direction: column; align-items: center; + height: 100%; + overflow-y: auto; + -webkit-overflow-scrolling: touch; + } +} + +/* Mobile landscape: prioritize image, minimize tag editor */ +@media (max-width: 900px) and (orientation: landscape) { + .modal-body { + flex-direction: row; + align-items: stretch; + } + .modal-image-wrapper { + max-width: calc(100% - 200px); + max-height: calc(100vh - 2rem); + height: calc(100vh - 2rem); + } + .tag-editor { + width: 180px; + min-width: 180px; + max-height: calc(100vh - 2rem); + overflow-y: auto; } } @@ -415,12 +437,17 @@ header { width: 90vw; height: auto; max-height: none; } -/* On smaller screens, full width for image wrapper */ -@media (max-width: 900px) { +/* On smaller screens (portrait), full width for image wrapper */ +@media (max-width: 900px) and (orientation: portrait) { .modal-image-wrapper { max-width: 100%; height: auto; - max-height: 60vh; + max-height: 55vh; + flex-shrink: 0; + } + .modal-image-wrapper img, + .modal-image-wrapper video { + max-height: 55vh; } } @@ -453,6 +480,17 @@ header { .modal-prev { left: 1rem; } .modal-next { right: 1rem; } +/* Mobile: always show nav buttons, larger touch targets */ +@media (max-width: 900px) { + .modal-button { + opacity: 0.8; + width: 44px; + height: 44px; + } + .modal-prev { left: 0.5rem; } + .modal-next { right: 0.5rem; } +} + /* Close button */ .modal-close-btn { position: absolute; @@ -535,13 +573,15 @@ header { flex-shrink: 0; } -/* On smaller screens, make tag editor full width */ -@media (max-width: 900px) { +/* On smaller screens (portrait), make tag editor full width and scrollable */ +@media (max-width: 900px) and (orientation: portrait) { .tag-editor { width: 100%; max-width: 500px; min-width: unset; - max-height: unset; + max-height: 35vh; + overflow-y: auto; + flex-shrink: 0; } } @@ -594,7 +634,7 @@ header { border-radius: 6px; background: rgba(255, 255, 255, 0.08); color: var(--text); - font-size: 0.9rem; + font-size: 16px; /* Must be 16px+ to prevent iOS auto-zoom on focus */ } .tag-form input::placeholder { color: var(--text-muted); @@ -714,8 +754,9 @@ header { .tag-thumb { flex: 1; - background-size: cover; - background-position: center; + min-width: 0; + height: 100%; + object-fit: cover; background-color: var(--bg-elevated); transition: filter 0.2s ease; } diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py new file mode 100644 index 0000000..8ed5a54 --- /dev/null +++ b/app/tasks/__init__.py @@ -0,0 +1,10 @@ +# app/tasks/__init__.py +""" +Celery task modules for the ImageRepo import pipeline. + +Task Types: +- scan: Directory scanning and task queuing +- import_file: Individual file import (images, videos, archives) +- thumbnail: Thumbnail generation +- sidecar: Gallery-DL metadata extraction and application +""" diff --git a/app/tasks/import_file.py b/app/tasks/import_file.py new file mode 100644 index 0000000..b9163b2 --- /dev/null +++ b/app/tasks/import_file.py @@ -0,0 +1,507 @@ +# app/tasks/import_file.py +""" +File import tasks for images, videos, and archives. + +These tasks handle the actual import of individual files, +including filtering, duplicate detection, and storage. +""" +import os +import shutil +import tempfile +import logging +from datetime import datetime + +import imagehash + +from app.celery_app import celery +from app import db +from app.models import ImportTask, ImageRecord, Tag, ArchiveRecord + +log = logging.getLogger('celery.tasks.import_file') + + +@celery.task(bind=True, name='app.tasks.import_file.import_media_file', + max_retries=3, default_retry_delay=60) +def import_media_file(self, task_id: int): + """ + Import a single media file (image or video). + + Steps: + 1. Load ImportTask record + 2. Apply filters (dimensions, transparency, single-color) + 3. Check for duplicates (hash and pHash) + 4. Copy to content-addressed storage + 5. Create ImageRecord + 6. Queue thumbnail task + 7. Queue sidecar metadata task + 8. Update ImportTask status + + Args: + task_id: ID of the ImportTask record + + Returns: + dict with status and result info + """ + from app.tasks.thumbnail import generate_thumbnail_task + from app.tasks.sidecar import apply_sidecar_metadata + from app.utils.image_importer import ( + extract_metadata, calculate_hash, calculate_perceptual_hash, + is_mostly_transparent, is_mostly_single_color, find_similar_image, + build_hashed_dest_path, load_import_settings, supersede_image, + generate_thumbnail, generate_video_thumbnail_mirrored + ) + from app.utils.metadata_enrichment import find_sidecar_json + + task = ImportTask.query.get(task_id) + if not task: + log.error(f"Task {task_id} not found") + return {'error': 'Task not found'} + + task.status = 'processing' + task.started_at = datetime.utcnow() + db.session.commit() + + try: + src_path = task.source_path + context = task.context or {} + artist = context.get('artist', 'unknown') + dest_dir = context.get('dest_dir', '/images') + + log.info(f"Processing: {src_path}") + + # Load settings + settings = load_import_settings() + + # Validate file exists + if not os.path.exists(src_path): + return _fail_task(task, 'Source file not found') + + filename = os.path.basename(src_path) + lower = filename.lower() + + # Extract metadata first for filtering + metadata = extract_metadata(src_path) + if not metadata.get('width') or not metadata.get('height'): + return _skip_task(task, 'Missing dimension data') + + # Apply dimension filter + min_width = settings.get('min_width', 0) + min_height = settings.get('min_height', 0) + if min_width > 0 and metadata['width'] < min_width: + return _skip_task(task, f"Too small: {metadata['width']}x{metadata['height']}") + if min_height > 0 and metadata['height'] < min_height: + return _skip_task(task, f"Too small: {metadata['width']}x{metadata['height']}") + + # Apply transparency filter + if settings.get('skip_transparent') and lower.endswith(('.png', '.gif', '.webp')): + threshold = settings.get('transparency_threshold', 0.9) + if is_mostly_transparent(src_path, threshold): + return _skip_task(task, 'Mostly transparent') + + # Apply single-color filter + if settings.get('skip_single_color', True) and lower.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp')): + if is_mostly_single_color(src_path, + settings.get('single_color_threshold', 0.95), + settings.get('single_color_tolerance', 30)): + return _skip_task(task, 'Mostly single color') + + # Compute content hash + file_hash = calculate_hash(src_path) + task.file_hash = file_hash + + # Check for exact duplicate by hash + existing = ImageRecord.query.filter_by(hash=file_hash).first() + if existing: + # Still try to enrich with sidecar metadata + sidecar_path = find_sidecar_json(src_path) + if sidecar_path: + apply_sidecar_metadata.delay(existing.id, sidecar_path) + return _skip_task(task, 'Duplicate by hash', result_image_id=existing.id) + + # Compute pHash for similarity check + phash = calculate_perceptual_hash(src_path) + phash_threshold = settings.get('phash_threshold', 10) + + if phash and phash_threshold > 0: + # Load existing pHashes for comparison + existing_phashes = [ + (imagehash.hex_to_hash(img.perceptual_hash), img.width, img.height, img.id) + for img in ImageRecord.query.filter(ImageRecord.perceptual_hash.isnot(None)).all() + ] + + relationship, similar_id = find_similar_image( + phash, metadata['width'], metadata['height'], + existing_phashes, threshold=phash_threshold + ) + + if relationship == 'larger_exists': + return _skip_task(task, 'Similar larger image exists') + + # Handle supersede case - new image is larger than existing + if relationship == 'smaller_exists' and settings.get('supersede_smaller', True) and similar_id: + old_record = ImageRecord.query.get(similar_id) + if old_record: + return _supersede_existing( + task, old_record, src_path, file_hash, phash, + metadata, artist, dest_dir, filename + ) + + # Normal import - copy to destination + target_dir = os.path.join(dest_dir, artist) + os.makedirs(target_dir, exist_ok=True) + dest_path = build_hashed_dest_path(target_dir, filename, file_hash) + shutil.copy2(src_path, dest_path) + + # Generate thumbnail + try: + if dest_path.lower().endswith(('.mp4', '.mov', '.webm', '.avi', '.mkv')): + thumb_path = generate_video_thumbnail_mirrored(dest_path) + else: + thumb_path = generate_thumbnail(dest_path) + except Exception as e: + log.warning(f"Thumbnail generation failed for {filename}: {e}") + thumb_path = None + + # Create ImageRecord + record = ImageRecord( + filename=filename, + filepath=dest_path, + thumb_path=thumb_path, + hash=file_hash, + perceptual_hash=str(phash) if phash else None, + file_size=metadata['file_size'], + width=metadata['width'], + height=metadata['height'], + format=metadata['format'], + camera_model=metadata.get('camera_model'), + taken_at=metadata.get('taken_at'), + imported_at=datetime.utcnow() + ) + + # Add artist tag + _add_artist_tag(record, artist) + + db.session.add(record) + db.session.commit() + + # Complete task + task.status = 'complete' + task.result_image_id = record.id + task.completed_at = datetime.utcnow() + db.session.commit() + + # Queue sidecar metadata enrichment + sidecar_path = find_sidecar_json(src_path) + if sidecar_path: + apply_sidecar_metadata.delay(record.id, sidecar_path) + + # Update batch stats + if task.batch_id: + from app.tasks.scan import update_batch_stats + update_batch_stats.delay(task.batch_id) + + log.info(f"Imported: {filename} -> {dest_path}") + + return { + 'status': 'imported', + 'image_id': record.id, + 'dest_path': dest_path + } + + except Exception as e: + db.session.rollback() + log.error(f"Import failed for task {task_id}: {e}", exc_info=True) + + task.retry_count += 1 + if task.retry_count < 3: + task.status = 'queued' + db.session.commit() + raise self.retry(exc=e) + else: + return _fail_task(task, str(e)) + + +@celery.task(bind=True, name='app.tasks.import_file.import_archive', + max_retries=2, default_retry_delay=120, + soft_time_limit=1800, time_limit=3600) # 30min soft, 60min hard for large archives +def import_archive(self, task_id: int): + """ + Import an archive file. + + Steps: + 1. Check if archive already processed (by hash) + 2. Extract to temp directory + 3. Create ImportTask for each media file inside + 4. Queue import tasks for extracted files + 5. Create archive tag if files were found + 6. Clean up is deferred until child tasks complete + + Args: + task_id: ID of the ImportTask record + + Returns: + dict with status and counts + """ + from app.utils.image_importer import calculate_hash, compute_next_archive_tag_name + from pyunpack import Archive + + task = ImportTask.query.get(task_id) + if not task: + return {'error': 'Task not found'} + + task.status = 'processing' + task.started_at = datetime.utcnow() + db.session.commit() + + tmpdir = None + try: + archive_path = task.source_path + context = task.context or {} + artist = context.get('artist', 'unknown') + dest_dir = context.get('dest_dir', '/images') + + log.info(f"Processing archive: {archive_path}") + + # Compute archive hash + archive_hash = calculate_hash(archive_path) + task.file_hash = archive_hash + + # Check if already processed + existing = ArchiveRecord.query.filter_by(hash=archive_hash).first() + if existing: + return _skip_task(task, 'Archive already imported') + + # Get archive size + archive_size = os.path.getsize(archive_path) + + # Extract archive + tmpdir = tempfile.mkdtemp(prefix='extract_') + log.info(f"Extracting to: {tmpdir}") + + try: + Archive(archive_path).extractall(tmpdir) + except Exception as e: + log.error(f"Extraction failed: {e}") + if tmpdir and os.path.exists(tmpdir): + shutil.rmtree(tmpdir, ignore_errors=True) + return _fail_task(task, f'Extraction failed: {e}') + + # Find media files in extracted content + media_files = [] + for root, _, files in os.walk(tmpdir): + for filename in files: + lower = filename.lower() + if lower.endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', + '.mp4', '.mov', '.webm', '.avi', '.mkv')): + media_files.append(os.path.join(root, filename)) + + if not media_files: + log.info(f"No media files found in archive: {archive_path}") + if tmpdir and os.path.exists(tmpdir): + shutil.rmtree(tmpdir, ignore_errors=True) + return _skip_task(task, 'No media files in archive') + + log.info(f"Found {len(media_files)} media files in archive") + + # Create archive tag + tag_name = compute_next_archive_tag_name(artist) + archive_tag = Tag.query.filter_by(name=tag_name).first() + if not archive_tag: + archive_tag = Tag(name=tag_name, kind='archive') + db.session.add(archive_tag) + db.session.flush() + + # Create ArchiveRecord + archive_record = ArchiveRecord( + filename=archive_path, + file_size=archive_size, + hash=archive_hash, + artist=artist, + tag_id=archive_tag.id + ) + db.session.add(archive_record) + db.session.commit() + + # Queue tasks for each media file + queued_count = 0 + for media_path in media_files: + child_task = ImportTask( + source_path=media_path, + task_type='import_image', + batch_id=task.batch_id, + context={ + 'artist': artist, + 'dest_dir': dest_dir, + 'archive_task_id': task.id, + 'archive_tag_id': archive_tag.id, + 'temp_dir': tmpdir + }, + status='pending' + ) + db.session.add(child_task) + db.session.flush() + + result = import_media_file.delay(child_task.id) + child_task.celery_task_id = result.id + child_task.status = 'queued' + queued_count += 1 + + # Store child task info for cleanup tracking + task.context = task.context or {} + task.context['temp_dir'] = tmpdir + task.context['archive_tag_id'] = archive_tag.id + task.context['child_count'] = queued_count + + task.status = 'complete' + task.completed_at = datetime.utcnow() + db.session.commit() + + log.info(f"Archive processed: queued {queued_count} files from {archive_path}") + + # Note: temp directory cleanup happens via a separate periodic task + # or when all child tasks complete + + return { + 'status': 'extracted', + 'queued_files': queued_count, + 'archive_tag': tag_name + } + + except Exception as e: + db.session.rollback() + log.error(f"Archive import failed: {e}", exc_info=True) + if tmpdir and os.path.exists(tmpdir): + shutil.rmtree(tmpdir, ignore_errors=True) + return _fail_task(task, str(e)) + + +def _fail_task(task: ImportTask, error: str, **kwargs) -> dict: + """Mark task as failed with error message.""" + task.status = 'failed' + task.error_message = error + task.completed_at = datetime.utcnow() + for key, value in kwargs.items(): + setattr(task, key, value) + db.session.commit() + + # Update batch stats + if task.batch_id: + from app.tasks.scan import update_batch_stats + update_batch_stats.delay(task.batch_id) + + log.error(f"Task {task.id} failed: {error}") + return {'status': 'failed', 'error': error} + + +def _skip_task(task: ImportTask, reason: str, **kwargs) -> dict: + """Mark task as skipped with reason.""" + task.status = 'skipped' + task.error_message = reason + task.completed_at = datetime.utcnow() + for key, value in kwargs.items(): + setattr(task, key, value) + db.session.commit() + + # Update batch stats + if task.batch_id: + from app.tasks.scan import update_batch_stats + update_batch_stats.delay(task.batch_id) + + log.info(f"Task {task.id} skipped: {reason}") + return {'status': 'skipped', 'reason': reason} + + +def _add_artist_tag(record: ImageRecord, artist: str): + """Add artist tag to image record.""" + if not artist: + return + + tag_name = f"artist:{artist}" + tag = Tag.query.filter_by(name=tag_name).first() + if not tag: + tag = Tag(name=tag_name, kind='artist') + db.session.add(tag) + db.session.flush() + + if tag not in record.tags: + record.tags.append(tag) + + +def _add_archive_tag(record: ImageRecord, archive_tag_id: int): + """Add archive tag to image record.""" + if not archive_tag_id: + return + + tag = Tag.query.get(archive_tag_id) + if tag and tag not in record.tags: + record.tags.append(tag) + + +def _supersede_existing(task: ImportTask, old_record: ImageRecord, src_path: str, + file_hash: str, phash, metadata: dict, artist: str, + dest_dir: str, filename: str) -> dict: + """ + Replace an existing smaller image with a larger version. + + Preserves tags and metadata from the old record. + """ + from app.utils.image_importer import ( + build_hashed_dest_path, generate_thumbnail, generate_video_thumbnail_mirrored, + supersede_image + ) + from app.utils.metadata_enrichment import find_sidecar_json, enrich_from_sidecar + from app.tasks.sidecar import apply_sidecar_metadata + + log.info(f"Superseding {old_record.filename} ({old_record.width}x{old_record.height}) " + f"with {filename} ({metadata['width']}x{metadata['height']})") + + # Enrich metadata from sidecar if available + sidecar_path = find_sidecar_json(src_path) + if sidecar_path: + enriched = enrich_from_sidecar(sidecar_path, {'taken_at': metadata.get('taken_at')}) + if enriched.get('taken_at'): + metadata['taken_at'] = enriched['taken_at'] + + # Copy new file to destination + target_dir = os.path.join(dest_dir, artist) + os.makedirs(target_dir, exist_ok=True) + dest_path = build_hashed_dest_path(target_dir, filename, file_hash) + shutil.copy2(src_path, dest_path) + + # Generate new thumbnail + try: + if dest_path.lower().endswith(('.mp4', '.mov', '.webm', '.avi', '.mkv')): + thumb_path = generate_video_thumbnail_mirrored(dest_path) + else: + thumb_path = generate_thumbnail(dest_path) + except Exception as e: + log.warning(f"Thumbnail generation failed: {e}") + thumb_path = None + + # Supersede the old record + record = supersede_image( + old_record, dest_path, thumb_path, file_hash, phash, metadata, filename + ) + + db.session.commit() + + # Complete task + task.status = 'complete' + task.result_image_id = record.id + task.completed_at = datetime.utcnow() + db.session.commit() + + # Apply sidecar metadata if available + if sidecar_path: + apply_sidecar_metadata.delay(record.id, sidecar_path) + + # Update batch stats + if task.batch_id: + from app.tasks.scan import update_batch_stats + update_batch_stats.delay(task.batch_id) + + return { + 'status': 'superseded', + 'image_id': record.id, + 'dest_path': dest_path + } diff --git a/app/tasks/scan.py b/app/tasks/scan.py new file mode 100644 index 0000000..f38e673 --- /dev/null +++ b/app/tasks/scan.py @@ -0,0 +1,349 @@ +# app/tasks/scan.py +""" +Directory scanning task for the import pipeline. + +This task walks the import directory, detects new/changed files, +creates ImportTask records, and queues appropriate processing tasks. +""" +import os +import logging +from datetime import datetime + +from app.celery_app import celery +from app import db +from app.models import ImportTask, ImportBatch + +log = logging.getLogger('celery.tasks.scan') + +# Supported file extensions (imported from image_importer for consistency) +ALLOWED_MEDIA_EXTS = ( + '.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tiff', '.tif', + '.mp4', '.mov', '.webm', '.avi', '.mkv' +) +ALLOWED_ARCHIVE_EXTS = ( + '.zip', '.rar', '.7z', '.tar', '.tar.gz', '.tgz', '.tar.bz2', '.tbz2', + '.cbr', '.cbz' +) + + +@celery.task(bind=True, name='app.tasks.scan.scan_directory') +def scan_directory(self, source_dir: str = '/import', dest_dir: str = '/images'): + """ + Scan source directory for new/changed files. + + Creates ImportTask records for each file and queues them for processing. + + Strategy: + 1. Walk directory tree under each artist folder + 2. For each file, check if already in ImportTask with same path+size + 3. If new/changed, create ImportTask and queue appropriate task + 4. Handle archives specially (first volume only, queue as atomic unit) + + Args: + source_dir: Root directory to scan (default: /import) + dest_dir: Destination for imported files (default: /images) + + Returns: + dict with batch_id and statistics + """ + from app.tasks.import_file import import_media_file, import_archive + from app.utils.image_importer import is_first_volume + + log.info(f"Starting directory scan: {source_dir}") + + # Create batch record + batch = ImportBatch(source_directory=source_dir) + db.session.add(batch) + db.session.commit() + + stats = { + 'total_files': 0, + 'queued_media': 0, + 'queued_archives': 0, + 'skipped_existing': 0, + 'skipped_non_first_volume': 0, + 'skipped_non_media': 0, + } + + try: + # Iterate through artist directories + for artist_dir in os.listdir(source_dir): + artist_path = os.path.join(source_dir, artist_dir) + if not os.path.isdir(artist_path): + continue + + log.debug(f"Scanning artist directory: {artist_dir}") + + # Walk all files in artist directory (including subdirectories) + for root, _, files in os.walk(artist_path): + for filename in files: + full_path = os.path.join(root, filename) + lower = filename.lower() + stats['total_files'] += 1 + + # Skip sidecar JSON files (handled during import) + if lower.endswith('.json'): + continue + + # Check if archive + if lower.endswith(ALLOWED_ARCHIVE_EXTS): + # Only process first volume of multi-part archives + if not is_first_volume(full_path): + stats['skipped_non_first_volume'] += 1 + continue + + # Create and queue archive task + task_record = _create_task_if_new( + source_path=full_path, + task_type='import_archive', + batch_id=batch.id, + context={'artist': artist_dir, 'dest_dir': dest_dir} + ) + + if task_record: + result = import_archive.delay(task_record.id) + task_record.celery_task_id = result.id + task_record.status = 'queued' + db.session.commit() + stats['queued_archives'] += 1 + log.debug(f"Queued archive: {filename}") + else: + stats['skipped_existing'] += 1 + continue + + # Check if media file + if not lower.endswith(ALLOWED_MEDIA_EXTS): + stats['skipped_non_media'] += 1 + continue + + # Create and queue media file task + task_record = _create_task_if_new( + source_path=full_path, + task_type='import_image', + batch_id=batch.id, + context={'artist': artist_dir, 'dest_dir': dest_dir} + ) + + if task_record: + result = import_media_file.delay(task_record.id) + task_record.celery_task_id = result.id + task_record.status = 'queued' + db.session.commit() + stats['queued_media'] += 1 + log.debug(f"Queued media: {filename}") + else: + stats['skipped_existing'] += 1 + + # Update batch with totals + batch.total_files = stats['total_files'] + db.session.commit() + + log.info(f"Scan complete. Queued {stats['queued_media']} media, " + f"{stats['queued_archives']} archives. " + f"Skipped {stats['skipped_existing']} existing.") + + return { + 'batch_id': batch.id, + 'stats': stats + } + + except Exception as e: + log.error(f"Scan failed: {e}", exc_info=True) + batch.status = 'failed' + db.session.commit() + raise + + +def _create_task_if_new(source_path: str, task_type: str, batch_id: int, context: dict) -> ImportTask: + """ + Create ImportTask if no existing task for this file. + + Uses file path and size for change detection. + Hash is computed later during actual processing. + + Args: + source_path: Full path to source file + task_type: Type of task ('import_image', 'import_archive', etc.) + batch_id: ID of the current ImportBatch + context: Context dict with artist, dest_dir, etc. + + Returns: + New ImportTask record if created, None if already exists + """ + try: + file_size = os.path.getsize(source_path) + except OSError as e: + log.warning(f"Cannot access file {source_path}: {e}") + return None + + # Check for existing pending/queued/processing task for same file + existing_active = ImportTask.query.filter_by( + source_path=source_path, + task_type=task_type + ).filter( + ImportTask.status.in_(['pending', 'queued', 'processing']) + ).first() + + if existing_active: + log.debug(f"Task already active for: {source_path}") + return None + + # Check for completed task with same file path and size + # If size changed, we should re-import + completed_same = ImportTask.query.filter_by( + source_path=source_path, + task_type=task_type, + file_size=file_size, + status='complete' + ).first() + + if completed_same: + log.debug(f"Already imported (same size): {source_path}") + return None + + # Also check 'skipped' status - don't re-queue skipped files unless size changed + skipped_same = ImportTask.query.filter_by( + source_path=source_path, + task_type=task_type, + file_size=file_size, + status='skipped' + ).first() + + if skipped_same: + log.debug(f"Previously skipped (same size): {source_path}") + return None + + # Create new task + task = ImportTask( + source_path=source_path, + task_type=task_type, + file_size=file_size, + batch_id=batch_id, + context=context, + status='pending' + ) + db.session.add(task) + db.session.commit() + + return task + + +@celery.task(name='app.tasks.scan.recover_interrupted_tasks') +def recover_interrupted_tasks(): + """ + Find and re-queue tasks that were interrupted. + + Called periodically by Celery Beat to recover from worker crashes. + Tasks with status 'processing' or 'queued' that have no active + Celery task are re-queued. + + Returns: + dict with count of recovered tasks + """ + from app.tasks.import_file import import_media_file, import_archive + from app.tasks.thumbnail import generate_thumbnail_task + + log.info("Checking for interrupted tasks...") + + # Find tasks that are stuck in processing/queued state + # We look for tasks that have been in this state for more than 10 minutes + cutoff = datetime.utcnow() + + interrupted = ImportTask.query.filter( + ImportTask.status.in_(['processing', 'queued']) + ).all() + + recovered = 0 + for task in interrupted: + # Check if Celery task is still active + if task.celery_task_id: + from app.celery_app import celery as celery_app + result = celery_app.AsyncResult(task.celery_task_id) + # If task is still pending/started, don't re-queue + if result.state in ('PENDING', 'STARTED', 'RETRY'): + continue + + log.info(f"Recovering task {task.id}: {task.source_path}") + + # Reset task state + task.status = 'pending' + task.celery_task_id = None + task.started_at = None + task.retry_count += 1 + + # Re-queue based on type + if task.task_type == 'import_image': + result = import_media_file.delay(task.id) + elif task.task_type == 'import_archive': + result = import_archive.delay(task.id) + elif task.task_type == 'thumbnail': + # Thumbnail tasks store image_id in context + image_id = task.context.get('image_id') if task.context else None + if image_id: + result = generate_thumbnail_task.delay(image_id) + else: + task.status = 'failed' + task.error_message = 'Missing image_id in context' + db.session.commit() + continue + else: + log.warning(f"Unknown task type: {task.task_type}") + continue + + task.celery_task_id = result.id + task.status = 'queued' + recovered += 1 + + db.session.commit() + + if recovered > 0: + log.info(f"Recovered {recovered} interrupted tasks") + + return {'recovered': recovered} + + +@celery.task(name='app.tasks.scan.update_batch_stats') +def update_batch_stats(batch_id: int): + """ + Update statistics for an ImportBatch based on its tasks. + + Args: + batch_id: ID of the batch to update + """ + batch = ImportBatch.query.get(batch_id) + if not batch: + return + + from sqlalchemy import func + + # Count tasks by status + status_counts = dict( + db.session.query( + ImportTask.status, + func.count(ImportTask.id) + ).filter( + ImportTask.batch_id == batch_id + ).group_by(ImportTask.status).all() + ) + + batch.processed_files = sum( + status_counts.get(s, 0) + for s in ['complete', 'skipped', 'failed'] + ) + batch.imported_count = status_counts.get('complete', 0) + batch.skipped_count = status_counts.get('skipped', 0) + batch.failed_count = status_counts.get('failed', 0) + + # Check if batch is complete + pending_or_active = sum( + status_counts.get(s, 0) + for s in ['pending', 'queued', 'processing'] + ) + + if pending_or_active == 0 and batch.status == 'running': + batch.status = 'complete' + batch.completed_at = datetime.utcnow() + log.info(f"Batch {batch_id} complete: {batch.imported_count} imported, " + f"{batch.skipped_count} skipped, {batch.failed_count} failed") + + db.session.commit() diff --git a/app/tasks/sidecar.py b/app/tasks/sidecar.py new file mode 100644 index 0000000..0e31331 --- /dev/null +++ b/app/tasks/sidecar.py @@ -0,0 +1,181 @@ +# app/tasks/sidecar.py +""" +Sidecar metadata extraction and application tasks. + +Handles finding and applying metadata from Gallery-DL sidecar JSON files +to imported images. +""" +import logging +from datetime import datetime + +from app.celery_app import celery +from app import db +from app.models import ImageRecord, Tag + +log = logging.getLogger('celery.tasks.sidecar') + + +@celery.task(bind=True, name='app.tasks.sidecar.apply_sidecar_metadata', + max_retries=2, default_retry_delay=30) +def apply_sidecar_metadata(self, image_id: int, sidecar_path: str): + """ + Apply metadata from a sidecar JSON file to an image record. + + Extracts platform info, artist, post ID, dates, and generates + appropriate tags (source:platform, post:platform:artist:id). + + Args: + image_id: ID of the ImageRecord to update + sidecar_path: Path to the sidecar JSON file + + Returns: + dict with status and applied metadata info + """ + from app.utils.metadata_enrichment import ( + enrich_from_sidecar, resolve_date_conflict, load_sidecar_metadata + ) + + record = ImageRecord.query.get(image_id) + if not record: + log.warning(f"Image {image_id} not found") + return {'error': 'Image not found'} + + try: + # Load sidecar data + sidecar_data = load_sidecar_metadata(sidecar_path) + if not sidecar_data: + return {'status': 'no_metadata', 'reason': 'Failed to load sidecar'} + + # Get enriched metadata + enriched = enrich_from_sidecar(sidecar_path, {'taken_at': record.taken_at}) + + if not enriched.get('raw_metadata'): + return {'status': 'no_metadata', 'reason': 'No metadata extracted'} + + tags_added = 0 + + # Update date if earlier + new_date = resolve_date_conflict(record.taken_at, enriched.get('taken_at')) + if new_date != record.taken_at: + record.taken_at = new_date + log.debug(f"Updated date for {image_id}: {new_date}") + + # Apply tags from metadata + for tag_info in enriched.get('tags', []): + tag_name = tag_info.get('name') + tag_kind = tag_info.get('kind') + if not tag_name: + continue + + tag = Tag.query.filter_by(name=tag_name).first() + if not tag: + tag = Tag(name=tag_name, kind=tag_kind) + db.session.add(tag) + db.session.flush() + log.debug(f"Created tag: {tag_name} ({tag_kind})") + + if tag not in record.tags: + record.tags.append(tag) + tags_added += 1 + + db.session.commit() + + log.info(f"Applied sidecar metadata to {image_id}: " + f"platform={enriched.get('platform')}, tags_added={tags_added}") + + return { + 'status': 'applied', + 'platform': enriched.get('platform'), + 'artist': enriched.get('artist'), + 'post_id': enriched.get('post_id'), + 'tags_added': tags_added + } + + except Exception as e: + db.session.rollback() + log.error(f"Sidecar application failed for {image_id}: {e}", exc_info=True) + raise self.retry(exc=e) + + +@celery.task(name='app.tasks.sidecar.scan_and_match_sidecars') +def scan_and_match_sidecars(batch_id: int = None, limit: int = 1000): + """ + Scan for sidecar files and match them to imported images. + + Looks at recently imported images and tries to find corresponding + sidecar JSON files for metadata enrichment. + + Args: + batch_id: Optional batch ID to limit scope + limit: Maximum number of images to process + + Returns: + dict with count of matched sidecars + """ + from app.utils.metadata_enrichment import find_sidecar_json + + log.info(f"Scanning for unmatched sidecars (batch={batch_id}, limit={limit})") + + # Build query for images that might need sidecar matching + query = ImageRecord.query.filter( + ImageRecord.filepath.isnot(None) + ).order_by(ImageRecord.imported_at.desc()) + + if limit: + query = query.limit(limit) + + matched = 0 + already_has_source = 0 + + for record in query.all(): + # Check if already has source tag (indicates sidecar was processed) + has_source_tag = any( + tag.kind == 'source' or tag.kind == 'post' + for tag in record.tags + ) + if has_source_tag: + already_has_source += 1 + continue + + # Try to find sidecar + sidecar_path = find_sidecar_json(record.filepath) + if sidecar_path: + apply_sidecar_metadata.delay(record.id, sidecar_path) + matched += 1 + + log.info(f"Sidecar scan complete: matched={matched}, already_tagged={already_has_source}") + return {'matched': matched, 'already_tagged': already_has_source} + + +@celery.task(name='app.tasks.sidecar.reprocess_sidecars_for_tag') +def reprocess_sidecars_for_tag(tag_name: str): + """ + Reprocess sidecar metadata for all images with a specific tag. + + Useful for re-running sidecar extraction after updates to + the metadata extraction logic. + + Args: + tag_name: Name of the tag to filter by + + Returns: + dict with count of queued tasks + """ + from app.utils.metadata_enrichment import find_sidecar_json + + tag = Tag.query.filter_by(name=tag_name).first() + if not tag: + return {'error': f'Tag not found: {tag_name}'} + + queued = 0 + for record in tag.images: + if not record.filepath: + continue + + sidecar_path = find_sidecar_json(record.filepath) + if sidecar_path: + apply_sidecar_metadata.delay(record.id, sidecar_path) + queued += 1 + + log.info(f"Queued {queued} sidecar reprocessing tasks for tag: {tag_name}") + return {'queued': queued, 'tag': tag_name} diff --git a/app/tasks/thumbnail.py b/app/tasks/thumbnail.py new file mode 100644 index 0000000..fdb0ecb --- /dev/null +++ b/app/tasks/thumbnail.py @@ -0,0 +1,153 @@ +# app/tasks/thumbnail.py +""" +Thumbnail generation tasks. + +Handles generating thumbnails for images and videos, +both individually and in bulk. +""" +import logging +from datetime import datetime + +from app.celery_app import celery +from app import db +from app.models import ImageRecord, ImportTask + +log = logging.getLogger('celery.tasks.thumbnail') + + +@celery.task(bind=True, name='app.tasks.thumbnail.generate_thumbnail_task', + max_retries=2, default_retry_delay=30) +def generate_thumbnail_task(self, image_id: int, overwrite: bool = False): + """ + Generate thumbnail for a single image/video. + + Args: + image_id: ID of the ImageRecord + overwrite: Whether to overwrite existing thumbnail + + Returns: + dict with status and thumbnail path + """ + from app.utils.image_importer import generate_thumbnail, generate_video_thumbnail_mirrored + + record = ImageRecord.query.get(image_id) + if not record: + log.warning(f"Image {image_id} not found") + return {'error': 'Image not found'} + + try: + filepath = record.filepath + if not filepath: + return {'error': 'No filepath'} + + # Check if thumbnail already exists and we're not overwriting + if record.thumb_path and not overwrite: + import os + if os.path.exists(record.thumb_path): + return {'status': 'exists', 'thumb_path': record.thumb_path} + + log.debug(f"Generating thumbnail for: {filepath}") + + is_video = filepath.lower().endswith(('.mp4', '.mov', '.webm', '.avi', '.mkv')) + + if is_video: + thumb_path = generate_video_thumbnail_mirrored(filepath) + else: + thumb_path = generate_thumbnail(filepath, overwrite=overwrite) + + if thumb_path: + record.thumb_path = thumb_path + db.session.commit() + log.info(f"Generated thumbnail: {thumb_path}") + return {'status': 'success', 'thumb_path': thumb_path} + else: + return {'status': 'failed', 'error': 'Thumbnail generation returned None'} + + except Exception as e: + db.session.rollback() + log.error(f"Thumbnail generation failed for {image_id}: {e}", exc_info=True) + raise self.retry(exc=e) + + +@celery.task(name='app.tasks.thumbnail.regenerate_all_thumbnails') +def regenerate_all_thumbnails(batch_size: int = 100, overwrite: bool = True): + """ + Queue thumbnail regeneration for all images. + + Processes images in batches to avoid memory issues. + + Args: + batch_size: Number of images to process per batch + overwrite: Whether to overwrite existing thumbnails + + Returns: + dict with count of queued tasks + """ + log.info("Starting bulk thumbnail regeneration") + + count = 0 + last_id = 0 + + while True: + records = ImageRecord.query.filter( + ImageRecord.id > last_id, + ImageRecord.filepath.isnot(None) + ).order_by(ImageRecord.id.asc()).limit(batch_size).all() + + if not records: + break + + for record in records: + generate_thumbnail_task.delay(record.id, overwrite=overwrite) + count += 1 + + last_id = records[-1].id + log.info(f"Queued {count} thumbnail tasks so far...") + + log.info(f"Thumbnail regeneration complete: queued {count} tasks") + return {'queued': count} + + +@celery.task(name='app.tasks.thumbnail.regenerate_missing_thumbnails') +def regenerate_missing_thumbnails(batch_size: int = 100): + """ + Generate thumbnails only for images that don't have them. + + Args: + batch_size: Number of images to process per batch + + Returns: + dict with count of queued tasks + """ + import os + + log.info("Checking for missing thumbnails") + + count = 0 + last_id = 0 + + while True: + records = ImageRecord.query.filter( + ImageRecord.id > last_id, + ImageRecord.filepath.isnot(None) + ).order_by(ImageRecord.id.asc()).limit(batch_size).all() + + if not records: + break + + for record in records: + # Check if thumbnail is missing + needs_thumb = False + if not record.thumb_path: + needs_thumb = True + elif not os.path.exists(record.thumb_path): + needs_thumb = True + + if needs_thumb: + generate_thumbnail_task.delay(record.id, overwrite=False) + count += 1 + + last_id = records[-1].id + + log.info(f"Missing thumbnail check complete: queued {count} tasks") + return {'queued': count} diff --git a/app/templates/settings.html b/app/templates/settings.html index 16f27de..16bff44 100644 --- a/app/templates/settings.html +++ b/app/templates/settings.html @@ -206,7 +206,140 @@ + +
+
+

Import Queue Status

+

Monitor and control the Celery-based import queue system.

+ +
+
+
+ - + Pending +
+
+ - + Queued +
+
+ - + Processing +
+
+ - + Complete +
+
+ - + Skipped +
+
+ - + Failed +
+
+ + + +
+ + + + + Open Flower Dashboard +
+ +
+

+ Workers: - | + Active Tasks: - +

+
+
+ + +
+

Recent Tasks

+
+ + + + + + + + + + + + + +
FileTypeStatusErrorTime
Loading...
+
+
+
+
+