# app/utils/image_importer.py from datetime import datetime import os, shutil, hashlib, time from PIL import Image import exifread import mimetypes import imagehash import uuid import subprocess from app import db from app.models import ImageRecord THUMB_SIZE = (400, 400) def import_images_task(source_dir, dest_dir): imported = [] batch_size = 10 batch_counter = 0 existing_phashes = [ (imagehash.hex_to_hash(img.perceptual_hash), img.width, img.height) for img in ImageRecord.query.filter(ImageRecord.perceptual_hash != None).all() ] 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: if file.lower().endswith(('.jpg', '.jpeg', '.jfif', '.png', '.gif', '.bmp', '.tiff', '.webp', '.mp4', '.mov')): full_path = os.path.join(root, file) filename = os.path.basename(full_path) print(f"[INFO] Processing: {full_path}") file_size = os.path.getsize(full_path) if ImageRecord.query.filter_by(filename=filename, file_size=file_size).first(): print(f"[SKIP] Duplicate by name+size: {filename}") continue file_hash = calculate_hash(full_path) if ImageRecord.query.filter_by(hash=file_hash).first(): print(f"[SKIP] Duplicate by hash: {filename}") continue metadata = extract_metadata(full_path) if not metadata['width'] or not metadata['height']: print(f"[SKIP] Missing dimension data for {filename}") continue phash = calculate_perceptual_hash(full_path) if phash and is_similar_image(phash, metadata['width'], metadata['height'], existing_phashes): print(f"[SKIP] {filename} is visually similar to an existing larger image.") continue target_dir = os.path.join(dest_dir, artist_dir) os.makedirs(target_dir, exist_ok=True) dest_path = os.path.join(target_dir, filename) shutil.copy2(full_path, dest_path) try: if dest_path.lower().endswith(('.mp4', '.mov')): thumb_filename = f"{uuid.uuid4().hex}.jpg" thumb_path = os.path.join("/images/thumbs", thumb_filename) thumb_path = generate_video_thumbnail(dest_path, thumb_path) else: thumb_path = generate_thumbnail(dest_path) except Exception as e: print(f"[WARN] Failed to generate thumbnail for {filename}: {e}") thumb_path = None record = ImageRecord( filename=filename, filepath=dest_path, thumb_path=thumb_path, hash=file_hash, perceptual_hash=str(phash) if phash else None, artist=artist_dir, file_size=metadata['file_size'], width=metadata['width'], height=metadata['height'], format=metadata['format'], camera_model=metadata['camera_model'], taken_at=metadata['taken_at'], imported_at=datetime.utcnow() ) db.session.add(record) imported.append(filename) if phash: existing_phashes.append((phash, metadata['width'], metadata['height'])) batch_counter += 1 if batch_counter >= batch_size: db.session.commit() print(f"[INFO] Committed batch of {batch_size} images.") 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} images.") print(f"[INFO] Import complete. Total images imported: {len(imported)}") return f"Imported {len(imported)} images." def generate_thumbnail(image_path, size=(400, 400), overwrite=False): from pathlib import Path 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}") thumb_path = images_root / "thumbs" / rel_path thumb_path.parent.mkdir(parents=True, exist_ok=True) if not overwrite and thumb_path.exists(): return str(thumb_path) with Image.open(image_path) as img: img.thumbnail(size) img.save(thumb_path) return str(thumb_path) def generate_video_thumbnail(video_path, output_path, time_position="00:00:01"): os.makedirs(os.path.dirname(output_path), exist_ok=True) command = [ "ffmpeg", "-i", video_path, "-ss", time_position, "-vframes", "1", "-vf", f"scale={THUMB_SIZE[0]}:-1", output_path ] try: subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return output_path except subprocess.CalledProcessError as e: print(f"[WARN] Failed to extract video thumbnail for {video_path}: {e}") return None def calculate_hash(file_path): hash_func = hashlib.sha256() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): hash_func.update(chunk) return hash_func.hexdigest() def calculate_perceptual_hash(file_path): try: with Image.open(file_path) as img: return imagehash.phash(img) except Exception as e: print(f"[WARN] Failed to compute perceptual hash for {file_path}: {e}") return None def is_similar_image(phash, width, height, existing_phashes, threshold=2): for existing, ew, eh in existing_phashes: try: distance = phash - existing if distance <= threshold and ew >= width and eh >= height: return True except Exception as e: print(f"[WARN] Failed pHash comparison: {e}") return False def extract_metadata(file_path): 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() if ext in ('.mp4', '.mov'): # Handle video dimensions using ffprobe try: import subprocess, json 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) data = json.loads(result.stdout) stream = data.get("streams", [{}])[0] metadata['width'] = stream.get('width') metadata['height'] = stream.get('height') metadata['format'] = 'mp4' except Exception as e: print(f"[WARN] Failed to extract video metadata: {file_path} – {e}") return metadata try: metadata['file_size'] = os.path.getsize(file_path) with Image.open(file_path) as img: metadata['width'], metadata['height'] = img.size metadata['format'] = img.format except Exception as e: ext = os.path.splitext(file_path)[1] mime_type, _ = mimetypes.guess_type(file_path) print(f"[WARN] Failed to read image metadata: {file_path}") print(f"[WARN] Reason: {e}") print(f"[INFO] File extension: {ext}, MIME type: {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 as e: print(f"[WARN] Failed to read EXIF data: {file_path} – {e}") return metadata