# 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 @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 ) 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') # 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) 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 sidecar_path = find_sidecar_json(src_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 sidecar_path = find_sidecar_json(src_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 ) 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 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: {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. 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, extract_archive_resilient, ExtractError ) 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 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") # 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 _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) -> 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 ) # 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 }