imagerepo/app/utils/image_importer.py

# 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

from app import db
from app.models import ImageRecord

THUMB_SIZE = (400, 400)

def import_images_task(source_dir, dest_dir):
    imported = []

    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', '.png', '.gif', '.bmp', '.tiff')):
                    full_path = os.path.join(root, file)
                    filename = os.path.basename(full_path)
                    file_hash = calculate_hash(full_path)

                    if ImageRecord.query.filter_by(hash=file_hash).first():
                        continue

                    metadata = extract_metadata(full_path)

                    if not metadata['width'] or not metadata['height']:
                        continue

                    phash = calculate_perceptual_hash(full_path)
                    if phash and is_similar_image(phash, metadata['width'], metadata['height']):
                        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)

                    # Generate thumbnail AFTER copying to /images
                    try:
                        thumb_path = generate_thumbnail(dest_path)
                    except Exception as e:
                        print(f"[WARN] Failed to generate thumbnail: {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)

                    db.session.commit()
                    time.sleep(1)

    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 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, threshold=2):
    for img in ImageRecord.query.filter(ImageRecord.perceptual_hash != None).all():
        try:
            existing_phash = imagehash.hex_to_hash(img.perceptual_hash)
            distance = phash - existing_phash
            if distance <= threshold:
                if img.width >= width and img.height >= 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
    }

    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