feat: structure

This commit is contained in:
2026-04-06 23:30:19 +02:00
commit 46a0d7aba8
41 changed files with 3480 additions and 0 deletions

View File

@@ -0,0 +1,31 @@
"""
Celery configuration and app initialization
"""
from celery import Celery
from app.config import settings
# Create Celery app
celery_app = Celery(
'mulita',
broker=settings.celery_broker_url,
backend=settings.celery_result_backend,
include=['app.tasks.scan', 'app.tasks.thumbs']
)
# Configure Celery
celery_app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
task_routes={
'app.tasks.thumbs.*': {'queue': 'high'},
'app.tasks.scan.*': {'queue': 'low'},
},
task_default_queue='default',
task_default_exchange='default',
task_default_exchange_type='direct',
task_default_routing_key='default',
broker_connection_retry_on_startup=True,
)

298
backend/app/tasks/scan.py Normal file
View File

@@ -0,0 +1,298 @@
"""
Celery tasks for scanning folders and indexing photos
"""
import os
import hashlib
import asyncio
from pathlib import Path
from datetime import datetime
import logging
import json
from typing import List, Dict, Optional
from celery import shared_task
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
import aiofiles
from app.database import AsyncSessionLocal
from app.models import Photo, Folder, SourceRoot
from app.config import settings
from app.tasks.thumbs import generate_thumbnails
from app.services.metadata import extract_metadata
logger = logging.getLogger(__name__)
# Supported file extensions
PHOTO_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.tiff', '.tif', '.webp', '.bmp'}
RAW_EXTENSIONS = {'.cr2', '.cr3', '.nef', '.arw', '.raf', '.dng', '.orf', '.rw2', '.pef', '.srw'}
HEIC_EXTENSIONS = {'.heic', '.heif'}
VIDEO_EXTENSIONS = {'.mp4', '.mov', '.avi', '.mkv', '.mts', '.m2ts', '.3gp', '.wmv', '.flv'}
SUPPORTED_EXTENSIONS = PHOTO_EXTENSIONS | RAW_EXTENSIONS | HEIC_EXTENSIONS | VIDEO_EXTENSIONS
def get_media_type(filepath: str) -> str:
"""Determine media type from file extension"""
ext = Path(filepath).suffix.lower()
if ext in PHOTO_EXTENSIONS:
return 'photo'
elif ext in RAW_EXTENSIONS:
return 'raw'
elif ext in HEIC_EXTENSIONS:
return 'heic'
elif ext in VIDEO_EXTENSIONS:
return 'video'
return 'unknown'
async def calculate_file_hash(filepath: str) -> str:
"""Calculate SHA-256 hash of a file"""
hash_sha256 = hashlib.sha256()
try:
async with aiofiles.open(filepath, 'rb') as f:
while chunk := await f.read(8192):
hash_sha256.update(chunk)
return hash_sha256.hexdigest()
except Exception as e:
logger.error(f"Error calculating hash for {filepath}: {e}")
return ""
@shared_task(bind=True, name='scan_folder')
def scan_folder(self, folder_path: str, source_root_id: Optional[str] = None):
"""
Scan a folder and index all photos/videos
"""
# Run async function in sync context
return asyncio.run(_scan_folder_async(folder_path, source_root_id, self))
async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], task):
"""Async implementation of folder scanning"""
logger.info(f"Starting scan of folder: {folder_path}")
async with AsyncSessionLocal() as session:
try:
# Get or create source root
if not source_root_id:
source_root = await get_or_create_source_root(session, folder_path)
source_root_id = source_root.id
# Walk the directory tree
total_files = 0
processed_files = 0
errors = []
for root, dirs, files in os.walk(folder_path):
# Get or create folder entry
folder = await get_or_create_folder(session, root, source_root_id)
# Filter supported files
supported_files = [f for f in files if Path(f).suffix.lower() in SUPPORTED_EXTENSIONS]
total_files += len(supported_files)
# Process files in batches
batch_size = settings.scanner.batch_size
for i in range(0, len(supported_files), batch_size):
batch = supported_files[i:i + batch_size]
for filename in batch:
filepath = os.path.join(root, filename)
try:
# Check if file already exists in database
existing = await session.execute(
select(Photo).where(Photo.filepath == filepath)
)
if existing.scalar_one_or_none():
logger.debug(f"File already indexed: {filepath}")
processed_files += 1
continue
# Get file stats
stat = os.stat(filepath)
# Calculate file hash for duplicate detection
file_hash = await calculate_file_hash(filepath)
# Check for duplicate by hash
duplicate = await session.execute(
select(Photo).where(Photo.file_hash == file_hash)
) if file_hash else None
# Create photo entry
photo = Photo(
filepath=filepath,
filename=filename,
folder_id=folder.id,
file_hash=file_hash,
media_type=get_media_type(filepath),
original_format=Path(filepath).suffix.upper()[1:],
file_size=stat.st_size,
taken_at=datetime.fromtimestamp(stat.st_mtime),
taken_at_source='filesystem',
is_duplicate=bool(duplicate.scalar_one_or_none() if duplicate else False),
processing_status='pending'
)
session.add(photo)
await session.flush() # Get the photo ID
# Queue thumbnail generation
generate_thumbnails.delay(photo.id)
# Queue metadata extraction
extract_metadata.delay(photo.id)
processed_files += 1
# Update progress
if processed_files % 10 == 0:
task.update_state(
state='PROGRESS',
meta={
'current': processed_files,
'total': total_files,
'folder': root
}
)
except Exception as e:
logger.error(f"Error processing file {filepath}: {e}")
errors.append({'file': filepath, 'error': str(e)})
continue
# Commit batch
await session.commit()
# Update folder scan timestamp
folder.last_scanned = datetime.utcnow()
folder.photo_count = processed_files
await session.commit()
logger.info(f"Scan complete. Processed {processed_files}/{total_files} files. Errors: {len(errors)}")
return {
'status': 'completed',
'processed': processed_files,
'total': total_files,
'errors': errors
}
except Exception as e:
logger.error(f"Scan failed: {e}")
await session.rollback()
raise
async def get_or_create_source_root(session: AsyncSession, path: str) -> SourceRoot:
"""Get or create a source root entry"""
from sqlalchemy import select
result = await session.execute(
select(SourceRoot).where(SourceRoot.path == path)
)
source_root = result.scalar_one_or_none()
if not source_root:
source_root = SourceRoot(
name=Path(path).name,
path=path
)
session.add(source_root)
await session.flush()
return source_root
async def get_or_create_folder(session: AsyncSession, path: str, source_root_id: str) -> Folder:
"""Get or create a folder entry"""
from sqlalchemy import select
result = await session.execute(
select(Folder).where(Folder.path == path)
)
folder = result.scalar_one_or_none()
if not folder:
parent_path = str(Path(path).parent)
parent = None
if parent_path != path: # Not root folder
parent_result = await session.execute(
select(Folder).where(Folder.path == parent_path)
)
parent = parent_result.scalar_one_or_none()
if parent:
parent_id = parent.id
else:
# Recursively create parent
parent = await get_or_create_folder(session, parent_path, source_root_id)
parent_id = parent.id
else:
parent_id = None
folder = Folder(
name=Path(path).name,
path=path,
parent_id=parent_id,
source_root_id=source_root_id
)
session.add(folder)
await session.flush()
return folder
@shared_task(name='scan_all_source_roots')
def scan_all_source_roots():
"""Scan all configured source roots"""
for source_root in settings.source_roots:
if os.path.exists(source_root.path):
scan_folder.delay(source_root.path)
else:
logger.warning(f"Source root path does not exist: {source_root.path}")
@shared_task(name='watch_folders')
def watch_folders():
"""
Watch folders for changes using watchfiles
This is a long-running task that monitors file system events
"""
from watchfiles import watch
paths = [sr.path for sr in settings.source_roots if os.path.exists(sr.path)]
if not paths:
logger.warning("No valid source roots to watch")
return
logger.info(f"Starting folder watcher for: {paths}")
for changes in watch(*paths):
for change_type, filepath in changes:
filepath = str(filepath)
# Check if it's a supported file type
if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS:
continue
if change_type == 'added' or change_type == 'modified':
# Queue scan for the parent folder
parent_dir = str(Path(filepath).parent)
scan_folder.delay(parent_dir)
logger.info(f"File {change_type}: {filepath}, queued scan for {parent_dir}")
elif change_type == 'deleted':
# Handle file deletion
asyncio.run(handle_file_deletion(filepath))
async def handle_file_deletion(filepath: str):
"""Handle deletion of a file from the filesystem"""
from sqlalchemy import select
async with AsyncSessionLocal() as session:
result = await session.execute(
select(Photo).where(Photo.filepath == filepath)
)
photo = result.scalar_one_or_none()
if photo:
# Mark as missing or delete from database
photo.is_trashed = True
photo.trashed_at = datetime.utcnow()
await session.commit()
logger.info(f"Marked photo as trashed: {filepath}")

277
backend/app/tasks/thumbs.py Normal file
View File

@@ -0,0 +1,277 @@
"""
Celery tasks for thumbnail generation
"""
import os
import asyncio
from pathlib import Path
import logging
from typing import Tuple, Optional
import json
from celery import shared_task
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
import pyvips
import rawpy
import imageio
from PIL import Image
from pillow_heif import register_heif_opener
import ffmpeg
from app.database import AsyncSessionLocal
from app.models import Photo
from app.config import settings
# Register HEIF opener with Pillow
register_heif_opener()
logger = logging.getLogger(__name__)
# Thumbnail sizes configuration
THUMB_SIZES = {
'small': settings.thumbnails.small,
'medium': settings.thumbnails.medium,
'large': settings.thumbnails.large
}
def get_thumb_path(photo_id: str, size: str) -> str:
"""Get the path for a thumbnail file"""
thumb_dir = f"/data/thumbs/{photo_id}"
os.makedirs(thumb_dir, exist_ok=True)
return f"{thumb_dir}/{size}.{settings.thumbnails.format}"
def process_standard_image(filepath: str) -> pyvips.Image:
"""Process standard image formats (JPEG, PNG, etc.)"""
return pyvips.Image.new_from_file(filepath, access='sequential')
def process_raw_image(filepath: str) -> pyvips.Image:
"""Process RAW image formats"""
try:
with rawpy.imread(filepath) as raw:
# Use half_size for faster processing
rgb = raw.postprocess(use_camera_wb=True, half_size=True)
# Convert numpy array to pyvips image
return pyvips.Image.new_from_array(rgb)
except Exception as e:
logger.error(f"Error processing RAW file {filepath}: {e}")
# Try to extract embedded JPEG preview
return extract_raw_preview(filepath)
def extract_raw_preview(filepath: str) -> Optional[pyvips.Image]:
"""Extract embedded JPEG preview from RAW file"""
try:
# Use exiftool to extract preview
import subprocess
import tempfile
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as tmp:
cmd = ['exiftool', '-b', '-PreviewImage', filepath]
result = subprocess.run(cmd, capture_output=True)
if result.returncode == 0 and result.stdout:
tmp.write(result.stdout)
tmp.flush()
return pyvips.Image.new_from_file(tmp.name, access='sequential')
except Exception as e:
logger.error(f"Error extracting RAW preview from {filepath}: {e}")
return None
def process_heic_image(filepath: str) -> pyvips.Image:
"""Process HEIC/HEIF image formats"""
try:
# Use pillow-heif to open the image
img = Image.open(filepath)
# Convert to RGB if needed
if img.mode != 'RGB':
img = img.convert('RGB')
# Save to temp file and load with pyvips
import tempfile
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as tmp:
img.save(tmp.name, 'JPEG')
return pyvips.Image.new_from_file(tmp.name, access='sequential')
except Exception as e:
logger.error(f"Error processing HEIC file {filepath}: {e}")
raise
def process_video_thumbnail(filepath: str) -> pyvips.Image:
"""Extract thumbnail from video file"""
try:
# Get video duration
probe = ffmpeg.probe(filepath)
duration = float(probe['streams'][0]['duration'])
# Extract frame at 10% of duration
timestamp = duration * 0.1
# Extract frame using ffmpeg
import tempfile
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as tmp:
stream = ffmpeg.input(filepath, ss=timestamp)
stream = ffmpeg.output(stream, tmp.name, vframes=1, format='image2', vcodec='mjpeg')
ffmpeg.run(stream, capture_stdout=True, capture_stderr=True)
return pyvips.Image.new_from_file(tmp.name, access='sequential')
except Exception as e:
logger.error(f"Error extracting video thumbnail from {filepath}: {e}")
# Create a placeholder thumbnail
return create_placeholder_thumbnail('video')
def create_placeholder_thumbnail(media_type: str) -> pyvips.Image:
"""Create a placeholder thumbnail for failed processing"""
# Create a simple gray placeholder
placeholder = pyvips.Image.black(640, 480)
placeholder = placeholder + [128, 128, 128] # Make it gray
return placeholder
def auto_rotate_image(image: pyvips.Image) -> pyvips.Image:
"""Auto-rotate image based on EXIF orientation"""
try:
orientation = image.get('orientation')
rotation_map = {
3: 180,
6: 90,
8: 270
}
if orientation in rotation_map:
image = image.rot(rotation_map[orientation])
except:
pass # No orientation data available
return image
def generate_thumbnail(image: pyvips.Image, size: int, output_path: str):
"""Generate a thumbnail of the specified size"""
# Calculate scale to fit within size (longest edge)
width = image.width
height = image.height
if width > height:
scale = size / width
else:
scale = size / height
# Only downscale, never upscale
if scale < 1:
image = image.resize(scale)
# Save as WebP with specified quality
image.webpsave(
output_path,
Q=settings.thumbnails.quality,
effort=4 # Balance between speed and compression
)
@shared_task(bind=True, name='generate_thumbnails')
def generate_thumbnails(self, photo_id: str):
"""Generate thumbnails for a photo"""
return asyncio.run(_generate_thumbnails_async(photo_id, self))
async def _generate_thumbnails_async(photo_id: str, task):
"""Async implementation of thumbnail generation"""
async with AsyncSessionLocal() as session:
try:
# Get photo from database
result = await session.execute(
select(Photo).where(Photo.id == photo_id)
)
photo = result.scalar_one_or_none()
if not photo:
logger.error(f"Photo not found: {photo_id}")
return {'status': 'error', 'message': 'Photo not found'}
# Check if file exists
if not os.path.exists(photo.filepath):
logger.error(f"File not found: {photo.filepath}")
photo.processing_status = 'failed'
photo.processing_error = 'File not found'
await session.commit()
return {'status': 'error', 'message': 'File not found'}
# Update processing status
photo.processing_status = 'processing'
await session.commit()
# Load and process the image based on type
image = None
if photo.media_type == 'photo':
image = process_standard_image(photo.filepath)
elif photo.media_type == 'raw':
image = process_raw_image(photo.filepath)
elif photo.media_type == 'heic':
image = process_heic_image(photo.filepath)
elif photo.media_type == 'video':
image = process_video_thumbnail(photo.filepath)
else:
logger.error(f"Unsupported media type: {photo.media_type}")
image = create_placeholder_thumbnail(photo.media_type)
if not image:
raise Exception("Failed to process image")
# Auto-rotate based on EXIF
image = auto_rotate_image(image)
# Store original dimensions
photo.width = image.width
photo.height = image.height
# Generate thumbnails for each size
for size_name, size_value in THUMB_SIZES.items():
thumb_path = get_thumb_path(photo_id, size_name)
generate_thumbnail(image, size_value, thumb_path)
# Update database with thumbnail path
setattr(photo, f'thumb_{size_name}', thumb_path)
# Update progress
task.update_state(
state='PROGRESS',
meta={'current_size': size_name, 'photo_id': photo_id}
)
# Update processing status
photo.processing_status = 'completed'
photo.processing_error = None
await session.commit()
logger.info(f"Thumbnails generated for photo {photo_id}")
return {'status': 'success', 'photo_id': photo_id}
except Exception as e:
logger.error(f"Error generating thumbnails for {photo_id}: {e}")
# Update error status
if photo:
photo.processing_status = 'failed'
photo.processing_error = str(e)
await session.commit()
return {'status': 'error', 'message': str(e)}
@shared_task(name='regenerate_all_thumbnails')
def regenerate_all_thumbnails():
"""Regenerate thumbnails for all photos"""
return asyncio.run(_regenerate_all_thumbnails_async())
async def _regenerate_all_thumbnails_async():
"""Async implementation of regenerating all thumbnails"""
async with AsyncSessionLocal() as session:
# Get all photos that need thumbnails
result = await session.execute(
select(Photo).where(
Photo.processing_status.in_(['pending', 'failed'])
)
)
photos = result.scalars().all()
logger.info(f"Regenerating thumbnails for {len(photos)} photos")
for photo in photos:
generate_thumbnails.delay(photo.id)
return {'status': 'queued', 'count': len(photos)}