initial functionality and styling pass
This commit is contained in:
Binary file not shown.
@@ -0,0 +1,129 @@
|
||||
# app/utils/image_importer.py
|
||||
|
||||
from datetime import datetime
|
||||
import os, shutil, hashlib
|
||||
from PIL import Image
|
||||
import exifread
|
||||
import mimetypes
|
||||
import imagehash
|
||||
|
||||
from app import db
|
||||
from app.models import ImageRecord
|
||||
|
||||
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)
|
||||
|
||||
record = ImageRecord(
|
||||
filename=filename,
|
||||
filepath=dest_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()
|
||||
return f"Imported {len(imported)} images."
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user