bvandeusen/imagerepo
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed pagination issues and thumbnail generation issues

Browse Source
This commit is contained in:
Bryan Van Deusen
2025-08-16 10:25:16 -04:00
parent 3ff34ec9c2
commit a77724dccc
5 changed files with 502 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/image_importer.py
+184 -31
View File
@@ -15,13 +15,19 @@ import tempfile
import time
import uuid
from PIL import Image
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
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
@@ -250,15 +256,13 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes):
archive_size = os.path.getsize(archive_path)
archive_hash = calculate_hash(archive_path)
# Skip if archive already processed
# 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
tag_name = f"archive:{artist}/{os.path.basename(archive_path)}"
tmpdir = tempfile.mkdtemp(prefix="extract_")
touched_records = [] # ImageRecord objects we imported or matched
imported_or_tagged = 0
created_tag = None
@@ -288,6 +292,8 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes):
# 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')
@@ -301,13 +307,13 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes):
newly_tagged += 1
imported_or_tagged += newly_tagged
# Record this archive as processed.
# 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 # None if we had no media
tag=created_tag
)
db.session.add(arch)
db.session.commit()
@@ -430,27 +436,89 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
# Helpers: Thumbnails, metadata, hashing, similarity
# =============================================================================
def generate_thumbnail(image_path: str, size: tuple[int, int] = (400, 400), overwrite: bool = False) -> str:
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/<artist>/<unique_filename>.
(unique_filename includes the hash suffix, so no collisions)
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`.
Returns the absolute path to the thumbnail.
"""
images_root = Path("/images").resolve()
image_path = Path(image_path).resolve()
rel_path = image_path.relative_to(images_root) # raises if not under /images
thumb_path = images_root / "thumbs" / rel_path
thumb_path = thumb_path.with_suffix(".jpg") if thumb_path.suffix.lower() not in (".jpg", ".jpeg") else thumb_path
thumb_path.parent.mkdir(parents=True, exist_ok=True)
try:
rel_path = image_path.relative_to(images_root)
except ValueError:
raise ValueError(f"{image_path} is not under {images_root}")
if not overwrite and thumb_path.exists():
return str(thumb_path)
# 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
img.thumbnail(size)
img.save(thumb_path)
return str(thumb_path)
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:
@@ -458,21 +526,11 @@ def generate_video_thumbnail_mirrored(video_path: str, time_position: str = "00:
Mirror the stored (hashed) video path under /images/thumbs/... and save as .jpg.
"""
images_root = Path("/images").resolve()
video_path = Path(video_path).resolve()
rel_path = video_path.relative_to(images_root) # raises if not under /images
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")
thumb_path.parent.mkdir(parents=True, exist_ok=True)
return generate_video_thumbnail(str(video_path_p), str(thumb_path), time_position=time_position)
cmd = [
"ffmpeg", "-i", str(video_path), "-ss", time_position,
"-vframes", "1", "-vf", f"scale={THUMB_SIZE[0]}:-1", str(thumb_path)
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return str(thumb_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: str) -> str:
@@ -556,6 +614,33 @@ def extract_metadata(file_path: str) -> dict:
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
@@ -596,4 +681,72 @@ def is_first_volume(path: str) -> bool:
if name.endswith((".zip", ".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz")):
return True
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