# 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. Supports two modes: - Quick import (default): Fast duplicate detection by content hash only. Skips pHash comparison for speed. - Deep import (via deep_scan): Full reprocessing with pHash comparison, optional thumbnail regeneration, and sidecar re-application. """ import os import shutil import tempfile import logging from datetime import datetime from functools import lru_cache 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') # Cache import settings to avoid repeated DB/file reads _settings_cache = None _settings_cache_time = None SETTINGS_CACHE_TTL = 300 # 5 minutes def _get_cached_settings() -> dict: """ Get import settings with caching. Caches for 5 minutes to avoid repeated DB/file reads during batch imports. """ global _settings_cache, _settings_cache_time from app.utils.image_importer import load_import_settings now = datetime.utcnow() if _settings_cache is None or _settings_cache_time is None: _settings_cache = load_import_settings() _settings_cache_time = now elif (now - _settings_cache_time).total_seconds() > SETTINGS_CACHE_TTL: _settings_cache = load_import_settings() _settings_cache_time = now return _settings_cache def _find_best_sidecar(src_path: str, archive_path: str = None, archive_sidecar_path: str = None) -> str | None: """ Find the best sidecar JSON for a file, checking multiple locations. For files extracted from archives, checks: 1. Per-file sidecar next to extracted file (temp dir) 2. Per-file sidecar next to original archive (by filename match) 3. Archive's sidecar as fallback (applies to all files in archive) Args: src_path: Path to the source file archive_path: Path to original archive (if extracted from archive) archive_sidecar_path: Path to archive's sidecar JSON Returns: Path to the best sidecar JSON found, or None """ from app.utils.metadata_enrichment import find_sidecar_json import re # First try: per-file sidecar next to the source file sidecar = find_sidecar_json(src_path) if sidecar: return sidecar # Second try: per-file sidecar next to original archive if archive_path: from pathlib import Path archive_dir = Path(archive_path).parent filename = Path(src_path).name stem = Path(src_path).stem # Check for exact stem match per_file_json = archive_dir / f"{stem}.json" if per_file_json.exists(): return str(per_file_json) # Check for stem without numeric prefix stem_no_prefix = re.sub(r'^\d+_', '', stem) if stem_no_prefix != stem: alt_json = archive_dir / f"{stem_no_prefix}.json" if alt_json.exists(): return str(alt_json) # Third try: archive-level sidecar (fallback for all files) if archive_sidecar_path and os.path.exists(archive_sidecar_path): return archive_sidecar_path return None @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). Supports two modes based on context.deep_scan flag: - Quick mode (default): Fast duplicate detection by content hash only. Skips pHash comparison for speed. - Deep mode: Full reprocessing with pHash comparison, optional thumbnail regeneration, and sidecar re-application. Steps: 1. Load ImportTask record 2. Apply filters (dimensions, transparency, single-color) 3. Check for duplicates (hash, and pHash in deep mode) 4. Copy to content-addressed storage 5. Create ImageRecord 6. Generate thumbnail 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, supersede_image, generate_thumbnail, generate_video_thumbnail_mirrored, transcode_video_to_mp4, needs_transcode ) 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') # Deep scan options is_deep_scan = context.get('deep_scan', False) regenerate_thumbnails = context.get('regenerate_thumbnails', False) reapply_sidecars = context.get('reapply_sidecars', False) verify_hashes = context.get('verify_hashes', False) # Archive sidecar (for files extracted from archives) archive_sidecar_path = context.get('archive_sidecar_path') archive_path = context.get('archive_path') log.info(f"Processing: {src_path} (deep_scan={is_deep_scan})") # Load settings with caching (avoids repeated DB reads) settings = _get_cached_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: # Merge archive tag if this file came from an archive # This ensures the existing image gets tagged with all archives it appears in archive_tag_id = context.get('archive_tag_id') if archive_tag_id: _add_archive_tag(existing, archive_tag_id) db.session.commit() log.info(f"Added archive tag {archive_tag_id} to existing image {existing.id}") # In deep scan mode with reapply_sidecars, always try to enrich # In quick mode, also try if sidecar exists # Uses archive sidecar as fallback for files from archives sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path) if sidecar_path and (is_deep_scan and reapply_sidecars or not is_deep_scan): apply_sidecar_metadata.delay(existing.id, sidecar_path) # In deep scan with regenerate_thumbnails, regenerate even for duplicates if is_deep_scan and regenerate_thumbnails: _regenerate_thumbnail(existing) return _skip_task(task, 'Duplicate by hash', result_image_id=existing.id) # pHash similarity check # Quick mode: SKIP pHash comparison (relies on content hash for duplicates) # Deep mode: Full pHash comparison for similarity detection phash = None if is_deep_scan: phash = calculate_perceptual_hash(src_path) phash_threshold = settings.get('phash_threshold', 10) if phash and phash_threshold > 0: # Load existing pHashes for comparison # Note: In production with millions of images, consider using # a spatial index or pre-computed pHash buckets 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': # Merge archive tag if this file came from an archive archive_tag_id = context.get('archive_tag_id') if archive_tag_id and similar_id: similar_record = ImageRecord.query.get(similar_id) if similar_record: _add_archive_tag(similar_record, archive_tag_id) db.session.commit() log.info(f"Added archive tag {archive_tag_id} to similar image {similar_id}") # Still try to enrich the existing larger image with sidecar metadata # Uses archive sidecar as fallback for files from archives sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path) if sidecar_path and similar_id: apply_sidecar_metadata.delay(similar_id, sidecar_path) return _skip_task(task, 'Similar larger image exists', result_image_id=similar_id) # 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, archive_path, archive_sidecar_path ) else: # Quick mode: only compute pHash for storage, not comparison phash = calculate_perceptual_hash(src_path) # Normal import - copy to destination target_dir = os.path.join(dest_dir, artist) os.makedirs(target_dir, exist_ok=True) # Transcode video if needed (non-MP4 formats to H.264 MP4) transcoded_temp = None actual_src = src_path actual_filename = filename if needs_transcode(src_path): log.info(f"Transcoding video: {filename}") # Transcode to temp location first transcoded_temp = transcode_video_to_mp4(src_path) if transcoded_temp: actual_src = transcoded_temp # Update filename to .mp4 extension actual_filename = os.path.splitext(filename)[0] + '.mp4' # Recalculate hash for transcoded file file_hash = calculate_hash(transcoded_temp) task.file_hash = file_hash log.info(f"Transcoded successfully: {actual_filename}") else: log.warning(f"Transcode failed for {filename}, importing original") dest_path = build_hashed_dest_path(target_dir, actual_filename, file_hash) shutil.copy2(actual_src, dest_path) # Clean up transcoded temp file if transcoded_temp and os.path.exists(transcoded_temp): os.remove(transcoded_temp) # 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 {actual_filename}: {e}") thumb_path = None # Create ImageRecord # Use actual_filename in case video was transcoded to .mp4 record = ImageRecord( filename=actual_filename, filepath=dest_path, thumb_path=thumb_path, hash=file_hash, perceptual_hash=str(phash) if phash else None, file_size=os.path.getsize(dest_path), # Use actual file size after transcode width=metadata['width'], height=metadata['height'], format=metadata['format'] if not needs_transcode(src_path) else 'mp4', camera_model=metadata.get('camera_model'), taken_at=metadata.get('taken_at'), imported_at=datetime.utcnow() ) # Add artist tag _add_artist_tag(record, artist) # Add archive tag if this file came from an archive archive_tag_id = context.get('archive_tag_id') if archive_tag_id: _add_archive_tag(record, archive_tag_id) 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 # Uses archive sidecar as fallback for files from archives sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_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: {actual_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. Find archive sidecar JSON if present 3. Extract to temp directory 4. Create ImportTask for each media file inside 5. Queue import tasks for extracted files (with archive sidecar path) 6. Create archive tag if files were found 7. 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_archive_tag_name, extract_archive_resilient, ExtractError ) from app.utils.metadata_enrichment import find_archive_sidecar 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') # Deep scan options is_deep_scan = context.get('deep_scan', False) log.info(f"Processing archive: {archive_path} (deep_scan={is_deep_scan})") # Find archive sidecar JSON (next to the archive file, not inside it) # This contains post metadata from Gallery-DL for the entire archive archive_sidecar_path = find_archive_sidecar(archive_path) if archive_sidecar_path: log.info(f"Found archive sidecar: {archive_sidecar_path}") else: log.debug(f"No sidecar found for archive: {archive_path}") # Compute archive hash archive_hash = calculate_hash(archive_path) task.file_hash = archive_hash # Check if already processed existing_archive = ArchiveRecord.query.filter_by(hash=archive_hash).first() if existing_archive and not is_deep_scan: return _skip_task(task, 'Archive already imported') if existing_archive and is_deep_scan: log.info(f"Deep scan: re-processing already-imported archive to update metadata") # Get archive size archive_size = os.path.getsize(archive_path) # Extract archive using resilient extractor (tries unar then 7z for RAR) tmpdir = tempfile.mkdtemp(prefix='extract_') log.info(f"Extracting to: {tmpdir}") try: extract_archive_resilient(archive_path, tmpdir) except ExtractError 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") # Get or create archive tag # For deep scan of existing archives, reuse the existing tag if existing_archive and existing_archive.tag_id: archive_tag = Tag.query.get(existing_archive.tag_id) if not archive_tag: # Tag was deleted, recreate it archive_filename = os.path.basename(archive_path) tag_name = compute_archive_tag_name(artist, archive_filename) archive_tag = Tag(name=tag_name, kind='archive') db.session.add(archive_tag) db.session.flush() else: # New archive - create tag using artist:archive_name format archive_filename = os.path.basename(archive_path) tag_name = compute_archive_tag_name(artist, archive_filename) 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 only if it doesn't exist if not existing_archive: 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() else: # Deep scan - commit any tag changes db.session.commit() # Queue tasks for each media file queued_count = 0 for media_path in media_files: # Build child context with archive info child_context = { 'artist': artist, 'dest_dir': dest_dir, 'archive_task_id': task.id, 'archive_tag_id': archive_tag.id, 'temp_dir': tmpdir, 'archive_path': archive_path, # Original archive location (for sidecar lookups) } # Pass archive sidecar path so child tasks can use it as fallback if archive_sidecar_path: child_context['archive_sidecar_path'] = archive_sidecar_path # Pass through deep scan flags from parent task for flag in ('deep_scan', 'reapply_sidecars', 'regenerate_thumbnails', 'verify_hashes'): if context.get(flag): child_context[flag] = context[flag] child_task = ImportTask( source_path=media_path, task_type='import_image', batch_id=task.batch_id, context=child_context, 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 if archive_sidecar_path: task.context['archive_sidecar_path'] = archive_sidecar_path 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 _regenerate_thumbnail(record: ImageRecord) -> bool: """ Regenerate thumbnail for an existing image record. Used by deep scan to refresh thumbnails. Returns: True if thumbnail was regenerated, False on error """ from app.utils.image_importer import generate_thumbnail, generate_video_thumbnail_mirrored try: filepath = record.filepath if not filepath or not os.path.exists(filepath): log.warning(f"Cannot regenerate thumbnail: file not found {filepath}") return False lower = filepath.lower() if lower.endswith(('.mp4', '.mov', '.webm', '.avi', '.mkv')): new_thumb = generate_video_thumbnail_mirrored(filepath) else: new_thumb = generate_thumbnail(filepath, overwrite=True) if new_thumb: record.thumb_path = new_thumb db.session.commit() log.info(f"Regenerated thumbnail for image {record.id}") return True return False except Exception as e: log.warning(f"Failed to regenerate thumbnail for image {record.id}: {e}") return False def _supersede_existing(task: ImportTask, old_record: ImageRecord, src_path: str, file_hash: str, phash, metadata: dict, artist: str, dest_dir: str, filename: str, archive_path: str = None, archive_sidecar_path: str = None) -> 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 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 # Uses archive sidecar as fallback for files from archives sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_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 ) # Add archive tag if this file came from an archive context = task.context or {} archive_tag_id = context.get('archive_tag_id') if archive_tag_id: _add_archive_tag(record, archive_tag_id) 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 }