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

tuning job log views and deep scan archive processing.

Browse Source
This commit is contained in:
Bryan Van Deusen
2026-02-02 18:48:01 -05:00
parent 042a69f9c3
commit 09883960d4
9 changed files with 1078 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/import_file.py
+150 -42
View File
@@ -51,6 +51,58 @@ def _get_cached_settings() -> dict:
return _settings_cache
def _find_best_sidecar(src_path: str, archive_path: str = None,
archive_sidecar_path: str = None) -> str | None:
"""
Find the best sidecar JSON for a file, checking multiple locations.
For files extracted from archives, checks:
1. Per-file sidecar next to extracted file (temp dir)
2. Per-file sidecar next to original archive (by filename match)
3. Archive's sidecar as fallback (applies to all files in archive)
Args:
src_path: Path to the source file
archive_path: Path to original archive (if extracted from archive)
archive_sidecar_path: Path to archive's sidecar JSON
Returns:
Path to the best sidecar JSON found, or None
"""
from app.utils.metadata_enrichment import find_sidecar_json
import re
# First try: per-file sidecar next to the source file
sidecar = find_sidecar_json(src_path)
if sidecar:
return sidecar
# Second try: per-file sidecar next to original archive
if archive_path:
from pathlib import Path
archive_dir = Path(archive_path).parent
filename = Path(src_path).name
stem = Path(src_path).stem
# Check for exact stem match
per_file_json = archive_dir / f"{stem}.json"
if per_file_json.exists():
return str(per_file_json)
# Check for stem without numeric prefix
stem_no_prefix = re.sub(r'^\d+_', '', stem)
if stem_no_prefix != stem:
alt_json = archive_dir / f"{stem_no_prefix}.json"
if alt_json.exists():
return str(alt_json)
# Third try: archive-level sidecar (fallback for all files)
if archive_sidecar_path and os.path.exists(archive_sidecar_path):
return archive_sidecar_path
return None
@celery.task(bind=True, name='app.tasks.import_file.import_media_file',
max_retries=3, default_retry_delay=60)
def import_media_file(self, task_id: int):
@@ -88,8 +140,6 @@ def import_media_file(self, task_id: int):
generate_thumbnail, generate_video_thumbnail_mirrored,
transcode_video_to_mp4, needs_transcode
)
from app.utils.metadata_enrichment import find_sidecar_json
task = ImportTask.query.get(task_id)
if not task:
log.error(f"Task {task_id} not found")
@@ -111,6 +161,10 @@ def import_media_file(self, task_id: int):
reapply_sidecars = context.get('reapply_sidecars', False)
verify_hashes = context.get('verify_hashes', False)
# Archive sidecar (for files extracted from archives)
archive_sidecar_path = context.get('archive_sidecar_path')
archive_path = context.get('archive_path')
log.info(f"Processing: {src_path} (deep_scan={is_deep_scan})")
# Load settings with caching (avoids repeated DB reads)
@@ -166,7 +220,8 @@ def import_media_file(self, task_id: int):
# In deep scan mode with reapply_sidecars, always try to enrich
# In quick mode, also try if sidecar exists
sidecar_path = find_sidecar_json(src_path)
# Uses archive sidecar as fallback for files from archives
sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path)
if sidecar_path and (is_deep_scan and reapply_sidecars or not is_deep_scan):
apply_sidecar_metadata.delay(existing.id, sidecar_path)
@@ -209,7 +264,8 @@ def import_media_file(self, task_id: int):
log.info(f"Added archive tag {archive_tag_id} to similar image {similar_id}")
# Still try to enrich the existing larger image with sidecar metadata
sidecar_path = find_sidecar_json(src_path)
# Uses archive sidecar as fallback for files from archives
sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path)
if sidecar_path and similar_id:
apply_sidecar_metadata.delay(similar_id, sidecar_path)
return _skip_task(task, 'Similar larger image exists', result_image_id=similar_id)
@@ -220,7 +276,8 @@ def import_media_file(self, task_id: int):
if old_record:
return _supersede_existing(
task, old_record, src_path, file_hash, phash,
metadata, artist, dest_dir, filename
metadata, artist, dest_dir, filename,
archive_path, archive_sidecar_path
)
else:
# Quick mode: only compute pHash for storage, not comparison
@@ -302,7 +359,8 @@ def import_media_file(self, task_id: int):
db.session.commit()
# Queue sidecar metadata enrichment
sidecar_path = find_sidecar_json(src_path)
# Uses archive sidecar as fallback for files from archives
sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path)
if sidecar_path:
apply_sidecar_metadata.delay(record.id, sidecar_path)
@@ -341,11 +399,12 @@ def import_archive(self, task_id: int):
Steps:
1. Check if archive already processed (by hash)
2. Extract to temp directory
3. Create ImportTask for each media file inside
4. Queue import tasks for extracted files
5. Create archive tag if files were found
6. Clean up is deferred until child tasks complete
2. Find archive sidecar JSON if present
3. Extract to temp directory
4. Create ImportTask for each media file inside
5. Queue import tasks for extracted files (with archive sidecar path)
6. Create archive tag if files were found
7. Clean up is deferred until child tasks complete
Args:
task_id: ID of the ImportTask record
@@ -354,8 +413,9 @@ def import_archive(self, task_id: int):
dict with status and counts
"""
from app.utils.image_importer import (
calculate_hash, compute_next_archive_tag_name, extract_archive_resilient, ExtractError
calculate_hash, compute_archive_tag_name, extract_archive_resilient, ExtractError
)
from app.utils.metadata_enrichment import find_archive_sidecar
task = ImportTask.query.get(task_id)
if not task:
@@ -372,17 +432,31 @@ def import_archive(self, task_id: int):
artist = context.get('artist', 'unknown')
dest_dir = context.get('dest_dir', '/images')
log.info(f"Processing archive: {archive_path}")
# Deep scan options
is_deep_scan = context.get('deep_scan', False)
log.info(f"Processing archive: {archive_path} (deep_scan={is_deep_scan})")
# Find archive sidecar JSON (next to the archive file, not inside it)
# This contains post metadata from Gallery-DL for the entire archive
archive_sidecar_path = find_archive_sidecar(archive_path)
if archive_sidecar_path:
log.info(f"Found archive sidecar: {archive_sidecar_path}")
else:
log.debug(f"No sidecar found for archive: {archive_path}")
# Compute archive hash
archive_hash = calculate_hash(archive_path)
task.file_hash = archive_hash
# Check if already processed
existing = ArchiveRecord.query.filter_by(hash=archive_hash).first()
if existing:
existing_archive = ArchiveRecord.query.filter_by(hash=archive_hash).first()
if existing_archive and not is_deep_scan:
return _skip_task(task, 'Archive already imported')
if existing_archive and is_deep_scan:
log.info(f"Deep scan: re-processing already-imported archive to update metadata")
# Get archive size
archive_size = os.path.getsize(archive_path)
@@ -415,39 +489,68 @@ def import_archive(self, task_id: int):
log.info(f"Found {len(media_files)} media files in archive")
# Create archive tag
tag_name = compute_next_archive_tag_name(artist)
archive_tag = Tag.query.filter_by(name=tag_name).first()
if not archive_tag:
archive_tag = Tag(name=tag_name, kind='archive')
db.session.add(archive_tag)
db.session.flush()
# Get or create archive tag
# For deep scan of existing archives, reuse the existing tag
if existing_archive and existing_archive.tag_id:
archive_tag = Tag.query.get(existing_archive.tag_id)
if not archive_tag:
# Tag was deleted, recreate it
archive_filename = os.path.basename(archive_path)
tag_name = compute_archive_tag_name(artist, archive_filename)
archive_tag = Tag(name=tag_name, kind='archive')
db.session.add(archive_tag)
db.session.flush()
else:
# New archive - create tag using artist:archive_name format
archive_filename = os.path.basename(archive_path)
tag_name = compute_archive_tag_name(artist, archive_filename)
archive_tag = Tag.query.filter_by(name=tag_name).first()
if not archive_tag:
archive_tag = Tag(name=tag_name, kind='archive')
db.session.add(archive_tag)
db.session.flush()
# Create ArchiveRecord
archive_record = ArchiveRecord(
filename=archive_path,
file_size=archive_size,
hash=archive_hash,
artist=artist,
tag_id=archive_tag.id
)
db.session.add(archive_record)
db.session.commit()
# Create ArchiveRecord only if it doesn't exist
if not existing_archive:
archive_record = ArchiveRecord(
filename=archive_path,
file_size=archive_size,
hash=archive_hash,
artist=artist,
tag_id=archive_tag.id
)
db.session.add(archive_record)
db.session.commit()
else:
# Deep scan - commit any tag changes
db.session.commit()
# Queue tasks for each media file
queued_count = 0
for media_path in media_files:
# Build child context with archive info
child_context = {
'artist': artist,
'dest_dir': dest_dir,
'archive_task_id': task.id,
'archive_tag_id': archive_tag.id,
'temp_dir': tmpdir,
'archive_path': archive_path, # Original archive location (for sidecar lookups)
}
# Pass archive sidecar path so child tasks can use it as fallback
if archive_sidecar_path:
child_context['archive_sidecar_path'] = archive_sidecar_path
# Pass through deep scan flags from parent task
for flag in ('deep_scan', 'reapply_sidecars', 'regenerate_thumbnails', 'verify_hashes'):
if context.get(flag):
child_context[flag] = context[flag]
child_task = ImportTask(
source_path=media_path,
task_type='import_image',
batch_id=task.batch_id,
context={
'artist': artist,
'dest_dir': dest_dir,
'archive_task_id': task.id,
'archive_tag_id': archive_tag.id,
'temp_dir': tmpdir
},
context=child_context,
status='pending'
)
db.session.add(child_task)
@@ -463,6 +566,8 @@ def import_archive(self, task_id: int):
task.context['temp_dir'] = tmpdir
task.context['archive_tag_id'] = archive_tag.id
task.context['child_count'] = queued_count
if archive_sidecar_path:
task.context['archive_sidecar_path'] = archive_sidecar_path
task.status = 'complete'
task.completed_at = datetime.utcnow()
@@ -585,7 +690,9 @@ def _regenerate_thumbnail(record: ImageRecord) -> bool:
def _supersede_existing(task: ImportTask, old_record: ImageRecord, src_path: str,
file_hash: str, phash, metadata: dict, artist: str,
dest_dir: str, filename: str) -> dict:
dest_dir: str, filename: str,
archive_path: str = None,
archive_sidecar_path: str = None) -> dict:
"""
Replace an existing smaller image with a larger version.
@@ -595,14 +702,15 @@ def _supersede_existing(task: ImportTask, old_record: ImageRecord, src_path: str
build_hashed_dest_path, generate_thumbnail, generate_video_thumbnail_mirrored,
supersede_image
)
from app.utils.metadata_enrichment import find_sidecar_json, enrich_from_sidecar
from app.utils.metadata_enrichment import enrich_from_sidecar
from app.tasks.sidecar import apply_sidecar_metadata
log.info(f"Superseding {old_record.filename} ({old_record.width}x{old_record.height}) "
f"with {filename} ({metadata['width']}x{metadata['height']})")
# Enrich metadata from sidecar if available
sidecar_path = find_sidecar_json(src_path)
# Uses archive sidecar as fallback for files from archives
sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path)
if sidecar_path:
enriched = enrich_from_sidecar(sidecar_path, {'taken_at': metadata.get('taken_at')})
if enriched.get('taken_at'):
app/tasks/scan.py
+102
View File
@@ -633,3 +633,105 @@ def update_batch_stats(batch_id: int):
f"{batch.skipped_count} skipped, {batch.failed_count} failed")
db.session.commit()
@celery.task(name='app.tasks.scan.update_system_stats')
def update_system_stats():
"""
Compute and cache system-wide statistics for the dashboard.
Called periodically by Celery Beat to avoid expensive queries on page load.
Stores results in AppSettings as JSON for quick retrieval.
Stats computed:
- Total images and storage usage
- Tag counts by kind
- Pending files/archives in import folder
Returns:
dict with computed stats
"""
import json
from sqlalchemy import func
from app.models import ImageRecord, Tag, AppSettings
log.info("Updating system stats cache...")
try:
# Database stats
total_images = ImageRecord.query.count()
total_tags = Tag.query.count()
# Tag kind breakdown
tag_kinds = dict(
db.session.query(Tag.kind, func.count(Tag.id))
.group_by(Tag.kind).all()
)
# Storage stats (sum of file_size column)
storage_result = db.session.query(func.sum(ImageRecord.file_size)).scalar()
total_storage_bytes = storage_result or 0
# Import folder stats
import_dir = '/import'
pending_files = 0
pending_archives = 0
if os.path.exists(import_dir):
for artist_dir in os.listdir(import_dir):
artist_path = os.path.join(import_dir, artist_dir)
if not os.path.isdir(artist_path):
continue
for root, _, files in os.walk(artist_path):
for f in files:
lower = f.lower()
if lower.endswith(ALLOWED_MEDIA_EXTS):
pending_files += 1
elif lower.endswith(ALLOWED_ARCHIVE_EXTS):
pending_archives += 1
# Format storage for display
storage_human = _format_bytes(total_storage_bytes)
# Build stats dict
stats = {
'images': {
'total': total_images,
'storage_bytes': total_storage_bytes,
'storage_human': storage_human
},
'tags': {
'total': total_tags,
'by_kind': tag_kinds
},
'import_folder': {
'pending_files': pending_files,
'pending_archives': pending_archives
},
'updated_at': datetime.utcnow().isoformat()
}
# Store in AppSettings
setting = AppSettings.query.filter_by(key='system_stats_cache').first()
if not setting:
setting = AppSettings(key='system_stats_cache')
db.session.add(setting)
setting.value = json.dumps(stats)
db.session.commit()
log.info(f"System stats updated: {total_images} images, {total_tags} tags, "
f"{storage_human} storage, {pending_files + pending_archives} pending")
return stats
except Exception as e:
log.error(f"Failed to update system stats: {e}", exc_info=True)
raise
def _format_bytes(size_bytes: int) -> str:
"""Format bytes to human readable string."""
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if size_bytes < 1024:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024
return f"{size_bytes:.1f} PB"