fixed some errors with archive handling
This commit is contained in:
Bryan Van Deusen
+23
@@ -1434,6 +1434,29 @@ def clear_completed_tasks():
|
||||
return jsonify({'ok': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@main.route('/api/import/clear-queue', methods=['POST'])
|
||||
def clear_import_queue():
|
||||
"""
|
||||
Clear pending and queued import tasks from the queue.
|
||||
This stops tasks that haven't started processing yet.
|
||||
"""
|
||||
try:
|
||||
# Find and delete pending/queued tasks
|
||||
deleted = ImportTask.query.filter(
|
||||
ImportTask.status.in_(['pending', 'queued'])
|
||||
).delete(synchronize_session=False)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({
|
||||
'ok': True,
|
||||
'deleted': deleted
|
||||
})
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({'ok': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@main.route('/api/import/task-stats')
|
||||
def import_task_stats():
|
||||
"""
|
||||
|
||||
+211
-42
@@ -452,7 +452,7 @@ def import_archive(self, task_id: int):
|
||||
# Check if already processed
|
||||
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')
|
||||
return _skip_task(task, 'Archive already imported (use deep scan to reprocess)')
|
||||
|
||||
if existing_archive and is_deep_scan:
|
||||
log.info(f"Deep scan: re-processing already-imported archive to update metadata")
|
||||
@@ -525,47 +525,46 @@ def import_archive(self, task_id: int):
|
||||
# Deep scan - commit any tag changes
|
||||
db.session.commit()
|
||||
|
||||
# Queue tasks for each media file
|
||||
queued_count = 0
|
||||
# Process each media file immediately (synchronous processing)
|
||||
# This ensures all files from the archive are processed together
|
||||
# with their archive tag and sidecar metadata applied immediately
|
||||
imported_count = 0
|
||||
skipped_count = 0
|
||||
failed_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
|
||||
try:
|
||||
result = _process_archive_file(
|
||||
src_path=media_path,
|
||||
artist=artist,
|
||||
dest_dir=dest_dir,
|
||||
archive_tag=archive_tag,
|
||||
archive_path=archive_path,
|
||||
archive_sidecar_path=archive_sidecar_path,
|
||||
is_deep_scan=is_deep_scan,
|
||||
reapply_sidecars=context.get('reapply_sidecars', False),
|
||||
regenerate_thumbnails=context.get('regenerate_thumbnails', False)
|
||||
)
|
||||
if result['status'] == 'imported' or result['status'] == 'superseded':
|
||||
imported_count += 1
|
||||
elif result['status'] == 'skipped':
|
||||
skipped_count += 1
|
||||
else:
|
||||
failed_count += 1
|
||||
except Exception as e:
|
||||
log.error(f"Failed to process {media_path}: {e}")
|
||||
failed_count += 1
|
||||
|
||||
# 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]
|
||||
# Clean up temp directory immediately since we're done processing
|
||||
if tmpdir and os.path.exists(tmpdir):
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
child_task = ImportTask(
|
||||
source_path=media_path,
|
||||
task_type='import_image',
|
||||
batch_id=task.batch_id,
|
||||
context=child_context,
|
||||
status='pending'
|
||||
)
|
||||
db.session.add(child_task)
|
||||
db.session.flush()
|
||||
|
||||
result = import_media_file.delay(child_task.id)
|
||||
child_task.celery_task_id = result.id
|
||||
child_task.status = 'queued'
|
||||
queued_count += 1
|
||||
|
||||
# Store child task info for cleanup tracking
|
||||
# Update task context with results
|
||||
task.context = task.context or {}
|
||||
task.context['temp_dir'] = tmpdir
|
||||
task.context['archive_tag_id'] = archive_tag.id
|
||||
task.context['child_count'] = queued_count
|
||||
task.context['imported_count'] = imported_count
|
||||
task.context['skipped_count'] = skipped_count
|
||||
task.context['failed_count'] = failed_count
|
||||
if archive_sidecar_path:
|
||||
task.context['archive_sidecar_path'] = archive_sidecar_path
|
||||
|
||||
@@ -573,14 +572,18 @@ def import_archive(self, task_id: int):
|
||||
task.completed_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
log.info(f"Archive processed: queued {queued_count} files from {archive_path}")
|
||||
log.info(f"Archive processed: {imported_count} imported, {skipped_count} skipped, {failed_count} failed from {archive_path}")
|
||||
|
||||
# Note: temp directory cleanup happens via a separate periodic task
|
||||
# or when all child tasks complete
|
||||
# Update batch stats
|
||||
if task.batch_id:
|
||||
from app.tasks.scan import update_batch_stats
|
||||
update_batch_stats.delay(task.batch_id)
|
||||
|
||||
return {
|
||||
'status': 'extracted',
|
||||
'queued_files': queued_count,
|
||||
'status': 'processed',
|
||||
'imported': imported_count,
|
||||
'skipped': skipped_count,
|
||||
'failed': failed_count,
|
||||
'archive_tag': tag_name
|
||||
}
|
||||
|
||||
@@ -592,6 +595,172 @@ def import_archive(self, task_id: int):
|
||||
return _fail_task(task, str(e))
|
||||
|
||||
|
||||
def _process_archive_file(src_path: str, artist: str, dest_dir: str,
|
||||
archive_tag: Tag, archive_path: str,
|
||||
archive_sidecar_path: str = None,
|
||||
is_deep_scan: bool = False,
|
||||
reapply_sidecars: bool = False,
|
||||
regenerate_thumbnails: bool = False) -> dict:
|
||||
"""
|
||||
Process a single file from an archive synchronously.
|
||||
|
||||
This is called directly from import_archive for each extracted file,
|
||||
ensuring all files from an archive are processed together with their
|
||||
archive tag and sidecar metadata applied immediately.
|
||||
|
||||
Returns:
|
||||
dict with 'status' key: 'imported', 'skipped', 'superseded', or 'failed'
|
||||
"""
|
||||
from app.utils.image_importer import (
|
||||
extract_metadata, calculate_hash, calculate_perceptual_hash,
|
||||
is_mostly_transparent, is_mostly_single_color,
|
||||
build_hashed_dest_path, generate_thumbnail, generate_video_thumbnail_mirrored,
|
||||
transcode_video_to_mp4, needs_transcode
|
||||
)
|
||||
from app.utils.metadata_enrichment import load_sidecar_metadata, enrich_from_sidecar
|
||||
from app.tasks.sidecar import apply_sidecar_metadata
|
||||
|
||||
filename = os.path.basename(src_path)
|
||||
lower = filename.lower()
|
||||
|
||||
# Load settings (use cached settings from module)
|
||||
settings = _get_cached_settings()
|
||||
|
||||
# Extract metadata first for filtering
|
||||
metadata = extract_metadata(src_path)
|
||||
if not metadata.get('width') or not metadata.get('height'):
|
||||
log.debug(f"Skipping {filename}: Missing dimension data")
|
||||
return {'status': 'skipped', 'reason': 'Missing dimension data'}
|
||||
|
||||
# Apply dimension filter
|
||||
min_width = settings.get('min_width', 0)
|
||||
min_height = settings.get('min_height', 0)
|
||||
if min_width > 0 and metadata['width'] < min_width:
|
||||
log.debug(f"Skipping {filename}: Too small")
|
||||
return {'status': 'skipped', 'reason': 'Too small'}
|
||||
if min_height > 0 and metadata['height'] < min_height:
|
||||
log.debug(f"Skipping {filename}: Too small")
|
||||
return {'status': 'skipped', 'reason': 'Too small'}
|
||||
|
||||
# Apply transparency filter
|
||||
if settings.get('skip_transparent') and lower.endswith(('.png', '.gif', '.webp')):
|
||||
threshold = settings.get('transparency_threshold', 0.9)
|
||||
if is_mostly_transparent(src_path, threshold):
|
||||
log.debug(f"Skipping {filename}: Mostly transparent")
|
||||
return {'status': 'skipped', 'reason': 'Mostly transparent'}
|
||||
|
||||
# Apply single-color filter
|
||||
if settings.get('skip_single_color', True) and lower.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp')):
|
||||
if is_mostly_single_color(src_path,
|
||||
settings.get('single_color_threshold', 0.95),
|
||||
settings.get('single_color_tolerance', 30)):
|
||||
log.debug(f"Skipping {filename}: Mostly single color")
|
||||
return {'status': 'skipped', 'reason': 'Mostly single color'}
|
||||
|
||||
# Compute content hash
|
||||
file_hash = calculate_hash(src_path)
|
||||
|
||||
# Find best sidecar for this file
|
||||
sidecar_path = _find_best_sidecar(src_path, archive_path, archive_sidecar_path)
|
||||
|
||||
# Check for exact duplicate by hash
|
||||
existing = ImageRecord.query.filter_by(hash=file_hash).first()
|
||||
if existing:
|
||||
# Add archive tag to existing image
|
||||
if archive_tag and archive_tag not in existing.tags:
|
||||
existing.tags.append(archive_tag)
|
||||
db.session.commit()
|
||||
log.debug(f"Added archive tag to existing image {existing.id}")
|
||||
|
||||
# Apply sidecar metadata
|
||||
if sidecar_path and (is_deep_scan and reapply_sidecars or not is_deep_scan):
|
||||
apply_sidecar_metadata.delay(existing.id, sidecar_path)
|
||||
|
||||
# Regenerate thumbnail if requested
|
||||
if is_deep_scan and regenerate_thumbnails:
|
||||
_regenerate_thumbnail(existing)
|
||||
|
||||
return {'status': 'skipped', 'reason': 'Duplicate', 'image_id': existing.id}
|
||||
|
||||
# Compute perceptual hash
|
||||
phash = calculate_perceptual_hash(src_path)
|
||||
|
||||
# Copy to destination
|
||||
target_dir = os.path.join(dest_dir, artist)
|
||||
os.makedirs(target_dir, exist_ok=True)
|
||||
|
||||
# Transcode video if needed
|
||||
transcoded_temp = None
|
||||
actual_src = src_path
|
||||
actual_filename = filename
|
||||
|
||||
if needs_transcode(src_path):
|
||||
log.info(f"Transcoding video: {filename}")
|
||||
transcoded_temp = transcode_video_to_mp4(src_path)
|
||||
if transcoded_temp:
|
||||
actual_src = transcoded_temp
|
||||
actual_filename = os.path.splitext(filename)[0] + '.mp4'
|
||||
file_hash = calculate_hash(transcoded_temp)
|
||||
log.info(f"Transcoded successfully: {actual_filename}")
|
||||
else:
|
||||
log.warning(f"Transcode failed for {filename}, importing original")
|
||||
|
||||
dest_path = build_hashed_dest_path(target_dir, actual_filename, file_hash)
|
||||
shutil.copy2(actual_src, dest_path)
|
||||
|
||||
# Clean up transcoded temp file
|
||||
if transcoded_temp and os.path.exists(transcoded_temp):
|
||||
os.remove(transcoded_temp)
|
||||
|
||||
# Generate thumbnail
|
||||
try:
|
||||
if dest_path.lower().endswith(('.mp4', '.mov', '.webm', '.avi', '.mkv')):
|
||||
thumb_path = generate_video_thumbnail_mirrored(dest_path)
|
||||
else:
|
||||
thumb_path = generate_thumbnail(dest_path)
|
||||
except Exception as e:
|
||||
log.warning(f"Thumbnail generation failed for {actual_filename}: {e}")
|
||||
thumb_path = None
|
||||
|
||||
# Create ImageRecord
|
||||
record = ImageRecord(
|
||||
filename=actual_filename,
|
||||
filepath=dest_path,
|
||||
thumb_path=thumb_path,
|
||||
hash=file_hash,
|
||||
perceptual_hash=str(phash) if phash else None,
|
||||
file_size=os.path.getsize(dest_path),
|
||||
width=metadata['width'],
|
||||
height=metadata['height'],
|
||||
format=metadata['format'] if not needs_transcode(src_path) else 'mp4',
|
||||
camera_model=metadata.get('camera_model'),
|
||||
taken_at=metadata.get('taken_at'),
|
||||
imported_at=datetime.utcnow()
|
||||
)
|
||||
|
||||
# Add artist tag
|
||||
_add_artist_tag(record, artist)
|
||||
|
||||
# Add archive tag
|
||||
if archive_tag:
|
||||
record.tags.append(archive_tag)
|
||||
|
||||
db.session.add(record)
|
||||
db.session.commit()
|
||||
|
||||
# Apply sidecar metadata asynchronously
|
||||
if sidecar_path:
|
||||
apply_sidecar_metadata.delay(record.id, sidecar_path)
|
||||
|
||||
log.debug(f"Imported from archive: {actual_filename} -> {dest_path}")
|
||||
|
||||
return {
|
||||
'status': 'imported',
|
||||
'image_id': record.id,
|
||||
'dest_path': dest_path
|
||||
}
|
||||
|
||||
|
||||
def _fail_task(task: ImportTask, error: str, **kwargs) -> dict:
|
||||
"""Mark task as failed with error message."""
|
||||
task.status = 'failed'
|
||||
|
||||
@@ -99,6 +99,7 @@
|
||||
<button id="triggerImportBtn" class="btn primary-btn">Quick Scan</button>
|
||||
<button id="triggerDeepScanBtn" class="btn warning-btn" title="Full reprocessing: re-extracts metadata, runs pHash comparison, merges archive tags">Deep Scan</button>
|
||||
<button id="retryFailedBtn" class="btn warning-btn">Retry Failed Tasks</button>
|
||||
<button id="clearQueueBtn" class="btn danger-btn" title="Remove all pending and queued tasks from the queue">Clear Queue</button>
|
||||
</div>
|
||||
|
||||
<!-- Task Cleanup Controls -->
|
||||
@@ -1090,6 +1091,44 @@ document.addEventListener('DOMContentLoaded', () => {
|
||||
}
|
||||
});
|
||||
|
||||
// Clear Queue (pending/queued tasks)
|
||||
document.getElementById('clearQueueBtn').addEventListener('click', async () => {
|
||||
const pendingCount = parseInt(document.getElementById('queuePending').textContent) || 0;
|
||||
const queuedCount = parseInt(document.getElementById('queueQueued').textContent) || 0;
|
||||
const totalQueued = pendingCount + queuedCount;
|
||||
|
||||
if (totalQueued === 0) {
|
||||
alert('No pending or queued tasks to clear.');
|
||||
return;
|
||||
}
|
||||
|
||||
if (!confirm(`Are you sure you want to clear ${totalQueued} pending/queued task(s) from the import queue?\n\nThis will stop these tasks from being processed.`)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const btn = document.getElementById('clearQueueBtn');
|
||||
btn.disabled = true;
|
||||
btn.textContent = 'Clearing...';
|
||||
|
||||
try {
|
||||
const r = await fetch('/api/import/clear-queue', { method: 'POST' });
|
||||
const data = await r.json();
|
||||
|
||||
if (data.ok) {
|
||||
alert(`Cleared ${data.deleted} task(s) from the queue.`);
|
||||
loadQueueStatus();
|
||||
loadTaskList();
|
||||
} else {
|
||||
alert('Failed to clear queue: ' + (data.error || 'Unknown error'));
|
||||
}
|
||||
} catch (e) {
|
||||
alert('Failed to clear queue: ' + e.message);
|
||||
} finally {
|
||||
btn.disabled = false;
|
||||
btn.textContent = 'Clear Queue';
|
||||
}
|
||||
});
|
||||
|
||||
// Clear Tasks with age/status options
|
||||
document.getElementById('clearTasksBtn').addEventListener('click', async () => {
|
||||
const btn = document.getElementById('clearTasksBtn');
|
||||
|
||||
Reference in New Issue
Block a user