# app/utils/image_importer.py from __future__ import annotations from datetime import datetime from pathlib import Path import hashlib import json import mimetypes import os import re import shutil import subprocess import tempfile import time import uuid from PIL import Image, ImageOps import exifread import imagehash from pyunpack import Archive from app import db from app.models import ImageRecord, Tag, ArchiveRecord FFMPEG_THUMB_TIMEOUT = int(os.environ.get("THUMBS_FFMPEG_TIMEOUT", "60")) # seconds FFMPEG_TRANSCODE_TIMEOUT = int(os.environ.get("FFMPEG_TRANSCODE_TIMEOUT", "600")) # 10 min default Image.MAX_IMAGE_PIXELS = int(os.environ.get("PIL_MAX_IMAGE_PIXELS", "178956970")) ARCHIVE_NUM_WIDTH = int(os.environ.get("ARCHIVE_NUM_WIDTH", "4")) # ============================================================================= # Configuration & Constants # ============================================================================= THUMB_SIZE = (400, 400) # Media and archive extensions we handle IMAGE_EXTS = (".jpg", ".jpeg", ".jfif", ".png", ".gif", ".bmp", ".tiff", ".webp") VIDEO_EXTS = (".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv") VIDEO_EXTS_NEED_TRANSCODE = (".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv") # Non-MP4 formats ALLOWED_MEDIA_EXTS = IMAGE_EXTS + VIDEO_EXTS ALLOWED_ARCHIVE_EXTS = ( ".zip", ".rar", ".cbr", ".7z", ".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz" ) # Import filter settings (loaded from file or env) IMPORT_SETTINGS_PATH = "/import/settings.json" IMPORT_STATS_PATH = "/import/stats.json" def load_import_settings() -> dict: """ Load import filter settings from database, with file and environment fallbacks. Priority: DB settings > file settings > environment variables """ # Start with environment defaults defaults = { "min_width": int(os.environ.get("IMPORT_MIN_WIDTH", "0")), "min_height": int(os.environ.get("IMPORT_MIN_HEIGHT", "0")), "skip_transparent": os.environ.get("IMPORT_SKIP_TRANSPARENT", "false").lower() == "true", "transparency_threshold": float(os.environ.get("IMPORT_TRANSPARENCY_THRESHOLD", "0.9")), "skip_single_color": os.environ.get("IMPORT_SKIP_SINGLE_COLOR", "true").lower() == "true", "single_color_threshold": float(os.environ.get("IMPORT_SINGLE_COLOR_THRESHOLD", "0.95")), "single_color_tolerance": int(os.environ.get("IMPORT_SINGLE_COLOR_TOLERANCE", "30")), "phash_threshold": int(os.environ.get("IMPORT_PHASH_THRESHOLD", "10")), # Default for 256-bit "supersede_smaller": os.environ.get("IMPORT_SUPERSEDE_SMALLER", "true").lower() == "true", } # Try to load from file (legacy support) try: if os.path.exists(IMPORT_SETTINGS_PATH): with open(IMPORT_SETTINGS_PATH, "r") as f: file_settings = json.load(f) defaults.update(file_settings) except Exception as e: print(f"[WARN] Failed to load import settings from file: {e}") # Try to load from database (preferred) try: from app.utils.version_migration import get_import_settings db_settings = get_import_settings() # Only override if DB has non-default values for key, value in db_settings.items(): if value is not None: defaults[key] = value except Exception as e: print(f"[WARN] Failed to load import settings from DB: {e}") return defaults def save_import_stats(stats: dict) -> None: """Save import statistics to file.""" try: with open(IMPORT_STATS_PATH, "w") as f: json.dump(stats, f, indent=2) except Exception as e: print(f"[WARN] Failed to save import stats: {e}") def is_mostly_transparent(file_path: str, threshold: float = 0.9) -> bool: """ Check if an image is mostly transparent (> threshold % of pixels are transparent). Returns False for images without alpha channel. """ try: with Image.open(file_path) as img: if img.mode not in ("RGBA", "LA", "P"): return False # Convert palette images with transparency if img.mode == "P" and "transparency" in img.info: img = img.convert("RGBA") elif img.mode == "LA": img = img.convert("RGBA") elif img.mode != "RGBA": return False # Get alpha channel and count transparent pixels alpha = img.split()[-1] total_pixels = alpha.size[0] * alpha.size[1] # Count pixels where alpha < 128 (more than 50% transparent) transparent_count = sum(1 for p in alpha.getdata() if p < 128) transparency_ratio = transparent_count / total_pixels return transparency_ratio > threshold except Exception as e: print(f"[WARN] Failed to check transparency for {file_path}: {e}") return False def is_mostly_single_color(file_path: str, threshold: float = 0.95, tolerance: int = 30) -> bool: """ Check if an image is mostly a single color (> threshold % of pixels are similar). Uses color distance tolerance to group similar colors. Args: file_path: Path to the image file threshold: Percentage of pixels that must be similar (0.0-1.0, default 0.95 = 95%) tolerance: Maximum color distance to be considered "same" color (0-255 per channel, default 30) Returns: True if the image is mostly a single solid color """ try: with Image.open(file_path) as img: # Convert to RGB for consistent color comparison if img.mode in ("RGBA", "LA"): # For images with alpha, only consider non-transparent pixels img_rgba = img.convert("RGBA") if img.mode != "RGBA" else img pixels = list(img_rgba.getdata()) # Filter out mostly transparent pixels (alpha < 128) opaque_pixels = [(r, g, b) for r, g, b, a in pixels if a >= 128] if len(opaque_pixels) == 0: return True # Fully transparent = effectively single color pixels = opaque_pixels elif img.mode == "P": img = img.convert("RGB") pixels = list(img.getdata()) elif img.mode == "L": # Grayscale - convert to RGB tuples pixels = [(p, p, p) for p in img.getdata()] else: if img.mode != "RGB": img = img.convert("RGB") pixels = list(img.getdata()) if len(pixels) == 0: return True # Sample pixels for performance on large images total_pixels = len(pixels) if total_pixels > 10000: # Sample evenly distributed pixels step = total_pixels // 10000 pixels = pixels[::step] # Find the dominant color (most common pixel or average of sampled area) # Use the pixel at center as reference center_idx = len(pixels) // 2 ref_color = pixels[center_idx] # Count pixels within tolerance of reference color similar_count = 0 for pixel in pixels: if isinstance(pixel, int): pixel = (pixel, pixel, pixel) # Calculate color distance (simple RGB distance) dist = abs(pixel[0] - ref_color[0]) + abs(pixel[1] - ref_color[1]) + abs(pixel[2] - ref_color[2]) if dist <= tolerance * 3: # tolerance per channel * 3 channels similar_count += 1 similarity_ratio = similar_count / len(pixels) return similarity_ratio > threshold except Exception as e: print(f"[WARN] Failed to check single color for {file_path}: {e}") return False def supersede_image(old_record: "ImageRecord", new_filepath: str, new_thumb_path: str, new_hash: str, new_phash, new_metadata: dict, new_filename: str) -> "ImageRecord": """ Replace a smaller image with a larger version, preserving tags and metadata. Args: old_record: The existing ImageRecord to supersede new_filepath: Path to the new larger image file new_thumb_path: Path to the new thumbnail new_hash: SHA256 hash of the new file new_phash: Perceptual hash of the new image new_metadata: Metadata dict for the new image new_filename: Original filename of the new image Returns: The updated ImageRecord """ # Store old file paths for cleanup old_filepath = old_record.filepath old_thumb_path = old_record.thumb_path print(f"[SUPERSEDE] Replacing {old_record.filename} ({old_record.width}x{old_record.height}) " f"with {new_filename} ({new_metadata['width']}x{new_metadata['height']})") # Update record with new file info old_record.filename = new_filename old_record.filepath = new_filepath old_record.thumb_path = new_thumb_path old_record.hash = new_hash old_record.perceptual_hash = str(new_phash) if new_phash else None old_record.file_size = new_metadata["file_size"] old_record.width = new_metadata["width"] old_record.height = new_metadata["height"] old_record.format = new_metadata["format"] # Keep the earlier taken_at date (prefer original date) if new_metadata.get("taken_at") and old_record.taken_at: if new_metadata["taken_at"] < old_record.taken_at: old_record.taken_at = new_metadata["taken_at"] elif new_metadata.get("taken_at") and not old_record.taken_at: old_record.taken_at = new_metadata["taken_at"] # Don't update imported_at - keep original import time # Delete old files try: if old_filepath and os.path.exists(old_filepath): os.remove(old_filepath) print(f"[SUPERSEDE] Deleted old file: {old_filepath}") except Exception as e: print(f"[WARN] Failed to delete old file {old_filepath}: {e}") try: if old_thumb_path and os.path.exists(old_thumb_path): os.remove(old_thumb_path) print(f"[SUPERSEDE] Deleted old thumbnail: {old_thumb_path}") except Exception as e: print(f"[WARN] Failed to delete old thumbnail {old_thumb_path}: {e}") return old_record # Regex for multi-part detection RAR_PART_RE = re.compile(r"\.part(\d+)\.rar$", re.IGNORECASE) SEVENZ_PART_RE = re.compile(r"\.7z\.(\d{3})$", re.IGNORECASE) OLD_RAR_VOL_RE = re.compile(r"\.r\d{2}$", re.IGNORECASE) # .r00, .r01, ... # Environment configuration (optional) ENV_TMPDIR = "ARCHIVE_TMPDIR" # where to extract archives (default: system tmp) ENV_MINFREE = "ARCHIVE_MIN_FREE_GB" # min free GB required to start extraction (float; 0 = disabled) # ============================================================================= # Utilities: Temp space management & free-space guard # ============================================================================= def get_tmp_base() -> Path: """Return the base temp directory for archive extraction.""" return Path(os.environ.get(ENV_TMPDIR, tempfile.gettempdir())).resolve() def cleanup_stale_tempdirs(prefix: str = "extract_") -> None: """Delete any leftover extraction directories from previous crashes.""" base = get_tmp_base() base.mkdir(parents=True, exist_ok=True) for entry in base.glob(f"{prefix}*"): if entry.is_dir(): try: shutil.rmtree(entry, ignore_errors=True) except Exception: pass def ensure_free_space_or_raise(path: Path, min_free_gb: float) -> None: """Raise if there is not enough free space at path.""" if min_free_gb <= 0: return total, used, free = shutil.disk_usage(str(path)) free_gb = free / (1024 ** 3) if free_gb < min_free_gb: raise RuntimeError( f"Insufficient free space at {path}: {free_gb:.1f} GB free " f"(required >= {min_free_gb:.1f} GB)." ) # ============================================================================= # Utilities: External extractors with resilient fallback # ============================================================================= class ExtractError(Exception): """Raised when archive extraction fails with all strategies.""" def _run_cmd(cmd: list[str]) -> tuple[int, str, str]: p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) return p.returncode, p.stdout, p.stderr def extract_with_unar(archive_path: str, out_dir: str) -> None: os.makedirs(out_dir, exist_ok=True) rc, out, err = _run_cmd(["unar", "-o", out_dir, archive_path]) if rc != 0: raise ExtractError(f"unar failed (rc={rc})\nSTDOUT:\n{out}\nSTDERR:\n{err}") def extract_with_7z(archive_path: str, out_dir: str) -> None: os.makedirs(out_dir, exist_ok=True) rc, out, err = _run_cmd(["7z", "x", "-y", f"-o{out_dir}", archive_path]) if rc != 0: raise ExtractError(f"7z failed (rc={rc})\nSTDOUT:\n{out}\nSTDERR:\n{err}") def looks_password_protected_with_lsar(archive_path: str) -> bool: """Best-effort pre-check for 'Encrypted/Password' in lsar output.""" try: rc, out, err = _run_cmd(["lsar", archive_path]) txt = (out + "\n" + err).lower() return ("encrypted" in txt) or ("password" in txt) except Exception: return False def extract_archive_resilient(archive_path: str, out_dir: str) -> None: """ Prefer unar for RAR/CBR (RAR5-friendly); fallback to 7z. For other formats, try 7z first then unar. Surfaces stderr so logs show CRC/password/permissions issues. """ lower = archive_path.lower() errors = [] # Optional password hint to skip early if lower.endswith((".rar", ".cbr")) and looks_password_protected_with_lsar(archive_path): raise ExtractError("Archive appears password-protected; skipping (no password provided).") if lower.endswith((".rar", ".cbr")): order = (extract_with_unar, extract_with_7z) else: order = (extract_with_7z, extract_with_unar) for fn in order: try: fn(archive_path, out_dir) return except ExtractError as e: errors.append(str(e)) raise ExtractError("Both extractors failed:\n" + "\n---\n".join(errors)) # ============================================================================= # Core: Import task & archive processing # ============================================================================= def import_images_task(source_dir: str, dest_dir: str) -> str: """ Walk the import tree, process archives first (only first volumes), then media files. Batches DB commits for efficiency. """ # Cleanup any stale extraction dirs from previous runs/crashes cleanup_stale_tempdirs() # Load import filter settings settings = load_import_settings() min_width = settings.get("min_width", 0) min_height = settings.get("min_height", 0) skip_transparent = settings.get("skip_transparent", False) transparency_threshold = settings.get("transparency_threshold", 0.9) skip_single_color = settings.get("skip_single_color", True) single_color_threshold = settings.get("single_color_threshold", 0.95) single_color_tolerance = settings.get("single_color_tolerance", 30) phash_threshold = settings.get("phash_threshold", 2) supersede_smaller = settings.get("supersede_smaller", True) print(f"[INFO] Import settings: min_width={min_width}, min_height={min_height}, " f"skip_transparent={skip_transparent}, transparency_threshold={transparency_threshold}, " f"skip_single_color={skip_single_color}, single_color_threshold={single_color_threshold}, " f"phash_threshold={phash_threshold}, supersede_smaller={supersede_smaller}") # Statistics tracking stats = { "total_processed": 0, "total_imported": 0, "total_superseded": 0, "filtered_by_dimension": 0, "filtered_by_transparency": 0, "filtered_by_single_color": 0, "filtered_by_duplicate": 0, } imported: list[str] = [] batch_size = 10 batch_counter = 0 # Preload existing pHashes to avoid N+1 lookups # Format: (phash, width, height, image_id) for supersede functionality existing_phashes = [ (imagehash.hex_to_hash(img.perceptual_hash), img.width, img.height, img.id) for img in ImageRecord.query.filter(ImageRecord.perceptual_hash != None).all() ] # Ensure thumbs base exists (Path(dest_dir) / "thumbs").mkdir(parents=True, exist_ok=True) # Optional free-space guard min_free = float(os.environ.get(ENV_MINFREE, "0") or "0") try: ensure_free_space_or_raise(get_tmp_base(), min_free) except Exception as e: print(f"[WARN] Free-space check failed: {e}") 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 for root, _, files in os.walk(artist_path): for file in files: full_path = os.path.join(root, file) lower = file.lower() # Archives first if lower.endswith(ALLOWED_ARCHIVE_EXTS): if not is_first_volume(full_path): print(f"[SKIP] Not first archive volume: {file}") continue try: count = process_archive( archive_path=full_path, dest_dir=dest_dir, artist=artist_dir, existing_phashes=existing_phashes, settings=settings, stats=stats ) imported.append(f"[archive]{file}:{count}") except Exception as e: print(f"[WARN] Failed processing archive {file}: {e}") continue # Non-archive media if not lower.endswith(ALLOWED_MEDIA_EXTS): continue stats["total_processed"] += 1 # Apply dimension filter if min_width > 0 or min_height > 0: md = extract_metadata(full_path) if md.get("width") and md.get("height"): if md["width"] < min_width or md["height"] < min_height: print(f"[SKIP] Too small ({md['width']}x{md['height']}): {file}") stats["filtered_by_dimension"] += 1 continue # Apply transparency filter if skip_transparent and lower.endswith((".png", ".gif", ".webp")): if is_mostly_transparent(full_path, transparency_threshold): print(f"[SKIP] Mostly transparent: {file}") stats["filtered_by_transparency"] += 1 continue # Apply single-color filter (skip images that are mostly one solid color) if skip_single_color and lower.endswith((".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp")): if is_mostly_single_color(full_path, single_color_threshold, single_color_tolerance): print(f"[SKIP] Mostly single color: {file}") stats["filtered_by_single_color"] += 1 continue added, phash, _rec, superseded = import_single_file( src_path=full_path, dest_dir=dest_dir, artist=artist_dir, existing_phashes=existing_phashes, phash_threshold=phash_threshold, supersede_smaller=supersede_smaller ) if added: imported.append(file) if superseded: stats["total_superseded"] += 1 else: stats["total_imported"] += 1 elif _rec is None and phash is None: # Was a duplicate stats["filtered_by_duplicate"] += 1 # Add new image's phash to the tracking list (for fresh imports only, not supersedes) if phash and _rec and not superseded: existing_phashes.append((phash, _rec.width, _rec.height, _rec.id)) batch_counter += 1 if batch_counter >= batch_size: db.session.commit() print(f"[INFO] Committed batch of {batch_size} items.") batch_counter = 0 if "sqlite" in db.engine.url.drivername: time.sleep(1) if batch_counter > 0: db.session.commit() print(f"[INFO] Committed final batch of {batch_counter} items.") # Save stats save_import_stats(stats) summary = (f"Imported {stats['total_imported']} items, superseded {stats['total_superseded']} smaller versions. " f"Filtered: {stats['filtered_by_dimension']} by size, {stats['filtered_by_transparency']} by transparency, " f"{stats['filtered_by_single_color']} by single color, {stats['filtered_by_duplicate']} duplicates.") print(f"[INFO] {summary}") return summary def process_archive(archive_path, dest_dir, artist, existing_phashes, settings=None, stats=None): """ Extracts archive to temp dir, imports/tag existing images, and records ArchiveRecord. Creates an archive:* tag ONLY if at least one image was imported or tagged. Returns number of items imported or tagged. """ if settings is None: settings = load_import_settings() if stats is None: stats = {"total_processed": 0, "total_imported": 0, "total_superseded": 0, "filtered_by_dimension": 0, "filtered_by_transparency": 0, "filtered_by_single_color": 0, "filtered_by_duplicate": 0} min_width = settings.get("min_width", 0) min_height = settings.get("min_height", 0) skip_transparent = settings.get("skip_transparent", False) transparency_threshold = settings.get("transparency_threshold", 0.9) skip_single_color = settings.get("skip_single_color", True) single_color_threshold = settings.get("single_color_threshold", 0.95) single_color_tolerance = settings.get("single_color_tolerance", 30) phash_threshold = settings.get("phash_threshold", 2) supersede_smaller = settings.get("supersede_smaller", True) archive_size = os.path.getsize(archive_path) archive_hash = calculate_hash(archive_path) # Skip if this exact archive file was already processed (hash-based) existing_archive = ArchiveRecord.query.filter_by(hash=archive_hash).first() if existing_archive: print(f"[SKIP] Archive already imported: {archive_path}") return 0 tmpdir = tempfile.mkdtemp(prefix="extract_") touched_records = [] # ImageRecord objects we imported or matched imported_or_tagged = 0 created_tag = None try: # Extract using pyunpack Archive(archive_path).extractall(tmpdir) # Import / collect all media for root, _, files in os.walk(tmpdir): for file in files: lower = file.lower() if not lower.endswith(ALLOWED_MEDIA_EXTS): continue src_path = os.path.join(root, file) stats["total_processed"] += 1 # Apply dimension filter if min_width > 0 or min_height > 0: md = extract_metadata(src_path) if md.get("width") and md.get("height"): if md["width"] < min_width or md["height"] < min_height: print(f"[SKIP] Too small ({md['width']}x{md['height']}): {file}") stats["filtered_by_dimension"] += 1 continue # Apply transparency filter if skip_transparent and lower.endswith((".png", ".gif", ".webp")): if is_mostly_transparent(src_path, transparency_threshold): print(f"[SKIP] Mostly transparent: {file}") stats["filtered_by_transparency"] += 1 continue # Apply single-color filter if skip_single_color and lower.endswith((".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp")): if is_mostly_single_color(src_path, single_color_threshold, single_color_tolerance): print(f"[SKIP] Mostly single color: {file}") stats["filtered_by_single_color"] += 1 continue added, _phash, rec, superseded = import_single_file( src_path=src_path, dest_dir=dest_dir, artist=artist, # still used for auto artist: tag existing_phashes=existing_phashes, commit=False, phash_threshold=phash_threshold, supersede_smaller=supersede_smaller ) if rec is not None: touched_records.append(rec) if added: imported_or_tagged += 1 if superseded: stats["total_superseded"] += 1 else: stats["total_imported"] += 1 elif rec is None and _phash is None: stats["filtered_by_duplicate"] += 1 # Only now: create/attach the archive tag if we actually touched images if touched_records: # Incremental name like 'archive:{artist}/0001' tag_name = compute_next_archive_tag_name(artist) created_tag = Tag.query.filter_by(name=tag_name).first() if not created_tag: created_tag = Tag(name=tag_name, kind='archive') db.session.add(created_tag) db.session.flush() newly_tagged = 0 for rec in touched_records: if created_tag not in rec.tags: rec.tags.append(created_tag) newly_tagged += 1 imported_or_tagged += newly_tagged # Record this archive as processed (tag may be None if no media) arch = ArchiveRecord( filename=archive_path, file_size=archive_size, hash=archive_hash, artist=artist, tag=created_tag ) db.session.add(arch) db.session.commit() print(f"[INFO] Archive processed: {archive_path} items={imported_or_tagged}") return imported_or_tagged finally: shutil.rmtree(tmpdir, ignore_errors=True) # ============================================================================= # Core: Single-file import path (content-addressed storage) # ============================================================================= def build_hashed_dest_path(target_dir: str, original_name: str, content_hash: str) -> str: """ Always produce a content-addressed destination: /__ If that path already exists, append a numeric suffix. """ base, ext = os.path.splitext(original_name) candidate = os.path.join(target_dir, f"{base}__{content_hash[:10]}{ext}") if not os.path.exists(candidate): return candidate # Rare: collision (e.g., concurrent import) → add counter i = 2 while True: candidate = os.path.join(target_dir, f"{base}__{content_hash[:10]}_{i}{ext}") if not os.path.exists(candidate): return candidate i += 1 def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phashes, commit: bool = True, phash_threshold: int = 2, supersede_smaller: bool = True): """ Import a single media file. If an equivalent file already exists, return that record so callers can tag it (no duplicate import). If a smaller visually-similar image exists and supersede_smaller is True, the smaller image will be replaced with the larger one (preserving tags). Also enriches metadata from Gallery-DL sidecar JSON files if present. Returns (added: bool, phash_or_None, record_or_None, superseded: bool). The superseded flag indicates if an existing smaller image was replaced. """ original_name = os.path.basename(src_path) print(f"[INFO] Processing: {src_path}") file_size = os.path.getsize(src_path) # Compute content hash first (authoritative de-dup + used in filepath) file_hash = calculate_hash(src_path) existing = ImageRecord.query.filter_by(hash=file_hash).first() if existing: print(f"[SKIP] Duplicate by hash: {original_name}") # Even for duplicates, try to enrich with sidecar metadata _enrich_existing_record(existing, src_path) return (False, None, existing, False) # Optional info: same name+size (but different content) → we still import with hashed name existing_ns = ImageRecord.query.filter_by(filename=original_name, file_size=file_size).first() if existing_ns: print(f"[INFO] Same name+size exists but different hash; storing with hashed name: {original_name}") # Metadata & pHash similarity metadata = extract_metadata(src_path) if not metadata["width"] or not metadata["height"]: print(f"[SKIP] Missing dimension data for {original_name}") return (False, None, None, False) phash = calculate_perceptual_hash(src_path) # Check for similar images superseded = False if phash and phash_threshold > 0: relationship, similar_id = find_similar_image( phash, metadata["width"], metadata["height"], existing_phashes, threshold=phash_threshold ) if relationship == "larger_exists": print(f"[SKIP] {original_name} is visually similar to an existing larger image (threshold={phash_threshold}).") return (False, phash, None, False) elif relationship == "smaller_exists" and supersede_smaller and similar_id: # Found a smaller version - supersede it old_record = ImageRecord.query.get(similar_id) if old_record: # Enrich metadata from Gallery-DL sidecar JSON if present enriched = _get_enriched_metadata(src_path, metadata) metadata["taken_at"] = enriched.get("taken_at") or metadata["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, original_name, file_hash) shutil.copy2(src_path, dest_path) # Generate new thumbnail try: if dest_path.lower().endswith((".mp4", ".mov")): thumb_path = generate_video_thumbnail_mirrored(dest_path) else: thumb_path = generate_thumbnail(dest_path) except Exception as e: print(f"[WARN] Failed to generate thumbnail for {original_name}: {e}") thumb_path = None # Supersede the old record (preserves tags) record = supersede_image( old_record, dest_path, thumb_path, file_hash, phash, metadata, original_name ) # Apply any new tags from sidecar _apply_enriched_tags(record, enriched) if commit: db.session.commit() # Update existing_phashes list with new dimensions for i, (ph, w, h, img_id) in enumerate(existing_phashes): if img_id == similar_id: existing_phashes[i] = (phash, metadata["width"], metadata["height"], img_id) break return (True, phash, record, True) # superseded=True # No similar image found (or supersede disabled) - normal import # Enrich metadata from Gallery-DL sidecar JSON if present enriched = _get_enriched_metadata(src_path, metadata) # Copy into /images//__ target_dir = os.path.join(dest_dir, artist) os.makedirs(target_dir, exist_ok=True) dest_path = build_hashed_dest_path(target_dir, original_name, file_hash) shutil.copy2(src_path, dest_path) # Thumbnails (mirrored for both images and videos) try: if dest_path.lower().endswith((".mp4", ".mov")): thumb_path = generate_video_thumbnail_mirrored(dest_path) else: thumb_path = generate_thumbnail(dest_path) except Exception as e: print(f"[WARN] Failed to generate thumbnail for {original_name}: {e}") thumb_path = None # Create DB record (keep display name as original; filepath is content-addressed) # Use enriched taken_at (earliest date wins) record = ImageRecord( filename=original_name, 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["camera_model"], taken_at=enriched.get("taken_at") or metadata["taken_at"], imported_at=datetime.utcnow() ) # Auto-tag with artist: (treat artist as a tag only) if artist: artist_tag_name = f"artist:{artist}" artist_tag = Tag.query.filter_by(name=artist_tag_name).first() if not artist_tag: artist_tag = Tag(name=artist_tag_name, kind="artist") db.session.add(artist_tag) db.session.flush() if artist_tag not in record.tags: record.tags.append(artist_tag) # Add tags from Gallery-DL metadata (source:platform, post:platform:artist:id) _apply_enriched_tags(record, enriched) db.session.add(record) if commit: db.session.commit() return (True, phash, record, False) # superseded=False (new import) def _get_enriched_metadata(src_path: str, existing_metadata: dict) -> dict: """ Get enriched metadata from Gallery-DL sidecar JSON. Returns empty dict if no sidecar found. """ try: from app.utils.metadata_enrichment import enrich_from_sidecar return enrich_from_sidecar(src_path, existing_metadata) except ImportError: return {} except Exception as e: print(f"[WARN] Failed to enrich metadata for {src_path}: {e}") return {} def _apply_enriched_tags(record: ImageRecord, enriched: dict) -> None: """ Apply tags from enriched metadata to an ImageRecord. Creates tags if they don't exist. """ if not enriched or "tags" not in enriched: return 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() if tag not in record.tags: record.tags.append(tag) print(f"[INFO] Added tag: {tag_name}") def _enrich_existing_record(record: ImageRecord, src_path: str) -> None: """ Enrich an existing record with metadata from sidecar JSON. Used when a duplicate is found - we still want to merge metadata. - Updates taken_at if sidecar date is earlier - Adds source/post tags if not already present """ try: from app.utils.metadata_enrichment import enrich_from_sidecar, resolve_date_conflict enriched = enrich_from_sidecar(src_path, {"taken_at": record.taken_at}) if not enriched.get("raw_metadata"): return # No sidecar found # Update date if sidecar has an earlier date new_date = resolve_date_conflict(record.taken_at, enriched.get("taken_at")) if new_date != record.taken_at: print(f"[INFO] Updating taken_at for {record.filename}: {record.taken_at} -> {new_date}") record.taken_at = new_date # Add any new tags _apply_enriched_tags(record, enriched) except Exception as e: print(f"[WARN] Failed to enrich existing record: {e}") # ============================================================================= # Helpers: Thumbnails, metadata, hashing, similarity # ============================================================================= def generate_thumbnail( image_path, size=(400, 400), overwrite=False, prefer_jpeg=False, # False = auto-pick PNG for alpha images; True = always JPEG jpeg_bg=(0, 0, 0), # background used when flattening to JPEG ): """ Save thumbnails mirrored under /images/thumbs/<...>. If prefer_jpeg is False (default), transparent images save as PNG, others as JPEG. If prefer_jpeg is True, all thumbs are JPEG with transparency flattened to `jpeg_bg`. For images with extreme aspect ratios (very tall like comic strips, or very wide), crops to the center region before thumbnailing to produce better quality previews. Returns the absolute path to the thumbnail. """ images_root = Path("/images").resolve() image_path = Path(image_path).resolve() try: rel_path = image_path.relative_to(images_root) except ValueError: raise ValueError(f"{image_path} is not under {images_root}") # Base path under /images/thumbs with same subfolders/filename (we may change the suffix) thumb_base = (images_root / "thumbs" / rel_path).with_suffix("") # drop original suffix for control # Decide output extension/format with Image.open(image_path) as img: img = ImageOps.exif_transpose(img) # honor camera rotation # Handle extreme aspect ratios by cropping to a more balanced region # This prevents very tall/wide images from becoming tiny slivers w, h = img.size aspect_ratio = w / h if h > 0 else 1 # If aspect ratio is extreme (< 0.5 means very tall, > 2.0 means very wide) # crop to a more balanced region before thumbnailing if aspect_ratio < 0.5: # Very tall image (like comic strip) - crop from top portion # Take a region that's closer to square, from the top crop_height = min(h, w * 2) # At most 2:1 tall img = img.crop((0, 0, w, crop_height)) elif aspect_ratio > 2.0: # Very wide image - crop from center crop_width = min(w, h * 2) # At most 2:1 wide left = (w - crop_width) // 2 img = img.crop((left, 0, left + crop_width, h)) # Use LANCZOS resampling for high-quality downscaling img.thumbnail(size, Image.Resampling.LANCZOS) has_alpha = _has_alpha(img) if prefer_jpeg: # Force JPEG for consistency out_path = thumb_base.with_suffix(".jpg") out_path.parent.mkdir(parents=True, exist_ok=True) if not overwrite and out_path.exists(): return str(out_path) img_rgb = _flatten_to_rgb(img, bg=jpeg_bg) img_rgb.save(out_path, format="JPEG", quality=85, optimize=True, progressive=True) return str(out_path) else: # Auto-pick: PNG for alpha, JPEG otherwise if has_alpha: out_path = thumb_base.with_suffix(".png") out_path.parent.mkdir(parents=True, exist_ok=True) if not overwrite and out_path.exists(): return str(out_path) # Keep transparency if present # Convert palette to RGBA for reliable PNG writing if img.mode == "P": img = img.convert("RGBA") img.save(out_path, format="PNG", optimize=True) return str(out_path) else: out_path = thumb_base.with_suffix(".jpg") out_path.parent.mkdir(parents=True, exist_ok=True) if not overwrite and out_path.exists(): return str(out_path) if img.mode != "RGB": img = img.convert("RGB") img.save(out_path, format="JPEG", quality=85, optimize=True, progressive=True) return str(out_path) def generate_video_thumbnail(video_path: str, output_path: str, time_position: str = "00:00:01") -> str | None: """ Extract a single JPG frame from a video using ffmpeg and write it to output_path. Returns output_path on success, or None on failure. """ try: os.makedirs(os.path.dirname(output_path), exist_ok=True) cmd = [ "ffmpeg", "-y", "-ss", time_position, "-i", str(video_path), "-vframes", "1", "-vf", f"scale={THUMB_SIZE[0]}:-1", str(output_path), ] subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=FFMPEG_THUMB_TIMEOUT) return output_path except subprocess.CalledProcessError as e: print(f"[WARN] Failed to extract video thumbnail for {video_path}: {e}") return None def generate_video_thumbnail_mirrored(video_path: str, time_position: str = "00:00:01") -> str | None: """ Mirror the stored (hashed) video path under /images/thumbs/... and save as .jpg. """ images_root = Path("/images").resolve() video_path_p = Path(video_path).resolve() rel_path = video_path_p.relative_to(images_root) # raises if not under /images thumb_path = (images_root / "thumbs" / rel_path).with_suffix(".jpg") return generate_video_thumbnail(str(video_path_p), str(thumb_path), time_position=time_position) def transcode_video_to_mp4(input_path: str, output_path: str = None) -> str | None: """ Transcode a video to H.264 MP4 for universal browser playback. Args: input_path: Path to source video file output_path: Optional destination path. If None, replaces extension with .mp4 in a temp location. Returns: Path to transcoded MP4 file on success, None on failure. """ input_p = Path(input_path) # Determine output path if output_path is None: # Create temp file with .mp4 extension output_path = str(input_p.with_suffix(".mp4")) if output_path == input_path: # Already .mp4, use temp suffix output_path = str(input_p.with_suffix(".transcoded.mp4")) try: os.makedirs(os.path.dirname(output_path), exist_ok=True) # FFmpeg command for H.264/AAC transcoding with good browser compatibility # -movflags +faststart: Enables progressive playback (moov atom at start) # -preset medium: Balance between speed and compression # -crf 23: Good quality (lower = better, 18-28 typical range) # -c:v libx264: H.264 video codec # -c:a aac: AAC audio codec # -pix_fmt yuv420p: Pixel format for maximum compatibility cmd = [ "ffmpeg", "-y", "-i", str(input_path), "-c:v", "libx264", "-preset", "medium", "-crf", "23", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", str(output_path), ] print(f"[INFO] Transcoding video: {input_path} -> {output_path}") result = subprocess.run( cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=FFMPEG_TRANSCODE_TIMEOUT ) print(f"[INFO] Transcoding complete: {output_path}") return output_path except subprocess.TimeoutExpired: print(f"[WARN] Video transcode timed out after {FFMPEG_TRANSCODE_TIMEOUT}s: {input_path}") # Clean up partial output if os.path.exists(output_path): os.remove(output_path) return None except subprocess.CalledProcessError as e: print(f"[WARN] Failed to transcode video {input_path}: {e}") if e.stderr: print(f"[WARN] FFmpeg stderr: {e.stderr.decode('utf-8', errors='ignore')[:500]}") # Clean up partial output if os.path.exists(output_path): os.remove(output_path) return None except Exception as e: print(f"[WARN] Unexpected error transcoding video {input_path}: {e}") if os.path.exists(output_path): os.remove(output_path) return None def needs_transcode(file_path: str) -> bool: """Check if a video file needs transcoding to MP4.""" ext = Path(file_path).suffix.lower() return ext in VIDEO_EXTS_NEED_TRANSCODE def calculate_hash(file_path: str) -> str: hash_func = hashlib.sha256() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): # 1MB chunks work well for big files hash_func.update(chunk) return hash_func.hexdigest() def calculate_perceptual_hash(file_path: str, hash_size: int = 16): """ Calculate a perceptual hash for an image. Args: file_path: Path to the image file hash_size: Size of the hash grid (default 16 for 256-bit hash) hash_size=8 gives 64-bit, hash_size=16 gives 256-bit Returns: ImageHash object or None if hashing fails (e.g., for videos) """ try: with Image.open(file_path) as img: return imagehash.phash(img, hash_size=hash_size) except Exception: # videos or unreadable images land here (phash not applicable) return None def find_similar_image(phash, width: int, height: int, existing_phashes, threshold: int = 10): """ Find visually similar images and determine relationship. Args: phash: Perceptual hash of the new image width: Width of the new image height: Height of the new image existing_phashes: List of (phash, width, height, image_id) tuples for existing images threshold: Maximum Hamming distance to consider images similar Default is 10 for 256-bit hashes (was 2 for 64-bit) Returns: tuple: (relationship, image_id) where relationship is one of: - None: No similar image found - "larger_exists": A larger or equal similar image exists (skip new image) - "smaller_exists": A smaller similar image exists (supersede it) image_id is the ID of the matching image, or None """ for existing_phash, ew, eh, image_id in existing_phashes: try: distance = phash - existing_phash if distance <= threshold: # Similar image found - check size relationship if ew >= width and eh >= height: # Existing is larger or equal - skip new image return ("larger_exists", image_id) elif width > ew or height > eh: # New image is larger - supersede the existing one return ("smaller_exists", image_id) except Exception as e: print(f"[WARN] Failed pHash comparison: {e}") return (None, None) def is_similar_image(phash, width: int, height: int, existing_phashes, threshold: int = 10) -> bool: """ Check if an image is visually similar to any existing larger image. Legacy wrapper for find_similar_image - returns True only if a larger version exists. """ relationship, _ = find_similar_image(phash, width, height, existing_phashes, threshold) return relationship == "larger_exists" def extract_metadata(file_path: str) -> dict: metadata = { "file_size": None, "width": None, "height": None, "format": None, "camera_model": None, "taken_at": None } metadata["file_size"] = os.path.getsize(file_path) ext = os.path.splitext(file_path)[1].lower() # Get file modification time as fallback for taken_at try: file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path)) except Exception: file_mtime = None # Video: use ffprobe if ext in (".mp4", ".mov"): try: cmd = [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height", "-of", "json", file_path ] result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False) data = json.loads(result.stdout or "{}") stream = data.get("streams", [{}])[0] metadata["width"] = stream.get("width") metadata["height"] = stream.get("height") metadata["format"] = ext.lstrip(".") # Use file mtime for videos since they don't have EXIF metadata["taken_at"] = file_mtime except Exception as e: print(f"[WARN] Failed to extract video metadata: {file_path} – {e}") return metadata # Image: PIL + EXIF (best-effort) try: with Image.open(file_path) as img: metadata["width"], metadata["height"] = img.size metadata["format"] = (img.format or "").lower() except Exception as e: mime_type, _ = mimetypes.guess_type(file_path) print(f"[WARN] Failed to read image metadata: {file_path} ({e}) ext={ext} mime={mime_type}") try: with open(file_path, "rb") as f: tags = exifread.process_file(f, stop_tag="UNDEF", details=False) if "EXIF DateTimeOriginal" in tags: metadata["taken_at"] = datetime.strptime(str(tags["EXIF DateTimeOriginal"]), "%Y:%m:%d %H:%M:%S") if "Image Model" in tags: metadata["camera_model"] = str(tags["Image Model"]) except Exception: pass # Fallback to file modification time if no EXIF date if metadata["taken_at"] is None: metadata["taken_at"] = file_mtime return metadata def _has_alpha(img: Image.Image) -> bool: """Return True if the image has any transparency channel.""" if img.mode in ("RGBA", "LA"): return True if img.mode == "P" and "transparency" in img.info: return True return False def _flatten_to_rgb(img: Image.Image, bg=(0, 0, 0)) -> Image.Image: """ Flatten an RGBA/LA/Palette-with-alpha image onto a solid background, returning an RGB image suitable for saving as JPEG. """ # Normalize to RGBA so we can use the alpha channel as a mask if img.mode == "P": img = img.convert("RGBA") if img.mode in ("RGBA", "LA"): # Ensure we have an explicit alpha channel as last band if img.mode == "LA": img = img.convert("RGBA") alpha = img.split()[-1] # A channel bg_img = Image.new("RGB", img.size, bg) bg_img.paste(img, mask=alpha) return bg_img # No alpha, just convert to RGB return img.convert("RGB") # ============================================================================= # Helpers: Multi-part detection # ============================================================================= def is_first_volume(path: str) -> bool: """ Determine if 'path' is the first volume of a multi-part archive. We only attempt extraction from the first volume. """ name = os.path.basename(path).lower() # RAR: .partN.rar (only N==1 is first) m = RAR_PART_RE.search(name) if m: n = m.group(1) return n in ("1", "01", "001") # RAR old-style: first is .rar, subsequent are .r00, .r01... if name.endswith(".rar"): return True if OLD_RAR_VOL_RE.search(name): return False # CBR is typically a single-volume RAR if name.endswith(".cbr"): return True # 7z multi-part: .7z.001 is first m = SEVENZ_PART_RE.search(name) if m: n = m.group(1) return n in ("1", "01", "001") if name.endswith(".7z"): return True # single-volume 7z # ZIP and tarballs: treat as single-volume (we don't include .z01 in our allowlist) if name.endswith((".zip", ".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz")): return True return True def compute_next_archive_tag_name(artist: str, width: int = ARCHIVE_NUM_WIDTH) -> str: """ Return a unique incremental tag name like: 'archive:{artist}/0001', '0002', ... - Looks only at Tag(kind='archive') with names starting 'archive:{artist}/' - Finds max numeric suffix and returns next number (zero-padded). - Ensures no collision even if tags were renamed. """ prefix = f"archive:{artist}/" # Pull existing tag names for this artist/prefix rows = (Tag.query .with_entities(Tag.name) .filter(Tag.kind == 'archive', Tag.name.like(prefix + '%')) .all()) max_n = 0 for (name,) in rows: tail = name[len(prefix):] if tail.isdigit(): try: n = int(tail) if n > max_n: max_n = n except Exception: pass # Propose next number and make sure it doesn't already exist n = max_n + 1 while True: candidate = f"{prefix}{str(n).zfill(width)}" if not Tag.query.filter_by(name=candidate).first(): return candidate n += 1 def compute_next_archive_tag_name(artist: str, width: int = ARCHIVE_NUM_WIDTH) -> str: """ Return a unique incremental tag name like: 'archive:{artist}/0001', '0002', ... - Looks only at Tag(kind='archive') with names starting 'archive:{artist}/' - Finds max numeric suffix and returns next number (zero-padded). - Ensures no collision even if tags were renamed. """ prefix = f"archive:{artist}/" # Pull existing tag names for this artist/prefix rows = (Tag.query .with_entities(Tag.name) .filter(Tag.kind == 'archive', Tag.name.like(prefix + '%')) .all()) max_n = 0 for (name,) in rows: tail = name[len(prefix):] if tail.isdigit(): try: n = int(tail) if n > max_n: max_n = n except Exception: pass # Propose next number and make sure it doesn't already exist n = max_n + 1 while True: candidate = f"{prefix}{str(n).zfill(width)}" if not Tag.query.filter_by(name=candidate).first(): return candidate n += 1