moving styling and views to be more consistent and for gallery view to be less cumbersome.
This commit is contained in:
+429
-37
@@ -50,21 +50,43 @@ IMPORT_STATS_PATH = "/import/stats.json"
|
||||
|
||||
|
||||
def load_import_settings() -> dict:
|
||||
"""Load import filter settings from file or environment."""
|
||||
"""
|
||||
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")),
|
||||
"phash_threshold": int(os.environ.get("IMPORT_PHASH_THRESHOLD", "2")),
|
||||
"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: {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
|
||||
|
||||
|
||||
@@ -106,6 +128,136 @@ def is_mostly_transparent(file_path: str, threshold: float = 0.9) -> bool:
|
||||
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)
|
||||
@@ -233,19 +385,25 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
|
||||
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"phash_threshold={phash_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,
|
||||
}
|
||||
|
||||
@@ -254,8 +412,9 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
|
||||
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)
|
||||
(imagehash.hex_to_hash(img.perceptual_hash), img.width, img.height, img.id)
|
||||
for img in ImageRecord.query.filter(ImageRecord.perceptual_hash != None).all()
|
||||
]
|
||||
|
||||
@@ -320,24 +479,34 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
|
||||
stats["filtered_by_transparency"] += 1
|
||||
continue
|
||||
|
||||
added, phash, _rec = import_single_file(
|
||||
# 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
|
||||
phash_threshold=phash_threshold,
|
||||
supersede_smaller=supersede_smaller
|
||||
)
|
||||
if added:
|
||||
imported.append(file)
|
||||
stats["total_imported"] += 1
|
||||
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
|
||||
|
||||
if phash:
|
||||
md = extract_metadata(full_path)
|
||||
if md.get("width") and md.get("height"):
|
||||
existing_phashes.append((phash, md["width"], md["height"]))
|
||||
# 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:
|
||||
@@ -354,7 +523,9 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
|
||||
# Save stats
|
||||
save_import_stats(stats)
|
||||
|
||||
summary = f"Imported {len(imported)} items. Filtered: {stats['filtered_by_dimension']} by size, {stats['filtered_by_transparency']} by transparency."
|
||||
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
|
||||
|
||||
@@ -368,14 +539,19 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes, settings=N
|
||||
if settings is None:
|
||||
settings = load_import_settings()
|
||||
if stats is None:
|
||||
stats = {"total_processed": 0, "total_imported": 0, "filtered_by_dimension": 0,
|
||||
"filtered_by_transparency": 0, "filtered_by_duplicate": 0}
|
||||
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)
|
||||
@@ -420,19 +596,30 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes, settings=N
|
||||
stats["filtered_by_transparency"] += 1
|
||||
continue
|
||||
|
||||
added, _phash, rec = import_single_file(
|
||||
# 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:<name> tag
|
||||
existing_phashes=existing_phashes,
|
||||
commit=False,
|
||||
phash_threshold=phash_threshold
|
||||
phash_threshold=phash_threshold,
|
||||
supersede_smaller=supersede_smaller
|
||||
)
|
||||
if rec is not None:
|
||||
touched_records.append(rec)
|
||||
if added:
|
||||
imported_or_tagged += 1
|
||||
stats["total_imported"] += 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
|
||||
|
||||
@@ -495,11 +682,19 @@ def build_hashed_dest_path(target_dir: str, original_name: str, content_hash: st
|
||||
i += 1
|
||||
|
||||
|
||||
def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phashes, commit: bool = True, phash_threshold: int = 2):
|
||||
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).
|
||||
Returns (added: bool, phash_or_None, record_or_None).
|
||||
|
||||
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}")
|
||||
@@ -510,7 +705,9 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
|
||||
existing = ImageRecord.query.filter_by(hash=file_hash).first()
|
||||
if existing:
|
||||
print(f"[SKIP] Duplicate by hash: {original_name}")
|
||||
return (False, None, existing)
|
||||
# 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()
|
||||
@@ -521,12 +718,67 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
|
||||
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)
|
||||
return (False, None, None, False)
|
||||
|
||||
phash = calculate_perceptual_hash(src_path)
|
||||
if phash and phash_threshold > 0 and is_similar_image(phash, metadata["width"], metadata["height"], existing_phashes, threshold=phash_threshold):
|
||||
print(f"[SKIP] {original_name} is visually similar to an existing larger image (threshold={phash_threshold}).")
|
||||
return (False, phash, None)
|
||||
|
||||
# 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/<artist>/<base>__<hash[:10]><ext>
|
||||
target_dir = os.path.join(dest_dir, artist)
|
||||
@@ -545,6 +797,7 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
|
||||
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,
|
||||
@@ -556,7 +809,7 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
|
||||
height=metadata["height"],
|
||||
format=metadata["format"],
|
||||
camera_model=metadata["camera_model"],
|
||||
taken_at=metadata["taken_at"],
|
||||
taken_at=enriched.get("taken_at") or metadata["taken_at"],
|
||||
imported_at=datetime.utcnow()
|
||||
)
|
||||
|
||||
@@ -571,11 +824,81 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
|
||||
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)
|
||||
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}")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
@@ -593,6 +916,10 @@ def generate_thumbnail(
|
||||
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()
|
||||
@@ -609,7 +936,27 @@ def generate_thumbnail(
|
||||
# Decide output extension/format
|
||||
with Image.open(image_path) as img:
|
||||
img = ImageOps.exif_transpose(img) # honor camera rotation
|
||||
img.thumbnail(size)
|
||||
|
||||
# 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)
|
||||
|
||||
@@ -687,24 +1034,69 @@ def calculate_hash(file_path: str) -> str:
|
||||
return hash_func.hexdigest()
|
||||
|
||||
|
||||
def calculate_perceptual_hash(file_path: str):
|
||||
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)
|
||||
return imagehash.phash(img, hash_size=hash_size)
|
||||
except Exception:
|
||||
# videos or unreadable images land here (phash not applicable)
|
||||
return None
|
||||
|
||||
|
||||
def is_similar_image(phash, width: int, height: int, existing_phashes, threshold: int = 2) -> bool:
|
||||
for existing, ew, eh in existing_phashes:
|
||||
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
|
||||
if distance <= threshold and ew >= width and eh >= height:
|
||||
return True
|
||||
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 False
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user