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

additional tag edditing functionality. and delete by tag. added import tuning options to settings

Browse Source
This commit is contained in:
Bryan Van Deusen
2026-01-18 16:52:32 -05:00
parent 4c56b73bc9
commit b3331bad76
7 changed files with 700 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/image_importer.py
+167 -9
View File
@@ -44,6 +44,68 @@ ALLOWED_ARCHIVE_EXTS = (
".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz"
)
# Import filter settings (loaded from file or env)
IMPORT_SETTINGS_PATH = "/import/settings.json"
IMPORT_STATS_PATH = "/import/stats.json"
def load_import_settings() -> dict:
"""Load import filter settings from file or environment."""
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")),
}
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}")
return defaults
def save_import_stats(stats: dict) -> None:
"""Save import statistics to file."""
try:
with open(IMPORT_STATS_PATH, "w") as f:
json.dump(stats, f, indent=2)
except Exception as e:
print(f"[WARN] Failed to save import stats: {e}")
def is_mostly_transparent(file_path: str, threshold: float = 0.9) -> bool:
"""
Check if an image is mostly transparent (> threshold % of pixels are transparent).
Returns False for images without alpha channel.
"""
try:
with Image.open(file_path) as img:
if img.mode not in ("RGBA", "LA", "P"):
return False
# Convert palette images with transparency
if img.mode == "P" and "transparency" in img.info:
img = img.convert("RGBA")
elif img.mode == "LA":
img = img.convert("RGBA")
elif img.mode != "RGBA":
return False
# Get alpha channel and count transparent pixels
alpha = img.split()[-1]
total_pixels = alpha.size[0] * alpha.size[1]
# Count pixels where alpha < 128 (more than 50% transparent)
transparent_count = sum(1 for p in alpha.getdata() if p < 128)
transparency_ratio = transparent_count / total_pixels
return transparency_ratio > threshold
except Exception as e:
print(f"[WARN] Failed to check transparency for {file_path}: {e}")
return False
# 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)
@@ -165,6 +227,28 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
# Cleanup any stale extraction dirs from previous runs/crashes
cleanup_stale_tempdirs()
# Load import filter settings
settings = load_import_settings()
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)
phash_threshold = settings.get("phash_threshold", 2)
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}")
# Statistics tracking
stats = {
"total_processed": 0,
"total_imported": 0,
"filtered_by_dimension": 0,
"filtered_by_transparency": 0,
"filtered_by_duplicate": 0,
}
imported: list[str] = []
batch_size = 10
batch_counter = 0
@@ -205,7 +289,9 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
archive_path=full_path,
dest_dir=dest_dir,
artist=artist_dir,
existing_phashes=existing_phashes
existing_phashes=existing_phashes,
settings=settings,
stats=stats
)
imported.append(f"[archive]{file}:{count}")
except Exception as e:
@@ -216,14 +302,37 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
if not lower.endswith(ALLOWED_MEDIA_EXTS):
continue
stats["total_processed"] += 1
# Apply dimension filter
if min_width > 0 or min_height > 0:
md = extract_metadata(full_path)
if md.get("width") and md.get("height"):
if md["width"] < min_width or md["height"] < min_height:
print(f"[SKIP] Too small ({md['width']}x{md['height']}): {file}")
stats["filtered_by_dimension"] += 1
continue
# Apply transparency filter
if skip_transparent and lower.endswith((".png", ".gif", ".webp")):
if is_mostly_transparent(full_path, transparency_threshold):
print(f"[SKIP] Mostly transparent: {file}")
stats["filtered_by_transparency"] += 1
continue
added, phash, _rec = import_single_file(
src_path=full_path,
dest_dir=dest_dir,
artist=artist_dir,
existing_phashes=existing_phashes
existing_phashes=existing_phashes,
phash_threshold=phash_threshold
)
if added:
imported.append(file)
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)
@@ -242,17 +351,32 @@ def import_images_task(source_dir: str, dest_dir: str) -> str:
db.session.commit()
print(f"[INFO] Committed final batch of {batch_counter} items.")
summary = f"Imported {len(imported)} items."
# 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."
print(f"[INFO] {summary}")
return summary
def process_archive(archive_path, dest_dir, artist, existing_phashes):
def process_archive(archive_path, dest_dir, artist, existing_phashes, settings=None, stats=None):
"""
Extracts archive to temp dir, imports/tag existing images, and records ArchiveRecord.
Creates an archive:* tag ONLY if at least one image was imported or tagged.
Returns number of items imported or tagged.
"""
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}
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)
phash_threshold = settings.get("phash_threshold", 2)
archive_size = os.path.getsize(archive_path)
archive_hash = calculate_hash(archive_path)
@@ -274,21 +398,43 @@ def process_archive(archive_path, dest_dir, artist, existing_phashes):
# Import / collect all media
for root, _, files in os.walk(tmpdir):
for file in files:
if not file.lower().endswith(ALLOWED_MEDIA_EXTS):
lower = file.lower()
if not lower.endswith(ALLOWED_MEDIA_EXTS):
continue
src_path = os.path.join(root, file)
stats["total_processed"] += 1
# Apply dimension filter
if min_width > 0 or min_height > 0:
md = extract_metadata(src_path)
if md.get("width") and md.get("height"):
if md["width"] < min_width or md["height"] < min_height:
print(f"[SKIP] Too small ({md['width']}x{md['height']}): {file}")
stats["filtered_by_dimension"] += 1
continue
# Apply transparency filter
if skip_transparent and lower.endswith((".png", ".gif", ".webp")):
if is_mostly_transparent(src_path, transparency_threshold):
print(f"[SKIP] Mostly transparent: {file}")
stats["filtered_by_transparency"] += 1
continue
added, _phash, rec = 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
commit=False,
phash_threshold=phash_threshold
)
if rec is not None:
touched_records.append(rec)
if added:
imported_or_tagged += 1
stats["total_imported"] += 1
elif rec is None and _phash is None:
stats["filtered_by_duplicate"] += 1
# Only now: create/attach the archive tag if we actually touched images
if touched_records:
@@ -349,7 +495,7 @@ 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):
def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phashes, commit: bool = True, phash_threshold: int = 2):
"""
Import a single media file. If an equivalent file already exists, return that
record so callers can tag it (no duplicate import).
@@ -378,8 +524,8 @@ def import_single_file(src_path: str, dest_dir: str, artist: str, existing_phash
return (False, None, None)
phash = calculate_perceptual_hash(src_path)
if phash and is_similar_image(phash, metadata["width"], metadata["height"], existing_phashes):
print(f"[SKIP] {original_name} is visually similar to an existing larger image.")
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)
# Copy into /images/<artist>/<base>__<hash[:10]><ext>
@@ -574,6 +720,12 @@ def extract_metadata(file_path: str) -> dict:
metadata["file_size"] = os.path.getsize(file_path)
ext = os.path.splitext(file_path)[1].lower()
# Get file modification time as fallback for taken_at
try:
file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
except Exception:
file_mtime = None
# Video: use ffprobe
if ext in (".mp4", ".mov"):
try:
@@ -589,6 +741,8 @@ def extract_metadata(file_path: str) -> dict:
metadata["width"] = stream.get("width")
metadata["height"] = stream.get("height")
metadata["format"] = ext.lstrip(".")
# Use file mtime for videos since they don't have EXIF
metadata["taken_at"] = file_mtime
except Exception as e:
print(f"[WARN] Failed to extract video metadata: {file_path} – {e}")
return metadata
@@ -612,6 +766,10 @@ def extract_metadata(file_path: str) -> dict:
except Exception:
pass
# Fallback to file modification time if no EXIF date
if metadata["taken_at"] is None:
metadata["taken_at"] = file_mtime
return metadata
def _has_alpha(img: Image.Image) -> bool: