bvandeusen
61f53a1ac1
Adds a header counter ('⚠ N characters need a fandom') that filters to
?null_fandom=1, plus an inline chip on each qualifying character card.
Backend accepts null_fandom_only in the list query.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2821 lines
90 KiB
Python
2821 lines
90 KiB
Python
# app/main.py
|
|
|
|
from flask import Blueprint, render_template, flash, redirect, url_for, abort, send_from_directory, request, jsonify, current_app
|
|
|
|
from sqlalchemy import desc, func, text, extract, and_
|
|
from sqlalchemy.orm import joinedload, aliased, contains_eager
|
|
from collections import OrderedDict
|
|
from datetime import datetime
|
|
|
|
from app.models import ImageRecord, Tag, ArchiveRecord, image_tags, ImportTask, ImportBatch, AppSettings
|
|
from app import db
|
|
from app.utils.tag_names import normalize_display_name
|
|
import os
|
|
import shutil
|
|
|
|
main = Blueprint('main', __name__)
|
|
|
|
|
|
def _ensure_fandom_tag(fandom_name):
|
|
"""Find or create a fandom tag by bare name. Returns the Tag object.
|
|
|
|
Normalizes the display name so callers don't have to remember.
|
|
"""
|
|
normalized = normalize_display_name(fandom_name.strip())
|
|
fandom_tag = Tag.query.filter_by(kind='fandom', name=normalized).first()
|
|
if not fandom_tag:
|
|
fandom_tag = Tag(kind='fandom', name=normalized)
|
|
db.session.add(fandom_tag)
|
|
db.session.flush()
|
|
return fandom_tag
|
|
|
|
|
|
@main.route('/')
|
|
def index():
|
|
"""
|
|
Showcase view: full-bleed masonry collage of random images/videos.
|
|
"""
|
|
images = (
|
|
ImageRecord.query
|
|
.options(joinedload(ImageRecord.tags))
|
|
.order_by(func.random())
|
|
.limit(20)
|
|
.all()
|
|
)
|
|
return render_template('showcase.html', images=images)
|
|
|
|
|
|
@main.route('/api/random-images')
|
|
def random_images_api():
|
|
"""
|
|
JSON endpoint returning random images for shuffle/infinite scroll.
|
|
|
|
Query params:
|
|
- count: number of images to return (default 12)
|
|
- exclude: comma-separated list of image IDs to exclude (for infinite scroll)
|
|
"""
|
|
count = request.args.get('count', 12, type=int)
|
|
exclude_str = request.args.get('exclude', '')
|
|
|
|
# Parse exclusion list
|
|
exclude_ids = []
|
|
if exclude_str:
|
|
try:
|
|
exclude_ids = [int(x) for x in exclude_str.split(',') if x.strip()]
|
|
except ValueError:
|
|
pass
|
|
|
|
# Build query
|
|
q = ImageRecord.query.options(joinedload(ImageRecord.tags))
|
|
|
|
if exclude_ids:
|
|
q = q.filter(~ImageRecord.id.in_(exclude_ids))
|
|
|
|
images = q.order_by(func.random()).limit(count).all()
|
|
|
|
result = []
|
|
for img in images:
|
|
is_video = img.filename.lower().endswith(('.mp4', '.mov'))
|
|
thumb_path = (img.thumb_path or img.filepath).replace('/images/', '')
|
|
full_path = img.filepath.replace('/images/', '')
|
|
|
|
result.append({
|
|
'id': img.id,
|
|
'filename': img.filename,
|
|
'thumb_url': url_for('main.serve_image', filename=thumb_path),
|
|
'full_url': url_for('main.serve_image', filename=full_path),
|
|
'is_video': is_video,
|
|
'width': img.width,
|
|
'height': img.height,
|
|
'date': (img.taken_at or img.imported_at).strftime('%Y-%m-%d') if (img.taken_at or img.imported_at) else '',
|
|
'tags': [{'name': t.name, 'kind': t.kind} for t in img.tags]
|
|
})
|
|
|
|
return jsonify(images=result)
|
|
|
|
|
|
@main.route('/gallery')
|
|
def gallery():
|
|
"""
|
|
Gallery with infinite scroll and year-month date dividers.
|
|
Server renders initial batch, then JS takes over for infinite scroll.
|
|
Optional tag filter via ?tag=artist:NAME or any tag name.
|
|
"""
|
|
from app.models import PostMetadata
|
|
|
|
initial_limit = 30
|
|
tag_id = request.args.get('tag_id', type=int)
|
|
active_tag_obj = (
|
|
Tag.query.options(joinedload(Tag.fandom)).get(tag_id)
|
|
if tag_id else None
|
|
)
|
|
|
|
# Base query + eager-load tags
|
|
q = ImageRecord.query.options(joinedload(ImageRecord.tags))
|
|
|
|
if active_tag_obj:
|
|
q = q.join(ImageRecord.tags).filter(Tag.id == active_tag_obj.id)
|
|
|
|
# Fetch initial batch ordered by date descending
|
|
images = q.order_by(
|
|
desc(func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at))
|
|
).limit(initial_limit + 1).all()
|
|
|
|
# Check if there are more images
|
|
has_more = len(images) > initial_limit
|
|
if has_more:
|
|
images = images[:initial_limit]
|
|
|
|
# Group by year-month for initial render
|
|
grouped_images = _group_images_by_month(images)
|
|
|
|
# Compute initial cursor for JS to continue from
|
|
initial_cursor = None
|
|
if images and has_more:
|
|
last_img = images[-1]
|
|
last_dt = _get_image_date(last_img)
|
|
if last_dt:
|
|
initial_cursor = last_dt.isoformat()
|
|
|
|
# Fetch post metadata if filtering by a post tag
|
|
post_metadata = None
|
|
if active_tag_obj and active_tag_obj.kind == 'post':
|
|
pm = PostMetadata.query.filter_by(tag_id=active_tag_obj.id).first()
|
|
if pm:
|
|
post_metadata = {
|
|
'title': pm.title,
|
|
'description': pm.description,
|
|
'source_url': pm.source_url,
|
|
'platform': pm.platform,
|
|
'artist': pm.artist,
|
|
'post_id': pm.post_id,
|
|
'attachment_count': pm.attachment_count,
|
|
'published_at': pm.published_at,
|
|
}
|
|
|
|
# Fetch series info if filtering by a series tag
|
|
from app.models import SeriesPage
|
|
series_info = None
|
|
if active_tag_obj and active_tag_obj.kind == 'series':
|
|
page_count = SeriesPage.query.filter_by(series_tag_id=active_tag_obj.id).count()
|
|
series_info = {
|
|
'tag_id': active_tag_obj.id,
|
|
'name': active_tag_obj.name,
|
|
'page_count': page_count,
|
|
}
|
|
|
|
return render_template(
|
|
'gallery.html',
|
|
grouped_images=grouped_images,
|
|
active_tag=active_tag_obj.display_name if active_tag_obj else None,
|
|
active_tag_obj=active_tag_obj,
|
|
initial_cursor=initial_cursor,
|
|
has_more=has_more,
|
|
post_metadata=post_metadata,
|
|
series_info=series_info
|
|
)
|
|
|
|
|
|
@main.route('/images/<path:filename>')
|
|
def serve_image(filename):
|
|
return send_from_directory('/images', filename)
|
|
|
|
|
|
# ----------------------------
|
|
# Gallery Infinite Scroll API
|
|
# ----------------------------
|
|
|
|
def _get_image_date(img):
|
|
"""Get the effective date for an image (taken_at or imported_at)."""
|
|
return img.taken_at or img.imported_at
|
|
|
|
|
|
def _format_year_month(dt):
|
|
"""Format a datetime as year-month key and label."""
|
|
if not dt:
|
|
return 'unknown', 'Unknown Date'
|
|
return dt.strftime('%Y-%m'), dt.strftime('%B %Y')
|
|
|
|
|
|
def _group_images_by_month(images):
|
|
"""
|
|
Group images by year-month, preserving order.
|
|
Returns list of (group_info, images) tuples.
|
|
"""
|
|
groups = OrderedDict()
|
|
|
|
for img in images:
|
|
dt = _get_image_date(img)
|
|
key, label = _format_year_month(dt)
|
|
|
|
if key not in groups:
|
|
groups[key] = {'key': key, 'label': label, 'images': []}
|
|
|
|
groups[key]['images'].append(img)
|
|
|
|
return [(info['key'], info['label'], info['images']) for info in groups.values()]
|
|
|
|
|
|
def _serialize_image(img):
|
|
"""Serialize an ImageRecord for JSON response."""
|
|
is_video = img.filename.lower().endswith(('.mp4', '.mov'))
|
|
thumb_path = (img.thumb_path or img.filepath).replace('/images/', '')
|
|
full_path = img.filepath.replace('/images/', '')
|
|
dt = _get_image_date(img)
|
|
key, label = _format_year_month(dt)
|
|
|
|
return {
|
|
'id': img.id,
|
|
'filename': img.filename,
|
|
'thumb_url': url_for('main.serve_image', filename=thumb_path),
|
|
'full_url': url_for('main.serve_image', filename=full_path),
|
|
'is_video': is_video,
|
|
'date': dt.strftime('%Y-%m-%d') if dt else '',
|
|
'year_month': key,
|
|
'year_month_label': label,
|
|
'tags': [{'name': t.name, 'kind': t.kind} for t in img.tags if t.kind != 'archive']
|
|
}
|
|
|
|
|
|
@main.route('/api/gallery/scroll')
|
|
def gallery_scroll():
|
|
"""
|
|
Cursor-based infinite scroll endpoint with date grouping.
|
|
|
|
Query params:
|
|
- cursor: ISO datetime string for pagination (fetch images older than this)
|
|
- limit: number of images per batch (default 30)
|
|
- tag: optional tag filter
|
|
|
|
Returns:
|
|
- images: list of image objects with year_month grouping info
|
|
- next_cursor: datetime string for next batch (null if no more)
|
|
- date_groups: summary of year-month groups in this batch
|
|
- has_more: boolean indicating if more images exist
|
|
"""
|
|
cursor_str = request.args.get('cursor')
|
|
limit = request.args.get('limit', 30, type=int)
|
|
tag_id = request.args.get('tag_id', type=int)
|
|
|
|
# Build base query
|
|
q = ImageRecord.query.options(joinedload(ImageRecord.tags))
|
|
|
|
# Optional tag filter
|
|
if tag_id:
|
|
q = q.join(ImageRecord.tags).filter(Tag.id == tag_id)
|
|
|
|
# Apply cursor filter (fetch images older than cursor)
|
|
if cursor_str:
|
|
try:
|
|
cursor_dt = datetime.fromisoformat(cursor_str.replace('Z', '+00:00'))
|
|
q = q.filter(
|
|
func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at) < cursor_dt
|
|
)
|
|
except ValueError:
|
|
pass # Invalid cursor, ignore
|
|
|
|
# Order by date descending and fetch with limit + 1 to check has_more
|
|
images = q.order_by(
|
|
desc(func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at))
|
|
).limit(limit + 1).all()
|
|
|
|
# Check if there are more images
|
|
has_more = len(images) > limit
|
|
if has_more:
|
|
images = images[:limit]
|
|
|
|
# Serialize images
|
|
serialized = [_serialize_image(img) for img in images]
|
|
|
|
# Build date_groups summary
|
|
groups_seen = OrderedDict()
|
|
for s in serialized:
|
|
key = s['year_month']
|
|
if key not in groups_seen:
|
|
groups_seen[key] = {'key': key, 'label': s['year_month_label'], 'count': 0}
|
|
groups_seen[key]['count'] += 1
|
|
|
|
# Compute next cursor
|
|
next_cursor = None
|
|
if images and has_more:
|
|
last_img = images[-1]
|
|
last_dt = _get_image_date(last_img)
|
|
if last_dt:
|
|
next_cursor = last_dt.isoformat()
|
|
|
|
return jsonify(
|
|
images=serialized,
|
|
next_cursor=next_cursor,
|
|
has_more=has_more,
|
|
date_groups=list(groups_seen.values())
|
|
)
|
|
|
|
|
|
@main.route('/api/gallery/timeline')
|
|
def gallery_timeline():
|
|
"""
|
|
Returns all year-month groups with counts for timeline navigation.
|
|
|
|
Query params:
|
|
- tag: optional tag filter (same as gallery)
|
|
|
|
Returns list of {key, label, count, year} ordered newest to oldest.
|
|
"""
|
|
tag_id = request.args.get('tag_id', type=int)
|
|
|
|
# Use SQL to aggregate by year-month
|
|
date_expr = func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at)
|
|
|
|
q = db.session.query(
|
|
extract('year', date_expr).label('year'),
|
|
extract('month', date_expr).label('month'),
|
|
func.count(ImageRecord.id).label('count')
|
|
)
|
|
|
|
if tag_id:
|
|
q = q.join(ImageRecord.tags).filter(Tag.id == tag_id)
|
|
|
|
q = q.group_by(
|
|
extract('year', date_expr),
|
|
extract('month', date_expr)
|
|
).order_by(
|
|
extract('year', date_expr).desc(),
|
|
extract('month', date_expr).desc()
|
|
)
|
|
|
|
results = q.all()
|
|
|
|
# Format results
|
|
groups = []
|
|
for row in results:
|
|
if row.year and row.month:
|
|
year = int(row.year)
|
|
month = int(row.month)
|
|
key = f"{year:04d}-{month:02d}"
|
|
# Create a date object to format the month name
|
|
dt = datetime(year, month, 1)
|
|
label = dt.strftime('%B %Y')
|
|
short_label = dt.strftime('%b') # Short month name for sidebar
|
|
groups.append({
|
|
'key': key,
|
|
'label': label,
|
|
'short_label': short_label,
|
|
'year': year,
|
|
'month': month,
|
|
'count': row.count
|
|
})
|
|
|
|
return jsonify(groups=groups)
|
|
|
|
|
|
@main.route('/api/gallery/jump')
|
|
def gallery_jump():
|
|
"""
|
|
Fetch images starting from a specific year-month for timeline jump navigation.
|
|
|
|
Query params:
|
|
- year_month: target year-month (e.g., '2024-03')
|
|
- limit: number of images (default 30)
|
|
- tag: optional tag filter
|
|
|
|
Returns same format as /api/gallery/scroll but starting from the target month.
|
|
"""
|
|
year_month = request.args.get('year_month', '')
|
|
limit = request.args.get('limit', 30, type=int)
|
|
tag_id = request.args.get('tag_id', type=int)
|
|
|
|
if not year_month:
|
|
return jsonify(error='year_month is required'), 400
|
|
|
|
try:
|
|
# Parse year-month to get start of that month
|
|
target_dt = datetime.strptime(year_month + '-01', '%Y-%m-%d')
|
|
# Get end of month (start of next month)
|
|
if target_dt.month == 12:
|
|
next_month = datetime(target_dt.year + 1, 1, 1)
|
|
else:
|
|
next_month = datetime(target_dt.year, target_dt.month + 1, 1)
|
|
except ValueError:
|
|
return jsonify(error='Invalid year_month format'), 400
|
|
|
|
# Build query for images in that month and earlier
|
|
q = ImageRecord.query.options(joinedload(ImageRecord.tags))
|
|
|
|
if tag_id:
|
|
q = q.join(ImageRecord.tags).filter(Tag.id == tag_id)
|
|
|
|
# Get images from start of target month onwards (but ordered descending)
|
|
date_expr = func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at)
|
|
q = q.filter(date_expr < next_month)
|
|
|
|
images = q.order_by(desc(date_expr)).limit(limit + 1).all()
|
|
|
|
has_more = len(images) > limit
|
|
if has_more:
|
|
images = images[:limit]
|
|
|
|
serialized = [_serialize_image(img) for img in images]
|
|
|
|
# Build date_groups summary
|
|
groups_seen = OrderedDict()
|
|
for s in serialized:
|
|
key = s['year_month']
|
|
if key not in groups_seen:
|
|
groups_seen[key] = {'key': key, 'label': s['year_month_label'], 'count': 0}
|
|
groups_seen[key]['count'] += 1
|
|
|
|
next_cursor = None
|
|
if images and has_more:
|
|
last_img = images[-1]
|
|
last_dt = _get_image_date(last_img)
|
|
if last_dt:
|
|
next_cursor = last_dt.isoformat()
|
|
|
|
return jsonify(
|
|
images=serialized,
|
|
next_cursor=next_cursor,
|
|
has_more=has_more,
|
|
date_groups=list(groups_seen.values()),
|
|
target_year_month=year_month
|
|
)
|
|
|
|
|
|
def _get_tags_with_previews(kind=None, limit=35, offset=0, images_per_tag=3, search=None, null_fandom_only=False):
|
|
"""
|
|
Helper function to get tags with counts and preview images.
|
|
Returns (tag_data list, total_count, has_more).
|
|
|
|
Optimized to use 2 queries instead of N+1:
|
|
1. Get tags with counts (paginated)
|
|
2. Get preview images for those tags using window function
|
|
"""
|
|
# Query for total count
|
|
count_q = db.session.query(func.count(Tag.id))
|
|
if kind:
|
|
count_q = count_q.filter(Tag.kind == kind)
|
|
if search:
|
|
count_q = count_q.filter(Tag.name.ilike(f'%{search}%'))
|
|
if null_fandom_only:
|
|
count_q = count_q.filter(Tag.kind == 'character', Tag.fandom_id.is_(None))
|
|
total_count = count_q.scalar()
|
|
|
|
# Query 1: Get tags with image counts (paginated)
|
|
q = db.session.query(
|
|
Tag,
|
|
func.count(image_tags.c.image_id).label('img_count')
|
|
).outerjoin(image_tags).group_by(Tag.id)
|
|
|
|
if kind:
|
|
q = q.filter(Tag.kind == kind)
|
|
if search:
|
|
q = q.filter(Tag.name.ilike(f'%{search}%'))
|
|
if null_fandom_only:
|
|
q = q.filter(Tag.kind == 'character', Tag.fandom_id.is_(None))
|
|
|
|
tags_with_counts = q.order_by(Tag.name.asc()).offset(offset).limit(limit).all()
|
|
|
|
# Build tag_id list and initialize tag_data dict
|
|
tag_ids = [t.id for t, _ in tags_with_counts]
|
|
tag_map = {}
|
|
for t, count in tags_with_counts:
|
|
label = t.display_name
|
|
tag_map[t.id] = {
|
|
"id": t.id,
|
|
"name": label,
|
|
"tag": t.name,
|
|
"images": [],
|
|
"kind": t.kind,
|
|
"fandom_id": t.fandom_id,
|
|
"count": count
|
|
}
|
|
|
|
# Query 2: Get preview images for all tags at once using a subquery with row_number
|
|
if tag_ids:
|
|
row_num = func.row_number().over(
|
|
partition_by=image_tags.c.tag_id,
|
|
order_by=desc(func.coalesce(ImageRecord.taken_at, ImageRecord.imported_at))
|
|
).label('rn')
|
|
|
|
subq = (
|
|
db.session.query(
|
|
image_tags.c.tag_id,
|
|
ImageRecord.id.label('image_id'),
|
|
row_num
|
|
)
|
|
.join(ImageRecord, ImageRecord.id == image_tags.c.image_id)
|
|
.filter(image_tags.c.tag_id.in_(tag_ids))
|
|
.subquery()
|
|
)
|
|
|
|
preview_rows = (
|
|
db.session.query(subq.c.tag_id, ImageRecord)
|
|
.join(ImageRecord, ImageRecord.id == subq.c.image_id)
|
|
.filter(subq.c.rn <= images_per_tag)
|
|
.all()
|
|
)
|
|
|
|
for tag_id, img in preview_rows:
|
|
if tag_id in tag_map:
|
|
tag_map[tag_id]["images"].append(img)
|
|
|
|
# Convert to list maintaining original order
|
|
tag_data = [tag_map[t.id] for t, _ in tags_with_counts]
|
|
has_more = (offset + len(tag_data)) < total_count
|
|
|
|
return tag_data, total_count, has_more
|
|
|
|
|
|
@main.route('/tags')
|
|
def tag_list():
|
|
"""
|
|
Generic tag explorer with infinite scroll:
|
|
/tags -> all tags
|
|
/tags?kind=user -> only user tags
|
|
/tags?search=xyz -> tags matching search term
|
|
/tags?null_fandom=1 -> only fandom-less character tags
|
|
Shows a few preview images per tag.
|
|
Initially loads 35 tags, more loaded via API as user scrolls.
|
|
"""
|
|
kind = request.args.get('kind')
|
|
search = request.args.get('search', '').strip() or None
|
|
null_fandom_only = request.args.get('null_fandom') == '1'
|
|
limit = 35
|
|
images_per_tag = 3
|
|
|
|
tag_data, total_count, has_more = _get_tags_with_previews(
|
|
kind=kind, limit=limit, offset=0, images_per_tag=images_per_tag,
|
|
search=search, null_fandom_only=null_fandom_only
|
|
)
|
|
|
|
null_fandom_character_count = (
|
|
db.session.query(func.count(Tag.id))
|
|
.filter(Tag.kind == 'character', Tag.fandom_id.is_(None))
|
|
.scalar() or 0
|
|
)
|
|
|
|
return render_template('tags_list.html',
|
|
tag_data=tag_data,
|
|
images_per_tag=images_per_tag,
|
|
active_kind=kind,
|
|
search_query=search or '',
|
|
total_count=total_count,
|
|
has_more=has_more,
|
|
page_size=limit,
|
|
null_fandom_character_count=null_fandom_character_count,
|
|
show_null_fandom_filter=null_fandom_only)
|
|
|
|
|
|
@main.route('/api/tags/list')
|
|
def api_tags_list():
|
|
"""
|
|
API endpoint for loading more tags (infinite scroll).
|
|
Returns JSON with tag cards HTML and pagination info.
|
|
"""
|
|
kind = request.args.get('kind')
|
|
search = request.args.get('search', '').strip() or None
|
|
null_fandom_only = request.args.get('null_fandom') == '1'
|
|
offset = request.args.get('offset', 0, type=int)
|
|
limit = request.args.get('limit', 35, type=int)
|
|
images_per_tag = 3
|
|
|
|
tag_data, total_count, has_more = _get_tags_with_previews(
|
|
kind=kind, limit=limit, offset=offset, images_per_tag=images_per_tag,
|
|
search=search, null_fandom_only=null_fandom_only
|
|
)
|
|
|
|
# Render tag cards to HTML
|
|
cards_html = render_template('_tag_cards.html', tag_data=tag_data, show_null_fandom_filter=null_fandom_only)
|
|
|
|
return jsonify(
|
|
ok=True,
|
|
html=cards_html,
|
|
count=len(tag_data),
|
|
total=total_count,
|
|
has_more=has_more,
|
|
offset=offset + len(tag_data)
|
|
)
|
|
|
|
@main.route('/artists')
|
|
def artist_list():
|
|
# Backward-compatible route that now shows the Tag Explorer filtered to artist tags
|
|
return redirect(url_for('main.tag_list', kind='artist'))
|
|
|
|
@main.route('/artist/<artist_name>')
|
|
def artist_gallery(artist_name):
|
|
# Backward-compatible: go to gallery filtered by artist tag
|
|
return redirect(url_for('main.gallery', tag=f"artist:{artist_name}"))
|
|
|
|
|
|
@main.route('/settings')
|
|
def settings():
|
|
valid_tabs = {'overview', 'import', 'maintenance'}
|
|
tab = request.args.get('tab', 'overview')
|
|
if tab not in valid_tabs:
|
|
tab = 'overview'
|
|
return render_template('settings.html', active_tab=tab)
|
|
|
|
|
|
@main.post('/settings/maintenance/ml-backfill')
|
|
def trigger_ml_backfill():
|
|
from app.tasks.ml import backfill
|
|
backfill.apply_async(queue='ml')
|
|
return redirect(url_for('main.settings', tab='maintenance'))
|
|
|
|
|
|
@main.post('/settings/maintenance/recompute-centroids')
|
|
def trigger_recompute_all_centroids():
|
|
from app.tasks.ml import recompute_all_centroids
|
|
recompute_all_centroids.apply_async(queue='ml')
|
|
return redirect(url_for('main.settings', tab='maintenance'))
|
|
|
|
|
|
# ----------------------------
|
|
# Tag add/remove endpoints
|
|
# ----------------------------
|
|
|
|
@main.get("/api/tags/search")
|
|
def search_tags():
|
|
"""
|
|
Search existing tags for autocomplete.
|
|
Query params:
|
|
- q: search query (matches against display_name, case-insensitive)
|
|
- limit: max results (default 10)
|
|
- exclude_kind: comma-separated list of tag kinds to exclude
|
|
- kind: filter to only include tags of this kind
|
|
"""
|
|
from sqlalchemy import case
|
|
|
|
query = request.args.get("q", "").strip().lower()
|
|
limit = request.args.get("limit", 10, type=int)
|
|
exclude_kind = request.args.get("exclude_kind", "").strip()
|
|
exclude_kinds = [k.strip() for k in exclude_kind.split(",") if k.strip()]
|
|
include_kind = request.args.get("kind", "").strip()
|
|
|
|
f = aliased(Tag)
|
|
display_expr = case(
|
|
(
|
|
and_(Tag.kind == "character", Tag.fandom_id.isnot(None)),
|
|
func.concat(Tag.name, " (", f.name, ")"),
|
|
),
|
|
else_=Tag.name,
|
|
)
|
|
|
|
base_query = (
|
|
db.session.query(Tag)
|
|
.outerjoin(f, f.id == Tag.fandom_id)
|
|
.options(contains_eager(Tag.fandom, alias=f))
|
|
)
|
|
if exclude_kinds:
|
|
base_query = base_query.filter(~Tag.kind.in_(exclude_kinds))
|
|
if include_kind:
|
|
base_query = base_query.filter(Tag.kind == include_kind)
|
|
|
|
if not query:
|
|
# No query: most-used tags.
|
|
base_query = base_query.join(image_tags, image_tags.c.tag_id == Tag.id).group_by(Tag.id, f.id)
|
|
tags = (
|
|
base_query
|
|
.order_by(func.count(image_tags.c.image_id).desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
else:
|
|
base_query = base_query.filter(display_expr.ilike(f"%{query}%"))
|
|
tags = (
|
|
base_query
|
|
.order_by(display_expr)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return jsonify(tags=[
|
|
{
|
|
"id": t.id,
|
|
"name": t.name,
|
|
"display_name": t.display_name,
|
|
"kind": t.kind,
|
|
"fandom_id": t.fandom_id if t.kind == "character" else None,
|
|
"fandom_name": (t.fandom.name if t.fandom else None) if t.kind == "character" else None,
|
|
}
|
|
for t in tags
|
|
])
|
|
|
|
|
|
@main.get("/image/<int:image_id>/tags")
|
|
def list_tags(image_id):
|
|
img = ImageRecord.query.get_or_404(image_id)
|
|
tags_with_fandom = (
|
|
db.session.query(Tag)
|
|
.join(image_tags, image_tags.c.tag_id == Tag.id)
|
|
.filter(image_tags.c.image_id == image_id)
|
|
.options(joinedload(Tag.fandom))
|
|
.order_by(Tag.kind.asc(), Tag.name.asc())
|
|
.all()
|
|
)
|
|
return jsonify(tags=[
|
|
{
|
|
"id": t.id,
|
|
"name": t.name,
|
|
"display_name": t.display_name,
|
|
"kind": t.kind,
|
|
"fandom_id": t.fandom_id if t.kind == "character" else None,
|
|
"fandom_name": (t.fandom.name if t.fandom else None) if t.kind == "character" else None,
|
|
}
|
|
for t in tags_with_fandom
|
|
])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tag suggestions (ML-backed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@main.get("/image/<int:image_id>/suggestions")
|
|
def get_image_suggestions(image_id):
|
|
from app.services.tag_suggestions import get_suggestions
|
|
suggestions = get_suggestions(image_id)
|
|
return jsonify({'ok': True, 'suggestions': suggestions})
|
|
|
|
|
|
@main.post("/api/suggestions/bulk")
|
|
def get_bulk_suggestions_api():
|
|
"""Consensus ML suggestions across a selection of images.
|
|
|
|
Body: {"image_ids": [...], "threshold": 0.8}
|
|
Returns grouped suggestions where each tag was suggested for or already
|
|
applied to at least `threshold` fraction of the selected images.
|
|
"""
|
|
from app.services.tag_suggestions import get_bulk_suggestions
|
|
payload = request.get_json(force=True, silent=True) or {}
|
|
image_ids = payload.get('image_ids') or []
|
|
try:
|
|
image_ids = [int(x) for x in image_ids]
|
|
except (TypeError, ValueError):
|
|
return jsonify({'ok': False, 'error': 'invalid_image_ids'}), 400
|
|
if not image_ids:
|
|
return jsonify({'ok': False, 'error': 'no_image_ids'}), 400
|
|
|
|
try:
|
|
threshold = float(payload.get('threshold', 0.8))
|
|
except (TypeError, ValueError):
|
|
threshold = 0.8
|
|
threshold = max(0.0, min(1.0, threshold))
|
|
|
|
suggestions = get_bulk_suggestions(image_ids, threshold=threshold)
|
|
return jsonify({'ok': True, 'suggestions': suggestions, 'evaluated': len(image_ids), 'threshold': threshold})
|
|
|
|
|
|
@main.post("/image/<int:image_id>/suggestions/accept")
|
|
def accept_image_suggestion(image_id):
|
|
"""
|
|
Accept a single tag suggestion for an image.
|
|
|
|
Post-refactor resolution:
|
|
- If `tag_id` is provided, attach that exact tag (used by the
|
|
frontend after the user resolves an ambiguous suggestion).
|
|
- Else for character category: search by (kind='character', name=bare).
|
|
0 matches -> create null-fandom tag. 1 match -> attach. N matches ->
|
|
return HTTP 409 with candidate list for the frontend to disambiguate.
|
|
- Else for fandom/general: find-or-create by (kind, name).
|
|
|
|
Records a SuggestionFeedback row on every accepted decision (including
|
|
the explicit_tag_id disambiguation path) and conditionally enqueues a
|
|
centroid recompute when enough new accepts have accumulated on an
|
|
eligible kind. The 409 ambiguous branch does NOT log feedback because
|
|
no decision has been made yet.
|
|
"""
|
|
from app.models import SuggestionFeedback
|
|
from app.services.tag_suggestions import _WD14_CATEGORY_TO_KIND, _canonicalize_wd14_name
|
|
|
|
data = request.get_json() or {}
|
|
category = data.get("category")
|
|
name = (data.get("name") or "").strip()
|
|
explicit_tag_id = data.get("tag_id")
|
|
source = data.get("source") or ""
|
|
confidence = float(data.get("confidence") or 0.0)
|
|
|
|
# Defensive: clients may send raw WD14 names with underscores or off casing.
|
|
if name:
|
|
name = _canonicalize_wd14_name(name, category)
|
|
|
|
img = ImageRecord.query.get_or_404(image_id)
|
|
kind = _WD14_CATEGORY_TO_KIND.get(category)
|
|
|
|
if explicit_tag_id:
|
|
tag = Tag.query.get(explicit_tag_id)
|
|
if tag is None:
|
|
return jsonify(ok=False, error="tag_not_found"), 404
|
|
elif kind == "character":
|
|
candidates = Tag.query.filter_by(kind="character", name=name).all()
|
|
if len(candidates) == 1:
|
|
tag = candidates[0]
|
|
elif len(candidates) == 0:
|
|
tag = Tag(kind="character", name=name, fandom_id=None)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
else:
|
|
return jsonify(
|
|
ok=False,
|
|
error="ambiguous",
|
|
candidates=[
|
|
{
|
|
"id": c.id,
|
|
"display_name": c.display_name,
|
|
"fandom_id": c.fandom_id,
|
|
"fandom_name": c.fandom.name if c.fandom else None,
|
|
}
|
|
for c in candidates
|
|
],
|
|
), 409
|
|
else:
|
|
effective_kind = kind if kind else "user"
|
|
tag = Tag.query.filter_by(kind=effective_kind, name=name).first()
|
|
if tag is None:
|
|
tag = Tag(kind=effective_kind, name=name)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
|
|
# Auto-apply the character's fandom if set.
|
|
fandom_tag = None
|
|
if tag.kind == "character" and tag.fandom_id:
|
|
fandom_tag = Tag.query.get(tag.fandom_id)
|
|
|
|
if tag not in img.tags:
|
|
img.tags.append(tag)
|
|
if fandom_tag and fandom_tag not in img.tags:
|
|
img.tags.append(fandom_tag)
|
|
|
|
db.session.add(SuggestionFeedback(
|
|
image_id=image_id,
|
|
tag_name=tag.name,
|
|
suggestion_source=source,
|
|
confidence=confidence,
|
|
decision='accepted',
|
|
))
|
|
db.session.commit()
|
|
|
|
# Conditionally schedule centroid recompute for eligible tag kinds.
|
|
from app.services.tag_suggestions import ELIGIBLE_CENTROID_KINDS
|
|
if tag.kind in ELIGIBLE_CENTROID_KINDS:
|
|
from app.models import TagReferenceEmbedding, TagSuggestionConfig
|
|
from app.tasks.ml import recompute_centroid
|
|
from app.ml.siglip import MODEL_VERSION as SIGLIP_VER
|
|
|
|
delta_cfg = TagSuggestionConfig.query.filter_by(key='centroid_recompute_delta').first()
|
|
delta_threshold = int(delta_cfg.value) if delta_cfg else 3
|
|
|
|
current_count = (
|
|
db.session.query(db.func.count(image_tags.c.image_id))
|
|
.filter(image_tags.c.tag_id == tag.id)
|
|
.scalar()
|
|
) or 0
|
|
ref = (
|
|
TagReferenceEmbedding.query
|
|
.filter_by(tag_name=tag.name, model_version=SIGLIP_VER)
|
|
.first()
|
|
)
|
|
last_count = ref.reference_count if ref else 0
|
|
if current_count - last_count >= delta_threshold:
|
|
try:
|
|
recompute_centroid.apply_async(args=[tag.name], queue='ml')
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify(
|
|
ok=True,
|
|
tag={
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
},
|
|
fandom_tag=(
|
|
{
|
|
"id": fandom_tag.id,
|
|
"name": fandom_tag.name,
|
|
"display_name": fandom_tag.display_name,
|
|
"kind": fandom_tag.kind,
|
|
}
|
|
if fandom_tag else None
|
|
),
|
|
)
|
|
|
|
|
|
@main.post("/image/<int:image_id>/suggestions/reject")
|
|
def reject_image_suggestion(image_id):
|
|
from app.models import SuggestionFeedback
|
|
payload = request.get_json(force=True, silent=True) or {}
|
|
tag_name = (payload.get('tag_name') or '').strip()
|
|
source = payload.get('source') or ''
|
|
confidence = float(payload.get('confidence') or 0.0)
|
|
if not tag_name or not source:
|
|
return jsonify({'ok': False, 'error': 'missing_fields'}), 400
|
|
|
|
if ImageRecord.query.get(image_id) is None:
|
|
return jsonify({'ok': False, 'error': 'image_not_found'}), 404
|
|
|
|
db.session.add(SuggestionFeedback(
|
|
image_id=image_id,
|
|
tag_name=tag_name,
|
|
suggestion_source=source,
|
|
confidence=confidence,
|
|
decision='rejected',
|
|
))
|
|
db.session.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
@main.post("/image/<int:image_id>/tags/add")
|
|
def add_tag(image_id):
|
|
"""
|
|
JSON endpoint to add a tag to an image.
|
|
|
|
Accepts the user-facing 'kind:name' shortcut in the 'name' field
|
|
(e.g. 'character:Saber'). Bare strings without a recognized prefix are
|
|
treated as user tags.
|
|
|
|
For character tags, optional 'fandom_id' (preferred) or 'fandom_name'
|
|
attaches the character to a fandom. Both omitted = null fandom (allowed).
|
|
"""
|
|
from app.utils.tag_prefix import parse_kind_prefix
|
|
from app.utils.tag_names import underscores_to_spaces
|
|
|
|
raw = (request.form.get("name") or "").strip()
|
|
if not raw:
|
|
abort(400)
|
|
|
|
kind, name = parse_kind_prefix(raw)
|
|
if kind is None:
|
|
# No recognized prefix — treat as a user-typed general tag.
|
|
kind = "user"
|
|
name = underscores_to_spaces(name)
|
|
elif kind in ("character", "fandom"):
|
|
name = normalize_display_name(name)
|
|
|
|
img = ImageRecord.query.get_or_404(image_id)
|
|
|
|
fandom_tag = None
|
|
if kind == "character":
|
|
fandom_id = request.form.get("fandom_id", type=int)
|
|
fandom_name = (request.form.get("fandom_name") or "").strip() or None
|
|
|
|
if fandom_id:
|
|
fandom_tag = Tag.query.filter_by(id=fandom_id, kind="fandom").first()
|
|
if fandom_tag is None:
|
|
return jsonify(ok=False, error="fandom_not_found"), 400
|
|
elif fandom_name:
|
|
fandom_tag = _ensure_fandom_tag(fandom_name)
|
|
|
|
tag = Tag.query.filter_by(
|
|
kind="character",
|
|
name=name,
|
|
fandom_id=fandom_tag.id if fandom_tag else None,
|
|
).first()
|
|
if tag is None:
|
|
tag = Tag(
|
|
kind="character",
|
|
name=name,
|
|
fandom_id=fandom_tag.id if fandom_tag else None,
|
|
)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
else:
|
|
tag = Tag.query.filter_by(kind=kind, name=name).first()
|
|
if tag is None:
|
|
tag = Tag(kind=kind, name=name)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
|
|
if tag not in img.tags:
|
|
img.tags.append(tag)
|
|
|
|
# Auto-apply fandom tag to the image (preserves today's UX feature).
|
|
if fandom_tag and fandom_tag not in img.tags:
|
|
img.tags.append(fandom_tag)
|
|
|
|
db.session.commit()
|
|
|
|
result = {
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
}
|
|
if fandom_tag:
|
|
result["fandom_tag"] = {
|
|
"id": fandom_tag.id,
|
|
"name": fandom_tag.name,
|
|
"display_name": fandom_tag.display_name,
|
|
"kind": fandom_tag.kind,
|
|
}
|
|
return jsonify(ok=True, tag=result)
|
|
|
|
|
|
@main.post("/image/<int:image_id>/tags/remove")
|
|
def remove_tag(image_id):
|
|
"""
|
|
Minimal JSON endpoint to remove a tag from an image.
|
|
"""
|
|
name = (request.form.get("name") or "").strip()
|
|
img = ImageRecord.query.get_or_404(image_id)
|
|
tag = Tag.query.filter_by(name=name).first()
|
|
if tag and tag in img.tags:
|
|
img.tags.remove(tag)
|
|
db.session.commit()
|
|
return jsonify(ok=True)
|
|
|
|
|
|
# ----------------------------
|
|
# Tag management endpoints
|
|
# ----------------------------
|
|
|
|
@main.get("/api/tag/<int:tag_id>")
|
|
def get_tag(tag_id):
|
|
"""Get a single tag's details."""
|
|
tag = Tag.query.options(joinedload(Tag.fandom)).get_or_404(tag_id)
|
|
result = {
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
"fandom_id": tag.fandom_id,
|
|
"fandom_name": tag.fandom.name if tag.fandom else None,
|
|
}
|
|
return jsonify(ok=True, tag=result)
|
|
|
|
|
|
@main.post("/api/tag/<int:tag_id>/set-fandom")
|
|
def set_tag_fandom(tag_id):
|
|
"""
|
|
Set or clear a character tag's fandom association.
|
|
|
|
Form params:
|
|
fandom_id: ID of an existing fandom tag (preferred).
|
|
fandom_name: Name to find/create a fandom tag (used if fandom_id
|
|
not given). If both omitted, the fandom is cleared.
|
|
"""
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
if tag.kind != "character":
|
|
return jsonify(ok=False, error="Only character tags can have fandoms"), 400
|
|
|
|
fandom_id = request.form.get("fandom_id", type=int)
|
|
fandom_name = (request.form.get("fandom_name") or "").strip()
|
|
|
|
if fandom_id:
|
|
fandom_tag = Tag.query.filter_by(id=fandom_id, kind="fandom").first()
|
|
if fandom_tag is None:
|
|
return jsonify(ok=False, error="Invalid fandom tag"), 400
|
|
tag.fandom_id = fandom_tag.id
|
|
elif fandom_name:
|
|
fandom_tag = _ensure_fandom_tag(fandom_name)
|
|
tag.fandom_id = fandom_tag.id
|
|
else:
|
|
tag.fandom_id = None
|
|
fandom_tag = None
|
|
|
|
db.session.commit()
|
|
|
|
# Auto-apply the fandom tag to images already tagged with this character.
|
|
if fandom_tag is not None:
|
|
try:
|
|
db.session.execute(
|
|
text("""
|
|
INSERT INTO image_tags (image_id, tag_id)
|
|
SELECT it.image_id, :fandom_id
|
|
FROM image_tags it
|
|
WHERE it.tag_id = :char_id
|
|
ON CONFLICT DO NOTHING
|
|
"""),
|
|
{"char_id": tag.id, "fandom_id": fandom_tag.id},
|
|
)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.warning(
|
|
"set_tag_fandom auto-apply failed for tag %s -> fandom %s: %s",
|
|
tag.id, fandom_tag.id, e,
|
|
)
|
|
|
|
return jsonify(ok=True, tag={
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
"fandom_id": tag.fandom_id,
|
|
"fandom_name": fandom_tag.name if fandom_tag else None,
|
|
})
|
|
|
|
|
|
@main.get("/api/post-metadata/<int:tag_id>")
|
|
def get_post_metadata(tag_id):
|
|
"""
|
|
Get post metadata for a post tag.
|
|
Returns title, description, source URL, etc. for display in the gallery.
|
|
"""
|
|
from app.models import PostMetadata
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
# Only post tags have metadata
|
|
if tag.kind != 'post':
|
|
return jsonify(ok=False, error="Not a post tag"), 400
|
|
|
|
pm = PostMetadata.query.filter_by(tag_id=tag_id).first()
|
|
if not pm:
|
|
return jsonify(ok=True, metadata=None)
|
|
|
|
return jsonify(ok=True, metadata={
|
|
"id": pm.id,
|
|
"tag_id": pm.tag_id,
|
|
"tag_name": tag.name,
|
|
"platform": pm.platform,
|
|
"post_id": pm.post_id,
|
|
"artist": pm.artist,
|
|
"title": pm.title,
|
|
"description": pm.description,
|
|
"source_url": pm.source_url,
|
|
"attachment_count": pm.attachment_count,
|
|
"published_at": pm.published_at.isoformat() if pm.published_at else None,
|
|
})
|
|
|
|
|
|
@main.get("/api/post-metadata/by-tag-name/<path:tag_name>")
|
|
def get_post_metadata_by_name(tag_name):
|
|
"""
|
|
Get post metadata by tag name (e.g., post:patreon:artist:12345).
|
|
Useful when filtering gallery by tag name.
|
|
"""
|
|
from app.models import PostMetadata
|
|
|
|
tag = Tag.query.filter_by(name=tag_name).first()
|
|
if not tag:
|
|
return jsonify(ok=True, metadata=None)
|
|
|
|
if tag.kind != 'post':
|
|
return jsonify(ok=False, error="Not a post tag"), 400
|
|
|
|
pm = PostMetadata.query.filter_by(tag_id=tag.id).first()
|
|
if not pm:
|
|
return jsonify(ok=True, metadata=None)
|
|
|
|
return jsonify(ok=True, metadata={
|
|
"id": pm.id,
|
|
"tag_id": pm.tag_id,
|
|
"tag_name": tag.name,
|
|
"platform": pm.platform,
|
|
"post_id": pm.post_id,
|
|
"artist": pm.artist,
|
|
"title": pm.title,
|
|
"description": pm.description,
|
|
"source_url": pm.source_url,
|
|
"attachment_count": pm.attachment_count,
|
|
"published_at": pm.published_at.isoformat() if pm.published_at else None,
|
|
})
|
|
|
|
|
|
@main.post("/api/tag/<int:tag_id>/update")
|
|
def update_tag(tag_id):
|
|
"""
|
|
Update a tag's name and/or kind.
|
|
|
|
For character tags: rename = update Tag.name (bare). Fandom association
|
|
is NOT changed here — use /api/tag/<id>/set-fandom for that.
|
|
|
|
If the new (name, kind [, fandom_id]) combination collides with an
|
|
existing tag, offer a merge (force_merge=true to commit).
|
|
"""
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
new_name = (request.form.get("name") or "").strip()
|
|
new_kind = (request.form.get("kind") or "user").strip()
|
|
force_merge = request.form.get("merge") == "true"
|
|
|
|
if not new_name:
|
|
return jsonify(ok=False, error="Name is required"), 400
|
|
|
|
if new_kind in ("character", "fandom"):
|
|
new_name = normalize_display_name(new_name)
|
|
|
|
# Check for duplicates using the new uniqueness shape
|
|
if new_kind == "character":
|
|
existing_q = Tag.query.filter(
|
|
Tag.kind == "character",
|
|
Tag.name == new_name,
|
|
Tag.fandom_id.is_(tag.fandom_id) if tag.fandom_id is None
|
|
else Tag.fandom_id == tag.fandom_id,
|
|
Tag.id != tag_id,
|
|
)
|
|
else:
|
|
existing_q = Tag.query.filter(
|
|
Tag.kind == new_kind,
|
|
Tag.name == new_name,
|
|
Tag.id != tag_id,
|
|
)
|
|
existing = existing_q.first()
|
|
|
|
if existing:
|
|
if not force_merge:
|
|
return jsonify(
|
|
ok=False,
|
|
error="A tag with this name already exists",
|
|
can_merge=True,
|
|
target_tag={
|
|
"id": existing.id,
|
|
"name": existing.name,
|
|
"display_name": existing.display_name,
|
|
"kind": existing.kind,
|
|
},
|
|
), 409
|
|
|
|
# Merge: move all images from source tag to target tag
|
|
source_images = tag.images[:]
|
|
merged_count = 0
|
|
for img in source_images:
|
|
if existing not in img.tags:
|
|
img.tags.append(existing)
|
|
merged_count += 1
|
|
img.tags.remove(tag)
|
|
db.session.delete(tag)
|
|
db.session.commit()
|
|
return jsonify(ok=True, merged=True, merged_count=merged_count, tag={
|
|
"id": existing.id,
|
|
"name": existing.name,
|
|
"display_name": existing.display_name,
|
|
"kind": existing.kind,
|
|
})
|
|
|
|
# Plain rename
|
|
tag.name = new_name
|
|
tag.kind = new_kind
|
|
db.session.commit()
|
|
return jsonify(ok=True, tag={
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
"fandom_id": tag.fandom_id,
|
|
})
|
|
|
|
|
|
@main.post("/api/tag/<int:tag_id>/delete")
|
|
def delete_tag(tag_id):
|
|
"""Delete a tag entirely."""
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
# Remove from all images first (handled by cascade, but explicit is clearer)
|
|
db.session.delete(tag)
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True)
|
|
|
|
|
|
# ----------------------------
|
|
# Filtered deletion endpoints
|
|
# ----------------------------
|
|
|
|
@main.post("/api/delete-by-tag")
|
|
def delete_images_by_tag():
|
|
"""
|
|
Delete all images that have a specific tag.
|
|
Also removes the image files and thumbnails from disk.
|
|
Query params:
|
|
- tag_id: ID of the tag whose images should be deleted
|
|
- tag_name: Name of the tag (must match for security validation)
|
|
- delete_tag: if "true", also delete the tag itself
|
|
"""
|
|
tag_id = request.form.get("tag_id", type=int)
|
|
tag_name = request.form.get("tag_name", "").strip()
|
|
delete_tag_also = request.form.get("delete_tag") == "true"
|
|
|
|
if not tag_id:
|
|
return jsonify(ok=False, error="tag_id is required"), 400
|
|
|
|
if not tag_name:
|
|
return jsonify(ok=False, error="tag_name confirmation is required"), 400
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
# Security: verify the provided tag name matches the actual tag
|
|
if tag.name != tag_name:
|
|
return jsonify(ok=False, error="Tag name confirmation does not match"), 403
|
|
|
|
images_to_delete = list(tag.images)
|
|
|
|
deleted_count = 0
|
|
errors = []
|
|
|
|
for img in images_to_delete:
|
|
try:
|
|
# Remove image file from disk
|
|
if img.filepath and os.path.exists(img.filepath):
|
|
os.remove(img.filepath)
|
|
# Remove thumbnail if exists
|
|
if img.thumb_path and os.path.exists(img.thumb_path):
|
|
os.remove(img.thumb_path)
|
|
# Delete DB record
|
|
db.session.delete(img)
|
|
deleted_count += 1
|
|
except Exception as e:
|
|
errors.append(f"Failed to delete {img.filename}: {e}")
|
|
|
|
# Optionally delete the tag itself
|
|
if delete_tag_also:
|
|
db.session.delete(tag)
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify(
|
|
ok=True,
|
|
deleted_count=deleted_count,
|
|
tag_deleted=delete_tag_also,
|
|
errors=errors if errors else None
|
|
)
|
|
|
|
|
|
@main.get("/api/tag/<int:tag_id>/image-count")
|
|
def get_tag_image_count(tag_id):
|
|
"""Get the number of images associated with a tag."""
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
count = len(tag.images)
|
|
return jsonify(ok=True, count=count, tag_name=tag.name)
|
|
|
|
|
|
# ----------------------------
|
|
# Import settings endpoints
|
|
# ----------------------------
|
|
|
|
@main.get("/api/import-settings")
|
|
def get_import_settings():
|
|
"""Get current import filter settings from database."""
|
|
from app.utils.version_migration import (
|
|
get_setting_int, get_setting_float, get_setting_bool
|
|
)
|
|
|
|
settings = {
|
|
"min_width": get_setting_int("import_min_width", int(os.environ.get("IMPORT_MIN_WIDTH", "0"))),
|
|
"min_height": get_setting_int("import_min_height", int(os.environ.get("IMPORT_MIN_HEIGHT", "0"))),
|
|
"skip_transparent": get_setting_bool("import_skip_transparent", os.environ.get("IMPORT_SKIP_TRANSPARENT", "false").lower() == "true"),
|
|
"transparency_threshold": get_setting_float("import_transparency_threshold", float(os.environ.get("IMPORT_TRANSPARENCY_THRESHOLD", "0.9"))),
|
|
"phash_threshold": get_setting_int("phash_threshold", int(os.environ.get("IMPORT_PHASH_THRESHOLD", "10"))),
|
|
}
|
|
return jsonify(ok=True, settings=settings)
|
|
|
|
|
|
@main.post("/api/import-settings")
|
|
def save_import_settings():
|
|
"""
|
|
Save import filter settings to database.
|
|
Settings are loaded by the Celery workers via load_import_settings().
|
|
"""
|
|
from app.utils.version_migration import set_setting
|
|
|
|
min_width = request.form.get("min_width", 0, type=int)
|
|
min_height = request.form.get("min_height", 0, type=int)
|
|
skip_transparent = request.form.get("skip_transparent") == "true"
|
|
transparency_threshold = request.form.get("transparency_threshold", 0.9, type=float)
|
|
phash_threshold = request.form.get("phash_threshold", 10, type=int)
|
|
|
|
try:
|
|
set_setting("import_min_width", str(min_width))
|
|
set_setting("import_min_height", str(min_height))
|
|
set_setting("import_skip_transparent", str(skip_transparent).lower())
|
|
set_setting("import_transparency_threshold", str(transparency_threshold))
|
|
set_setting("phash_threshold", str(phash_threshold))
|
|
|
|
settings = {
|
|
"min_width": min_width,
|
|
"min_height": min_height,
|
|
"skip_transparent": skip_transparent,
|
|
"transparency_threshold": transparency_threshold,
|
|
"phash_threshold": phash_threshold,
|
|
}
|
|
return jsonify(ok=True, settings=settings)
|
|
except Exception as e:
|
|
return jsonify(ok=False, error=str(e)), 500
|
|
|
|
|
|
# ----------------------------
|
|
# Image filter/cleanup endpoints
|
|
# ----------------------------
|
|
|
|
@main.get("/api/filter-scan")
|
|
def scan_filterable_images():
|
|
"""
|
|
Scan existing images against current filter settings.
|
|
Returns counts and IDs of images that would be filtered.
|
|
Query params:
|
|
- filter_type: 'transparency' or 'dimensions'
|
|
- threshold: for transparency (0.0-1.0), or min_width/min_height for dimensions
|
|
- min_width: minimum width (for dimensions filter)
|
|
- min_height: minimum height (for dimensions filter)
|
|
- limit: max results to return (default 100)
|
|
"""
|
|
from app.utils.image_importer import is_mostly_transparent
|
|
|
|
filter_type = request.args.get("filter_type", "transparency")
|
|
limit = request.args.get("limit", 100, type=int)
|
|
|
|
results = []
|
|
|
|
if filter_type == "transparency":
|
|
threshold = request.args.get("threshold", 0.9, type=float)
|
|
|
|
# Only check images that could have transparency
|
|
images = ImageRecord.query.filter(
|
|
ImageRecord.filepath.ilike("%.png") |
|
|
ImageRecord.filepath.ilike("%.gif") |
|
|
ImageRecord.filepath.ilike("%.webp")
|
|
).limit(limit * 5).all() # Check more to find enough matches
|
|
|
|
for img in images:
|
|
if len(results) >= limit:
|
|
break
|
|
if img.filepath and os.path.exists(img.filepath):
|
|
try:
|
|
if is_mostly_transparent(img.filepath, threshold):
|
|
results.append({
|
|
"id": img.id,
|
|
"filename": img.filename,
|
|
"filepath": img.filepath,
|
|
"thumb_url": url_for('main.serve_image',
|
|
filename=(img.thumb_path or img.filepath).replace('/images/', ''))
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
elif filter_type == "dimensions":
|
|
min_width = request.args.get("min_width", 0, type=int)
|
|
min_height = request.args.get("min_height", 0, type=int)
|
|
|
|
if min_width <= 0 and min_height <= 0:
|
|
return jsonify(ok=True, count=0, images=[], message="No dimension filter set")
|
|
|
|
query = ImageRecord.query
|
|
if min_width > 0:
|
|
query = query.filter(ImageRecord.width < min_width)
|
|
if min_height > 0:
|
|
query = query.filter(ImageRecord.height < min_height)
|
|
|
|
images = query.limit(limit).all()
|
|
|
|
for img in images:
|
|
results.append({
|
|
"id": img.id,
|
|
"filename": img.filename,
|
|
"filepath": img.filepath,
|
|
"width": img.width,
|
|
"height": img.height,
|
|
"thumb_url": url_for('main.serve_image',
|
|
filename=(img.thumb_path or img.filepath).replace('/images/', ''))
|
|
})
|
|
|
|
# Get total count (without limit) for progress indication
|
|
total_count = len(results)
|
|
|
|
return jsonify(ok=True, count=total_count, images=results, filter_type=filter_type)
|
|
|
|
|
|
@main.post("/api/filter-delete")
|
|
def delete_filtered_images():
|
|
"""
|
|
Delete images by their IDs after filter review.
|
|
Query params:
|
|
- image_ids: comma-separated list of image IDs to delete
|
|
- confirm: must be 'true' to actually delete
|
|
"""
|
|
image_ids_str = request.form.get("image_ids", "")
|
|
confirm = request.form.get("confirm") == "true"
|
|
|
|
if not confirm:
|
|
return jsonify(ok=False, error="Confirmation required"), 400
|
|
|
|
if not image_ids_str:
|
|
return jsonify(ok=False, error="No image IDs provided"), 400
|
|
|
|
try:
|
|
image_ids = [int(x) for x in image_ids_str.split(",") if x.strip()]
|
|
except ValueError:
|
|
return jsonify(ok=False, error="Invalid image IDs"), 400
|
|
|
|
deleted_count = 0
|
|
errors = []
|
|
|
|
for img_id in image_ids:
|
|
img = ImageRecord.query.get(img_id)
|
|
if not img:
|
|
continue
|
|
|
|
try:
|
|
# Remove image file from disk
|
|
if img.filepath and os.path.exists(img.filepath):
|
|
os.remove(img.filepath)
|
|
# Remove thumbnail if exists
|
|
if img.thumb_path and os.path.exists(img.thumb_path):
|
|
os.remove(img.thumb_path)
|
|
# Delete DB record
|
|
db.session.delete(img)
|
|
deleted_count += 1
|
|
except Exception as e:
|
|
errors.append(f"Failed to delete {img.filename}: {e}")
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify(
|
|
ok=True,
|
|
deleted_count=deleted_count,
|
|
errors=errors if errors else None
|
|
)
|
|
|
|
|
|
# ----------------------------
|
|
# Import Task Queue API (Celery)
|
|
# ----------------------------
|
|
|
|
@main.route('/api/import/trigger', methods=['POST'])
|
|
def trigger_import_celery():
|
|
"""
|
|
Trigger a directory scan and import via Celery.
|
|
|
|
Supports two modes:
|
|
- Quick scan (default): Fast O(1) duplicate detection using pre-loaded lookups.
|
|
Only queues truly new files. Skips pHash comparison for speed.
|
|
- Deep scan: Full reprocessing of all files. Re-extracts metadata,
|
|
runs pHash comparison, optionally regenerates thumbnails and re-applies sidecars.
|
|
|
|
POST params:
|
|
source_dir: Source directory (default: /import)
|
|
dest_dir: Destination directory (default: /images)
|
|
deep: Set to 'true' for deep scan mode
|
|
regenerate_thumbnails: For deep scan, regenerate all thumbnails (default: true)
|
|
reapply_sidecars: For deep scan, re-apply sidecar metadata (default: true)
|
|
verify_hashes: For deep scan, verify content hashes (default: true)
|
|
"""
|
|
try:
|
|
from app.tasks.scan import scan_directory, deep_scan_directory
|
|
|
|
source_dir = request.form.get('source_dir', '/import')
|
|
dest_dir = request.form.get('dest_dir', '/images')
|
|
is_deep = request.form.get('deep', 'false').lower() == 'true'
|
|
|
|
if is_deep:
|
|
# Deep scan with full reprocessing
|
|
regenerate_thumbnails = request.form.get('regenerate_thumbnails', 'true').lower() == 'true'
|
|
reapply_sidecars = request.form.get('reapply_sidecars', 'true').lower() == 'true'
|
|
verify_hashes = request.form.get('verify_hashes', 'true').lower() == 'true'
|
|
|
|
result = deep_scan_directory.delay(
|
|
source_dir, dest_dir,
|
|
regenerate_thumbnails=regenerate_thumbnails,
|
|
reapply_sidecars=reapply_sidecars,
|
|
verify_hashes=verify_hashes
|
|
)
|
|
message = 'Deep import scan started'
|
|
else:
|
|
# Quick scan with optimized lookups
|
|
result = scan_directory.delay(source_dir, dest_dir)
|
|
message = 'Quick import scan started'
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'task_id': result.id,
|
|
'message': message,
|
|
'mode': 'deep' if is_deep else 'quick'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/import/status')
|
|
def import_queue_status():
|
|
"""
|
|
Get current import queue status.
|
|
Returns counts by task status and recent task info.
|
|
"""
|
|
try:
|
|
# Count tasks by status
|
|
status_counts = dict(
|
|
db.session.query(
|
|
ImportTask.status,
|
|
func.count(ImportTask.id)
|
|
).group_by(ImportTask.status).all()
|
|
)
|
|
|
|
# Get active batch info
|
|
active_batch = ImportBatch.query.filter_by(status='running').order_by(
|
|
ImportBatch.started_at.desc()
|
|
).first()
|
|
|
|
# Recent tasks (last 20)
|
|
recent_tasks = ImportTask.query.order_by(
|
|
ImportTask.created_at.desc()
|
|
).limit(20).all()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'status_counts': {
|
|
'pending': status_counts.get('pending', 0),
|
|
'queued': status_counts.get('queued', 0),
|
|
'processing': status_counts.get('processing', 0),
|
|
'complete': status_counts.get('complete', 0),
|
|
'failed': status_counts.get('failed', 0),
|
|
'skipped': status_counts.get('skipped', 0),
|
|
},
|
|
'active_batch': {
|
|
'id': active_batch.id,
|
|
'source_directory': active_batch.source_directory,
|
|
'total_files': active_batch.total_files,
|
|
'processed_files': active_batch.processed_files,
|
|
'imported_count': active_batch.imported_count,
|
|
'skipped_count': active_batch.skipped_count,
|
|
'failed_count': active_batch.failed_count,
|
|
'started_at': active_batch.started_at.isoformat() if active_batch.started_at else None
|
|
} if active_batch else None,
|
|
'recent_tasks': [
|
|
{
|
|
'id': t.id,
|
|
'source_path': os.path.basename(t.source_path) if t.source_path else None,
|
|
'task_type': t.task_type,
|
|
'status': t.status,
|
|
'error_message': t.error_message[:100] if t.error_message else None,
|
|
'created_at': t.created_at.isoformat() if t.created_at else None
|
|
}
|
|
for t in recent_tasks
|
|
]
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/import/tasks')
|
|
def list_import_tasks():
|
|
"""
|
|
List import tasks with filtering and pagination.
|
|
|
|
Query params:
|
|
status: Filter by status (comma-separated, e.g., 'failed,skipped')
|
|
task_type: Filter by task type ('import_image', 'import_archive')
|
|
search: Search in source_path or error_message
|
|
limit: Max results (default 50, max 200)
|
|
offset: Skip first N results
|
|
sort: Sort field ('created_at', 'completed_at', default: 'created_at')
|
|
order: Sort order ('asc', 'desc', default: 'desc')
|
|
"""
|
|
try:
|
|
# Parse parameters
|
|
status_filter = request.args.get('status', '')
|
|
task_type = request.args.get('task_type', '')
|
|
search = request.args.get('search', '')
|
|
limit = min(int(request.args.get('limit', 50)), 200)
|
|
offset = int(request.args.get('offset', 0))
|
|
sort_field = request.args.get('sort', 'created_at')
|
|
sort_order = request.args.get('order', 'desc')
|
|
|
|
# Build query
|
|
query = ImportTask.query
|
|
|
|
# Filter by status
|
|
if status_filter:
|
|
statuses = [s.strip() for s in status_filter.split(',') if s.strip()]
|
|
if statuses:
|
|
query = query.filter(ImportTask.status.in_(statuses))
|
|
|
|
# Filter by task type
|
|
if task_type:
|
|
query = query.filter_by(task_type=task_type)
|
|
|
|
# Search in source_path or error_message
|
|
if search:
|
|
search_pattern = f'%{search}%'
|
|
query = query.filter(
|
|
db.or_(
|
|
ImportTask.source_path.ilike(search_pattern),
|
|
ImportTask.error_message.ilike(search_pattern)
|
|
)
|
|
)
|
|
|
|
# Get total count before pagination
|
|
total = query.count()
|
|
|
|
# Sort
|
|
sort_col = getattr(ImportTask, sort_field, ImportTask.created_at)
|
|
if sort_order == 'asc':
|
|
query = query.order_by(sort_col.asc())
|
|
else:
|
|
query = query.order_by(sort_col.desc())
|
|
|
|
# Paginate
|
|
tasks = query.offset(offset).limit(limit).all()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'total': total,
|
|
'offset': offset,
|
|
'limit': limit,
|
|
'tasks': [
|
|
{
|
|
'id': t.id,
|
|
'source_path': t.source_path,
|
|
'filename': os.path.basename(t.source_path) if t.source_path else None,
|
|
'task_type': t.task_type,
|
|
'status': t.status,
|
|
'error_message': t.error_message, # Full error message
|
|
'file_size': t.file_size,
|
|
'retry_count': t.retry_count,
|
|
'created_at': t.created_at.isoformat() if t.created_at else None,
|
|
'completed_at': t.completed_at.isoformat() if t.completed_at else None
|
|
}
|
|
for t in tasks
|
|
]
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/import/task/<int:task_id>')
|
|
def get_import_task(task_id):
|
|
"""
|
|
Get detailed status for a specific import task.
|
|
"""
|
|
task = ImportTask.query.get_or_404(task_id)
|
|
|
|
# Check Celery task status if still processing
|
|
celery_status = None
|
|
if task.celery_task_id:
|
|
try:
|
|
from app.celery_app import celery
|
|
result = celery.AsyncResult(task.celery_task_id)
|
|
celery_status = result.status
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'task': {
|
|
'id': task.id,
|
|
'source_path': task.source_path,
|
|
'file_hash': task.file_hash,
|
|
'file_size': task.file_size,
|
|
'task_type': task.task_type,
|
|
'status': task.status,
|
|
'celery_status': celery_status,
|
|
'celery_task_id': task.celery_task_id,
|
|
'error_message': task.error_message,
|
|
'retry_count': task.retry_count,
|
|
'context': task.context,
|
|
'batch_id': task.batch_id,
|
|
'result_image_id': task.result_image_id,
|
|
'created_at': task.created_at.isoformat() if task.created_at else None,
|
|
'started_at': task.started_at.isoformat() if task.started_at else None,
|
|
'completed_at': task.completed_at.isoformat() if task.completed_at else None
|
|
}
|
|
})
|
|
|
|
|
|
@main.route('/api/import/retry-failed', methods=['POST'])
|
|
def retry_failed_import_tasks():
|
|
"""
|
|
Re-queue all failed import tasks.
|
|
"""
|
|
try:
|
|
from app.tasks.import_file import import_media_file, import_archive
|
|
|
|
failed_tasks = ImportTask.query.filter_by(status='failed').all()
|
|
retried = 0
|
|
|
|
for task in failed_tasks:
|
|
task.status = 'pending'
|
|
task.error_message = None
|
|
task.retry_count = 0
|
|
task.started_at = None
|
|
task.completed_at = None
|
|
|
|
if task.task_type == 'import_image':
|
|
result = import_media_file.delay(task.id)
|
|
elif task.task_type == 'import_archive':
|
|
result = import_archive.delay(task.id)
|
|
else:
|
|
continue
|
|
|
|
task.celery_task_id = result.id
|
|
task.status = 'queued'
|
|
retried += 1
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'retried': retried
|
|
})
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/import/clear-completed', methods=['POST'])
|
|
def clear_completed_tasks():
|
|
"""
|
|
Clear import tasks from the database with optional age filtering.
|
|
|
|
POST params:
|
|
statuses: Comma-separated statuses to clear (default: complete,skipped)
|
|
older_than_days: Only clear tasks older than N days (default: 0 = all)
|
|
"""
|
|
from datetime import timedelta
|
|
|
|
try:
|
|
# Parse parameters
|
|
statuses_param = request.form.get('statuses', 'complete,skipped')
|
|
older_than_days = int(request.form.get('older_than_days', 0))
|
|
|
|
# Validate statuses
|
|
allowed_statuses = {'complete', 'skipped', 'failed'}
|
|
statuses = [s.strip() for s in statuses_param.split(',') if s.strip() in allowed_statuses]
|
|
if not statuses:
|
|
statuses = ['complete', 'skipped']
|
|
|
|
# Build query
|
|
query = ImportTask.query.filter(ImportTask.status.in_(statuses))
|
|
|
|
# Apply age filter if specified
|
|
if older_than_days > 0:
|
|
cutoff = datetime.utcnow() - timedelta(days=older_than_days)
|
|
query = query.filter(ImportTask.completed_at < cutoff)
|
|
|
|
deleted = query.delete(synchronize_session=False)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'deleted': deleted,
|
|
'statuses': statuses,
|
|
'older_than_days': older_than_days
|
|
})
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
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():
|
|
"""
|
|
Get detailed task statistics with age breakdown.
|
|
"""
|
|
from datetime import timedelta
|
|
|
|
try:
|
|
now = datetime.utcnow()
|
|
today = now - timedelta(days=1)
|
|
week_ago = now - timedelta(days=7)
|
|
|
|
stats = {}
|
|
for status in ['complete', 'skipped', 'failed']:
|
|
total = ImportTask.query.filter_by(status=status).count()
|
|
today_count = ImportTask.query.filter(
|
|
ImportTask.status == status,
|
|
ImportTask.completed_at >= today
|
|
).count()
|
|
week_count = ImportTask.query.filter(
|
|
ImportTask.status == status,
|
|
ImportTask.completed_at >= week_ago,
|
|
ImportTask.completed_at < today
|
|
).count()
|
|
older = total - today_count - week_count
|
|
|
|
stats[status] = {
|
|
'total': total,
|
|
'today': today_count,
|
|
'this_week': week_count,
|
|
'older': older
|
|
}
|
|
|
|
return jsonify({'ok': True, 'stats': stats})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/system/stats')
|
|
def system_stats():
|
|
"""
|
|
Get system-wide statistics for the dashboard.
|
|
|
|
Returns cached stats computed by the update_system_stats Celery task.
|
|
Stats are updated every 5 minutes by Celery Beat.
|
|
|
|
Query params:
|
|
refresh: If 'true', triggers an immediate stats update (async)
|
|
"""
|
|
import json
|
|
|
|
try:
|
|
# Check for refresh request
|
|
if request.args.get('refresh') == 'true':
|
|
from app.tasks.scan import update_system_stats
|
|
update_system_stats.delay()
|
|
|
|
# Get cached stats from AppSettings
|
|
setting = AppSettings.query.filter_by(key='system_stats_cache').first()
|
|
|
|
if setting and setting.value:
|
|
stats = json.loads(setting.value)
|
|
return jsonify({
|
|
'ok': True,
|
|
'cached': True,
|
|
**stats
|
|
})
|
|
|
|
# No cached stats - return empty with flag to indicate first run
|
|
return jsonify({
|
|
'ok': True,
|
|
'cached': False,
|
|
'images': {'total': 0, 'storage_bytes': 0, 'storage_human': '0 B'},
|
|
'tags': {'total': 0, 'by_kind': {}},
|
|
'import_folder': {'pending_files': 0, 'pending_archives': 0},
|
|
'message': 'Stats not yet computed. Will be available shortly.'
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/thumbnails/regenerate', methods=['POST'])
|
|
def regenerate_thumbnails_celery():
|
|
"""
|
|
Trigger thumbnail regeneration for all images via Celery.
|
|
"""
|
|
try:
|
|
from app.tasks.thumbnail import regenerate_all_thumbnails
|
|
|
|
overwrite = request.form.get('overwrite', 'true').lower() == 'true'
|
|
result = regenerate_all_thumbnails.delay(overwrite=overwrite)
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'task_id': result.id,
|
|
'message': 'Thumbnail regeneration started'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/thumbnails/regenerate-missing', methods=['POST'])
|
|
def regenerate_missing_thumbnails_celery():
|
|
"""
|
|
Generate thumbnails only for images that don't have them.
|
|
"""
|
|
try:
|
|
from app.tasks.thumbnail import regenerate_missing_thumbnails
|
|
|
|
result = regenerate_missing_thumbnails.delay()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'task_id': result.id,
|
|
'message': 'Missing thumbnail generation started'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/reapply-artist-tags', methods=['POST'])
|
|
def reapply_artist_tags_from_paths():
|
|
"""
|
|
Re-apply artist tags to all images based on their file paths.
|
|
Artist is extracted from the first directory component after /images/
|
|
(e.g., /images/artist_name/file.jpg -> artist:artist_name)
|
|
"""
|
|
try:
|
|
images = ImageRecord.query.all()
|
|
updated = 0
|
|
skipped = 0
|
|
errors = []
|
|
|
|
for img in images:
|
|
if not img.filepath:
|
|
skipped += 1
|
|
continue
|
|
|
|
# Extract artist from path: /images/{artist}/...
|
|
# Handle both /images/artist/file.jpg and /images/artist/subdir/file.jpg
|
|
path_parts = img.filepath.split('/')
|
|
try:
|
|
# Find 'images' in path and get the next component
|
|
images_idx = path_parts.index('images')
|
|
if images_idx + 1 < len(path_parts):
|
|
artist_name = path_parts[images_idx + 1]
|
|
else:
|
|
skipped += 1
|
|
continue
|
|
except ValueError:
|
|
# 'images' not in path, try alternative format
|
|
# Maybe it's a Windows path or different structure
|
|
skipped += 1
|
|
continue
|
|
|
|
if not artist_name or artist_name == '':
|
|
skipped += 1
|
|
continue
|
|
|
|
# Create or get the artist tag
|
|
tag_name = f"artist:{artist_name}"
|
|
tag = Tag.query.filter_by(name=tag_name).first()
|
|
if not tag:
|
|
tag = Tag(name=tag_name, kind='artist')
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
|
|
# Add tag if not already present
|
|
if tag not in img.tags:
|
|
img.tags.append(tag)
|
|
updated += 1
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'updated': updated,
|
|
'skipped': skipped,
|
|
'total': len(images),
|
|
'errors': errors if errors else None
|
|
})
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/cleanup-orphaned-tags', methods=['POST'])
|
|
def cleanup_orphaned_tags():
|
|
"""
|
|
Delete tags that aren't attached to any images.
|
|
Returns the count and names of deleted tags.
|
|
"""
|
|
from sqlalchemy import func
|
|
|
|
try:
|
|
# Find tags with no associated images
|
|
# A tag is orphaned if it has no entries in the image_tags junction table
|
|
orphaned = (
|
|
Tag.query
|
|
.outerjoin(image_tags)
|
|
.group_by(Tag.id)
|
|
.having(func.count(image_tags.c.image_id) == 0)
|
|
.all()
|
|
)
|
|
|
|
deleted_tags = []
|
|
for tag in orphaned:
|
|
deleted_tags.append({'name': tag.name, 'kind': tag.kind})
|
|
db.session.delete(tag)
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'deleted_count': len(deleted_tags),
|
|
'deleted_tags': deleted_tags
|
|
})
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({'ok': False, 'error': str(e)}), 500
|
|
|
|
|
|
@main.route('/api/celery/status')
|
|
def celery_cluster_status():
|
|
"""
|
|
Get Celery cluster status (workers, queues, etc.).
|
|
"""
|
|
try:
|
|
from app.celery_app import celery
|
|
|
|
inspect = celery.control.inspect()
|
|
|
|
# Get worker stats
|
|
active = inspect.active() or {}
|
|
scheduled = inspect.scheduled() or {}
|
|
reserved = inspect.reserved() or {}
|
|
stats = inspect.stats() or {}
|
|
|
|
# Count active tasks
|
|
active_count = sum(len(tasks) for tasks in active.values())
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'workers': list(stats.keys()),
|
|
'worker_count': len(stats),
|
|
'active_tasks': active_count,
|
|
'active': active,
|
|
'scheduled': scheduled,
|
|
'reserved': reserved
|
|
})
|
|
except Exception as e:
|
|
return jsonify({
|
|
'ok': False,
|
|
'error': str(e),
|
|
'message': 'Celery may not be running'
|
|
}), 503
|
|
|
|
|
|
@main.route('/api/import/batches')
|
|
def list_import_batches():
|
|
"""
|
|
List recent import batches with statistics.
|
|
"""
|
|
limit = request.args.get('limit', 10, type=int)
|
|
|
|
batches = ImportBatch.query.order_by(
|
|
ImportBatch.started_at.desc()
|
|
).limit(limit).all()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'batches': [
|
|
{
|
|
'id': b.id,
|
|
'source_directory': b.source_directory,
|
|
'status': b.status,
|
|
'total_files': b.total_files,
|
|
'processed_files': b.processed_files,
|
|
'imported_count': b.imported_count,
|
|
'skipped_count': b.skipped_count,
|
|
'failed_count': b.failed_count,
|
|
'started_at': b.started_at.isoformat() if b.started_at else None,
|
|
'completed_at': b.completed_at.isoformat() if b.completed_at else None
|
|
}
|
|
for b in batches
|
|
]
|
|
})
|
|
|
|
|
|
# =============================================
|
|
# Duplicate Detection API
|
|
# =============================================
|
|
|
|
@main.route('/api/duplicates/scan', methods=['POST'])
|
|
def scan_duplicates():
|
|
"""
|
|
Scan for duplicate images using perceptual hash comparison.
|
|
Returns groups of similar images within the specified hamming distance.
|
|
"""
|
|
import imagehash
|
|
|
|
threshold = request.form.get('threshold', 10, type=int)
|
|
limit = request.form.get('limit', 100, type=int)
|
|
|
|
# Get all images with perceptual hashes
|
|
images = ImageRecord.query.filter(
|
|
ImageRecord.perceptual_hash.isnot(None)
|
|
).all()
|
|
|
|
if not images:
|
|
return jsonify({'ok': True, 'groups': [], 'message': 'No images with perceptual hashes found'})
|
|
|
|
# Build hash lookup
|
|
hash_data = []
|
|
for img in images:
|
|
try:
|
|
phash = imagehash.hex_to_hash(img.perceptual_hash)
|
|
hash_data.append({
|
|
'id': img.id,
|
|
'hash': phash,
|
|
'width': img.width,
|
|
'height': img.height,
|
|
'filename': img.filename,
|
|
'filepath': img.filepath,
|
|
'thumb_path': img.thumb_path,
|
|
'file_size': img.file_size
|
|
})
|
|
except Exception:
|
|
continue
|
|
|
|
# Find duplicate groups
|
|
duplicate_groups = []
|
|
processed = set()
|
|
|
|
for i, img1 in enumerate(hash_data):
|
|
if img1['id'] in processed:
|
|
continue
|
|
|
|
group = [img1]
|
|
|
|
for j, img2 in enumerate(hash_data[i+1:], start=i+1):
|
|
if img2['id'] in processed:
|
|
continue
|
|
|
|
distance = img1['hash'] - img2['hash']
|
|
if distance <= threshold:
|
|
group.append(img2)
|
|
processed.add(img2['id'])
|
|
|
|
if len(group) > 1:
|
|
processed.add(img1['id'])
|
|
# Sort group by resolution (largest first)
|
|
group.sort(key=lambda x: x['width'] * x['height'], reverse=True)
|
|
duplicate_groups.append(group)
|
|
|
|
if len(duplicate_groups) >= limit:
|
|
break
|
|
|
|
# Format response
|
|
result_groups = []
|
|
for group in duplicate_groups:
|
|
result_groups.append([
|
|
{
|
|
'id': img['id'],
|
|
'filename': img['filename'],
|
|
'width': img['width'],
|
|
'height': img['height'],
|
|
'file_size': img['file_size'],
|
|
'thumb_url': url_for('main.serve_image',
|
|
filename=(img['thumb_path'] or img['filepath']).replace('/images/', ''))
|
|
}
|
|
for img in group
|
|
])
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'groups': result_groups,
|
|
'total_groups': len(result_groups),
|
|
'threshold': threshold
|
|
})
|
|
|
|
|
|
@main.route('/api/duplicates/delete', methods=['POST'])
|
|
def delete_duplicates():
|
|
"""
|
|
Delete specified duplicate images.
|
|
Expects JSON body with 'image_ids' array.
|
|
"""
|
|
data = request.get_json() or {}
|
|
image_ids = data.get('image_ids', [])
|
|
|
|
if not image_ids:
|
|
return jsonify({'ok': False, 'error': 'No image IDs provided'})
|
|
|
|
deleted_count = 0
|
|
errors = []
|
|
|
|
for image_id in image_ids:
|
|
try:
|
|
image = ImageRecord.query.get(image_id)
|
|
if not image:
|
|
errors.append(f"Image {image_id} not found")
|
|
continue
|
|
|
|
# Delete physical files
|
|
if image.filepath and os.path.exists(image.filepath):
|
|
os.remove(image.filepath)
|
|
if image.thumb_path and os.path.exists(image.thumb_path):
|
|
os.remove(image.thumb_path)
|
|
|
|
# Delete database record
|
|
db.session.delete(image)
|
|
deleted_count += 1
|
|
|
|
except Exception as e:
|
|
errors.append(f"Error deleting {image_id}: {str(e)}")
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'deleted_count': deleted_count,
|
|
'errors': errors if errors else None
|
|
})
|
|
|
|
|
|
# =============================================
|
|
# Bulk Tag Operations API
|
|
# =============================================
|
|
|
|
@main.route('/api/images/common-tags', methods=['POST'])
|
|
def get_common_tags():
|
|
"""
|
|
Get tags that are common to all specified images.
|
|
Expects JSON body with 'image_ids' array.
|
|
Returns tags that appear on EVERY selected image.
|
|
"""
|
|
data = request.get_json() or {}
|
|
image_ids = data.get('image_ids', [])
|
|
|
|
if not image_ids:
|
|
return jsonify({'ok': True, 'tags': []})
|
|
|
|
# Find tags that appear on ALL selected images
|
|
# We use a subquery approach: count occurrences of each tag and filter
|
|
# for tags where the count equals the number of images
|
|
num_images = len(image_ids)
|
|
|
|
# Query: get tags where the count of associated images (from our list) equals num_images
|
|
common_tags = db.session.query(Tag).join(image_tags).filter(
|
|
image_tags.c.image_id.in_(image_ids)
|
|
).group_by(Tag.id).having(
|
|
func.count(image_tags.c.image_id) == num_images
|
|
).order_by(Tag.name).all()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'tags': [
|
|
{'id': t.id, 'name': t.name, 'kind': t.kind}
|
|
for t in common_tags
|
|
]
|
|
})
|
|
|
|
|
|
@main.route('/api/images/bulk-add-tag', methods=['POST'])
|
|
def bulk_add_tag():
|
|
"""
|
|
Bulk-apply one tag to many images in a single transaction.
|
|
|
|
Accepts the same 'kind:name' shortcut as /image/<id>/tags/add. For
|
|
character tags, accepts optional fandom_id / fandom_name just like
|
|
the single-image endpoint.
|
|
"""
|
|
from app.utils.tag_prefix import parse_kind_prefix
|
|
from app.utils.tag_names import underscores_to_spaces
|
|
|
|
data = request.get_json() or {}
|
|
image_ids = data.get("image_ids") or []
|
|
raw = (data.get("tag_name") or "").strip()
|
|
if not image_ids or not raw:
|
|
return jsonify(ok=False, error="image_ids and tag_name required"), 400
|
|
|
|
kind, name = parse_kind_prefix(raw)
|
|
if kind is None:
|
|
kind = "user"
|
|
name = underscores_to_spaces(name)
|
|
elif kind in ("character", "fandom"):
|
|
name = normalize_display_name(name)
|
|
|
|
fandom_tag = None
|
|
if kind == "character":
|
|
fandom_id = data.get("fandom_id")
|
|
fandom_name = (data.get("fandom_name") or "").strip() or None
|
|
if fandom_id:
|
|
fandom_tag = Tag.query.filter_by(id=fandom_id, kind="fandom").first()
|
|
if fandom_tag is None:
|
|
return jsonify(ok=False, error="fandom_not_found"), 400
|
|
elif fandom_name:
|
|
fandom_tag = _ensure_fandom_tag(fandom_name)
|
|
|
|
tag = Tag.query.filter_by(
|
|
kind="character",
|
|
name=name,
|
|
fandom_id=fandom_tag.id if fandom_tag else None,
|
|
).first()
|
|
if tag is None:
|
|
tag = Tag(
|
|
kind="character",
|
|
name=name,
|
|
fandom_id=fandom_tag.id if fandom_tag else None,
|
|
)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
else:
|
|
tag = Tag.query.filter_by(kind=kind, name=name).first()
|
|
if tag is None:
|
|
tag = Tag(kind=kind, name=name)
|
|
db.session.add(tag)
|
|
db.session.flush()
|
|
|
|
added_count = 0
|
|
for image_id in image_ids:
|
|
img = ImageRecord.query.get(image_id)
|
|
if img is None:
|
|
continue
|
|
if tag not in img.tags:
|
|
img.tags.append(tag)
|
|
added_count += 1
|
|
if fandom_tag and fandom_tag not in img.tags:
|
|
img.tags.append(fandom_tag)
|
|
|
|
db.session.commit()
|
|
return jsonify(
|
|
ok=True,
|
|
added_count=added_count,
|
|
total=len(image_ids),
|
|
tag={
|
|
"id": tag.id,
|
|
"name": tag.name,
|
|
"display_name": tag.display_name,
|
|
"kind": tag.kind,
|
|
},
|
|
)
|
|
|
|
|
|
@main.route('/api/images/bulk-remove-tag', methods=['POST'])
|
|
def bulk_remove_tag():
|
|
"""
|
|
Remove a tag from multiple images at once.
|
|
Expects JSON body with 'image_ids' array and 'tag_name' string.
|
|
"""
|
|
data = request.get_json() or {}
|
|
image_ids = data.get('image_ids', [])
|
|
tag_name = (data.get('tag_name') or '').strip()
|
|
|
|
if not image_ids:
|
|
return jsonify({'ok': False, 'error': 'No image IDs provided'}), 400
|
|
|
|
if not tag_name:
|
|
return jsonify({'ok': False, 'error': 'No tag name provided'}), 400
|
|
|
|
# Find the tag
|
|
tag = Tag.query.filter_by(name=tag_name).first()
|
|
if not tag:
|
|
return jsonify({'ok': True, 'removed_count': 0, 'message': 'Tag not found'})
|
|
|
|
# Get images that have this tag
|
|
images = ImageRecord.query.filter(
|
|
ImageRecord.id.in_(image_ids)
|
|
).all()
|
|
|
|
removed_count = 0
|
|
for img in images:
|
|
if tag in img.tags:
|
|
img.tags.remove(tag)
|
|
removed_count += 1
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'ok': True,
|
|
'removed_count': removed_count
|
|
})
|
|
|
|
|
|
# =============================================
|
|
# Series / Reader API
|
|
# =============================================
|
|
|
|
@main.route('/read/<int:tag_id>')
|
|
def read_series(tag_id):
|
|
"""
|
|
Reader view for a series - displays images in page order.
|
|
Supports jump-to-page via ?page=N parameter.
|
|
"""
|
|
from app.models import SeriesPage
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
if tag.kind != 'series':
|
|
return "Not a series tag", 400
|
|
|
|
# Get all pages in order
|
|
pages = SeriesPage.query.filter_by(series_tag_id=tag_id).order_by(
|
|
SeriesPage.page_number
|
|
).all()
|
|
|
|
# Get starting page from query param
|
|
start_page = request.args.get('page', 1, type=int)
|
|
|
|
# Get series name (strip prefix if present)
|
|
series_name = tag.name
|
|
|
|
return render_template(
|
|
'reader.html',
|
|
tag=tag,
|
|
series_name=series_name,
|
|
pages=pages,
|
|
start_page=start_page,
|
|
total_pages=len(pages)
|
|
)
|
|
|
|
|
|
@main.get('/api/series/<int:tag_id>/pages')
|
|
def get_series_pages(tag_id):
|
|
"""Get all pages in a series with their images."""
|
|
from app.models import SeriesPage
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
if tag.kind != 'series':
|
|
return jsonify(ok=False, error="Not a series tag"), 400
|
|
|
|
pages = SeriesPage.query.filter_by(series_tag_id=tag_id).order_by(
|
|
SeriesPage.page_number
|
|
).all()
|
|
|
|
return jsonify(ok=True, pages=[
|
|
{
|
|
'id': p.id,
|
|
'page_number': p.page_number,
|
|
'image_id': p.image_id,
|
|
'image': {
|
|
'id': p.image.id,
|
|
'filename': p.image.filename,
|
|
'filepath': p.image.filepath,
|
|
'thumb_path': p.image.thumb_path,
|
|
'width': p.image.width,
|
|
'height': p.image.height,
|
|
}
|
|
}
|
|
for p in pages
|
|
], total_pages=len(pages))
|
|
|
|
|
|
@main.post('/api/series/<int:tag_id>/pages/add')
|
|
def add_series_page(tag_id):
|
|
"""
|
|
Add an image to a series with a specific page number.
|
|
Expects form data: image_id, page_number
|
|
"""
|
|
from app.models import SeriesPage
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
if tag.kind != 'series':
|
|
return jsonify(ok=False, error="Not a series tag"), 400
|
|
|
|
image_id = request.form.get('image_id', type=int)
|
|
page_number = request.form.get('page_number', type=int)
|
|
|
|
if not image_id or page_number is None:
|
|
return jsonify(ok=False, error="image_id and page_number required"), 400
|
|
|
|
# Check if image exists
|
|
image = ImageRecord.query.get(image_id)
|
|
if not image:
|
|
return jsonify(ok=False, error="Image not found"), 404
|
|
|
|
# Check if image is already in a series
|
|
existing = SeriesPage.query.filter_by(image_id=image_id).first()
|
|
if existing:
|
|
return jsonify(ok=False, error="Image is already in a series",
|
|
existing_series_id=existing.series_tag_id), 409
|
|
|
|
# Check if page number is already taken
|
|
page_exists = SeriesPage.query.filter_by(
|
|
series_tag_id=tag_id, page_number=page_number
|
|
).first()
|
|
if page_exists:
|
|
return jsonify(ok=False, error=f"Page {page_number} already exists"), 409
|
|
|
|
# Create the series page
|
|
sp = SeriesPage(
|
|
series_tag_id=tag_id,
|
|
image_id=image_id,
|
|
page_number=page_number
|
|
)
|
|
db.session.add(sp)
|
|
|
|
# Also ensure the image has the series tag
|
|
if tag not in image.tags:
|
|
image.tags.append(tag)
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True, page={
|
|
'id': sp.id,
|
|
'page_number': sp.page_number,
|
|
'image_id': sp.image_id
|
|
})
|
|
|
|
|
|
@main.post('/api/series/<int:tag_id>/pages/bulk-add')
|
|
def bulk_add_series_pages(tag_id):
|
|
"""
|
|
Add multiple images to a series with sequential page numbers.
|
|
Expects JSON: { image_ids: [id1, id2, ...], start_page: N }
|
|
Images are added in order starting from start_page.
|
|
"""
|
|
from app.models import SeriesPage
|
|
|
|
tag = Tag.query.get_or_404(tag_id)
|
|
|
|
if tag.kind != 'series':
|
|
return jsonify(ok=False, error="Not a series tag"), 400
|
|
|
|
data = request.get_json() or {}
|
|
image_ids = data.get('image_ids', [])
|
|
start_page = data.get('start_page', 1)
|
|
|
|
if not image_ids:
|
|
return jsonify(ok=False, error="No image IDs provided"), 400
|
|
|
|
added = []
|
|
skipped = []
|
|
current_page = start_page
|
|
|
|
for image_id in image_ids:
|
|
image = ImageRecord.query.get(image_id)
|
|
if not image:
|
|
skipped.append({'image_id': image_id, 'reason': 'not found'})
|
|
continue
|
|
|
|
# Check if image is already in a series
|
|
existing = SeriesPage.query.filter_by(image_id=image_id).first()
|
|
if existing:
|
|
skipped.append({'image_id': image_id, 'reason': 'already in series'})
|
|
continue
|
|
|
|
# Find next available page number
|
|
while SeriesPage.query.filter_by(
|
|
series_tag_id=tag_id, page_number=current_page
|
|
).first():
|
|
current_page += 1
|
|
|
|
sp = SeriesPage(
|
|
series_tag_id=tag_id,
|
|
image_id=image_id,
|
|
page_number=current_page
|
|
)
|
|
db.session.add(sp)
|
|
|
|
# Ensure image has the series tag
|
|
if tag not in image.tags:
|
|
image.tags.append(tag)
|
|
|
|
added.append({'image_id': image_id, 'page_number': current_page})
|
|
current_page += 1
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True, added=added, skipped=skipped)
|
|
|
|
|
|
@main.post('/api/series/page/<int:page_id>/update')
|
|
def update_series_page(page_id):
|
|
"""Update a series page's page number."""
|
|
from app.models import SeriesPage
|
|
|
|
sp = SeriesPage.query.get_or_404(page_id)
|
|
new_page_number = request.form.get('page_number', type=int)
|
|
|
|
if new_page_number is None:
|
|
return jsonify(ok=False, error="page_number required"), 400
|
|
|
|
# Check if the new page number is taken
|
|
existing = SeriesPage.query.filter(
|
|
SeriesPage.series_tag_id == sp.series_tag_id,
|
|
SeriesPage.page_number == new_page_number,
|
|
SeriesPage.id != page_id
|
|
).first()
|
|
|
|
if existing:
|
|
return jsonify(ok=False, error=f"Page {new_page_number} already exists"), 409
|
|
|
|
sp.page_number = new_page_number
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True, page={
|
|
'id': sp.id,
|
|
'page_number': sp.page_number,
|
|
'image_id': sp.image_id
|
|
})
|
|
|
|
|
|
@main.post('/api/series/page/<int:page_id>/delete')
|
|
def delete_series_page(page_id):
|
|
"""Remove an image from a series (doesn't delete the image)."""
|
|
from app.models import SeriesPage
|
|
|
|
sp = SeriesPage.query.get_or_404(page_id)
|
|
db.session.delete(sp)
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True)
|
|
|
|
|
|
@main.get('/api/image/<int:image_id>/series-info')
|
|
def get_image_series_info(image_id):
|
|
"""Check if an image is part of a series and return its page info."""
|
|
from app.models import SeriesPage
|
|
|
|
sp = SeriesPage.query.filter_by(image_id=image_id).first()
|
|
|
|
if not sp:
|
|
# Return any series tags the image has (for auto-select in dropdown)
|
|
image = ImageRecord.query.get(image_id)
|
|
series_tags = []
|
|
if image:
|
|
series_tags = [
|
|
{'id': t.id, 'name': t.name}
|
|
for t in image.tags if t.kind == 'series'
|
|
]
|
|
return jsonify(ok=True, in_series=False, series_tags=series_tags)
|
|
|
|
return jsonify(ok=True, in_series=True, series_page={
|
|
'id': sp.id,
|
|
'page_number': sp.page_number,
|
|
'series_tag_id': sp.series_tag_id,
|
|
'series_name': sp.series_tag.name
|
|
})
|
|
|
|
|
|
@main.post('/api/image/<int:image_id>/add-to-series')
|
|
def add_image_to_series(image_id):
|
|
"""Add or update an image in a series with a specific page number.
|
|
|
|
Handles three cases:
|
|
1. Image not in any series -> create new SeriesPage
|
|
2. Image already in this series -> update page number
|
|
3. Image in a different series -> move to new series
|
|
"""
|
|
from app.models import SeriesPage, Tag
|
|
|
|
data = request.get_json() or {}
|
|
series_tag_id = data.get('series_tag_id')
|
|
page_number = data.get('page_number')
|
|
|
|
if not series_tag_id:
|
|
return jsonify(ok=False, error="series_tag_id required"), 400
|
|
if not page_number:
|
|
return jsonify(ok=False, error="page_number required"), 400
|
|
|
|
# Verify the series tag exists and is a series
|
|
series_tag = Tag.query.get(series_tag_id)
|
|
if not series_tag:
|
|
return jsonify(ok=False, error="Series not found"), 404
|
|
if series_tag.kind != 'series':
|
|
return jsonify(ok=False, error="Tag is not a series"), 400
|
|
|
|
# Check if image is already in this series
|
|
existing_in_target = SeriesPage.query.filter_by(
|
|
series_tag_id=series_tag_id,
|
|
image_id=image_id
|
|
).first()
|
|
|
|
# Check if image is in any other series
|
|
existing_in_other = SeriesPage.query.filter(
|
|
SeriesPage.image_id == image_id,
|
|
SeriesPage.series_tag_id != series_tag_id
|
|
).first()
|
|
|
|
if existing_in_target:
|
|
# Case 2: Already in this series - update page number
|
|
if existing_in_target.page_number == page_number:
|
|
# No change needed
|
|
return jsonify(ok=True, series_page={
|
|
'id': existing_in_target.id,
|
|
'page_number': existing_in_target.page_number,
|
|
'series_tag_id': existing_in_target.series_tag_id,
|
|
'series_name': series_tag.name
|
|
})
|
|
|
|
# Check if new page number is taken by another image
|
|
page_taken = SeriesPage.query.filter(
|
|
SeriesPage.series_tag_id == series_tag_id,
|
|
SeriesPage.page_number == page_number,
|
|
SeriesPage.image_id != image_id
|
|
).first()
|
|
if page_taken:
|
|
return jsonify(ok=False, error=f"Page {page_number} already exists in this series"), 409
|
|
|
|
existing_in_target.page_number = page_number
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True, series_page={
|
|
'id': existing_in_target.id,
|
|
'page_number': existing_in_target.page_number,
|
|
'series_tag_id': existing_in_target.series_tag_id,
|
|
'series_name': series_tag.name
|
|
})
|
|
|
|
# Check if page number is taken in target series
|
|
page_taken = SeriesPage.query.filter_by(
|
|
series_tag_id=series_tag_id,
|
|
page_number=page_number
|
|
).first()
|
|
if page_taken:
|
|
return jsonify(ok=False, error=f"Page {page_number} already exists in this series"), 409
|
|
|
|
if existing_in_other:
|
|
# Case 3: In a different series - move to new series
|
|
db.session.delete(existing_in_other)
|
|
|
|
# Case 1 & 3: Create new series page
|
|
sp = SeriesPage(
|
|
series_tag_id=series_tag_id,
|
|
image_id=image_id,
|
|
page_number=page_number
|
|
)
|
|
db.session.add(sp)
|
|
db.session.commit()
|
|
|
|
return jsonify(ok=True, series_page={
|
|
'id': sp.id,
|
|
'page_number': sp.page_number,
|
|
'series_tag_id': sp.series_tag_id,
|
|
'series_name': series_tag.name
|
|
})
|