"""Gallery API: cursor scroll, timeline, jump, image detail, facets.""" from datetime import UTC, datetime, timedelta from quart import Blueprint, jsonify, request from ..extensions import get_session from ..services.gallery_service import GalleryService gallery_bp = Blueprint("gallery", __name__, url_prefix="/api/gallery") def _image_json(i): """Serialize a GalleryImage for the scroll/similar list responses.""" return { "id": i.id, "sha256": i.sha256, "mime": i.mime, "width": i.width, "height": i.height, "created_at": i.created_at.isoformat(), "posted_at": i.posted_at.isoformat() if i.posted_at else None, "thumbnail_url": i.thumbnail_url, "artist": i.artist, } def _parse_date(raw): """Parse a YYYY-MM-DD query value to a UTC midnight datetime, or None. Raises ValueError (→ 400) on a malformed value.""" if not raw: return None return datetime.strptime(raw, "%Y-%m-%d").replace(tzinfo=UTC) def _parse_filters(): """Parse the composable gallery filters from query args, returning ``(filters_dict, sort)``. Raises ValueError (→ 400) on malformed ids/dates. The structured tag filter (#6) is AND-of-OR plus exclusions: - `tag_id` accepts a single id or a comma-separated list — all ANDed (the include common case; back-compat). - `tag_or` is REPEATABLE; each instance is a comma-separated OR-group, and the image must match at least one tag from EACH group (groups ANDed). - `tag_not` is a comma-separated exclude list (image must carry none). `media` is image|video; `sort` is newest|oldest|posted_new|posted_old (default posted_new); `platform` selects one platform (or the UNSOURCED_PLATFORM sentinel); `untagged`/`no_artist` are boolean flags; `date_from`/`date_to` are inclusive calendar-day bounds (date_to is widened by a day so the whole day is covered by the service's half-open `< date_to`).""" tag_raw = request.args.get("tag_id") tag_ids = ( [int(x) for x in tag_raw.split(",") if x.strip()] if tag_raw else None ) or None tag_or_groups = [ grp for raw in request.args.getlist("tag_or") if (grp := [int(x) for x in raw.split(",") if x.strip()]) ] or None not_raw = request.args.get("tag_not") tag_exclude = ( [int(x) for x in not_raw.split(",") if x.strip()] if not_raw else None ) or None post_id_raw = request.args.get("post_id") post_id = int(post_id_raw) if post_id_raw else None artist_id_raw = request.args.get("artist_id") artist_id = int(artist_id_raw) if artist_id_raw else None media = request.args.get("media") media_type = media if media in ("image", "video") else None # newest/oldest key off effective_date (primary post / download); posted_new/ # posted_old off earliest_post_date (original publish across all posts). The # default is posted_new so the grid leads with original publish date, not the # download/repost the primary post points at (operator-flagged 2026-07-01). sort = request.args.get("sort") sort = sort if sort in ("newest", "oldest", "posted_new", "posted_old") else "posted_new" platform = request.args.get("platform") or None untagged = request.args.get("untagged") in ("1", "true", "yes") no_artist = request.args.get("no_artist") in ("1", "true", "yes") # Show the presentation chrome (banner / editor screenshot) that the default # gallery hides — the Hidden view sets this (milestone 141). include_hidden = request.args.get("include_hidden") in ("1", "true", "yes") date_from = _parse_date(request.args.get("date_from")) date_to = _parse_date(request.args.get("date_to")) if date_to is not None: date_to += timedelta(days=1) # inclusive of the date_to calendar day filters = { "tag_ids": tag_ids, "post_id": post_id, "artist_id": artist_id, "media_type": media_type, "tag_or_groups": tag_or_groups, "tag_exclude": tag_exclude, "platform": platform, "untagged": untagged, "no_artist": no_artist, "date_from": date_from, "date_to": date_to, "include_hidden": include_hidden, } return filters, sort @gallery_bp.route("/scroll", methods=["GET"]) async def scroll(): cursor = request.args.get("cursor") or None try: limit = int(request.args.get("limit", "50")) filters, sort = _parse_filters() except ValueError: return jsonify({"error": "invalid filter or limit parameter"}), 400 async with get_session() as session: svc = GalleryService(session) try: page = await svc.scroll( cursor=cursor, limit=limit, sort=sort, **filters, ) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify( { "images": [_image_json(i) for i in page.images], "next_cursor": page.next_cursor, "date_groups": [ {"year": y, "month": m, "image_ids": ids} for y, m, ids in page.date_groups ], } ) @gallery_bp.route("/similar", methods=["GET"]) async def similar(): """Visual "more like this": images ranked by cosine distance to the `similar_to` image's embedding. Composes with the scope filters (AND) but ignores post_id and sort. Bounded top-N, no cursor.""" try: similar_to = int(request.args["similar_to"]) limit = int(request.args.get("limit", "100")) filters, _sort = _parse_filters() except (KeyError, ValueError): return jsonify({"error": "similar_to query param required"}), 400 # post_id is the exclusive post-detail view — not a similarity scope. scope = {k: v for k, v in filters.items() if k != "post_id"} async with get_session() as session: svc = GalleryService(session) try: images = await svc.similar(image_id=similar_to, limit=limit, **scope) except ValueError as exc: return jsonify({"error": str(exc)}), 400 if images is None: return jsonify({"error": "not found"}), 404 return jsonify( { "images": [_image_json(i) for i in images], "next_cursor": None, "date_groups": [], } ) @gallery_bp.route("/timeline", methods=["GET"]) async def timeline(): try: filters, _sort = _parse_filters() except ValueError: return jsonify({"error": "invalid filter parameter"}), 400 async with get_session() as session: svc = GalleryService(session) try: buckets = await svc.timeline(**filters) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify( [{"year": b.year, "month": b.month, "count": b.count} for b in buckets] ) @gallery_bp.route("/facets", methods=["GET"]) async def facets(): try: filters, _sort = _parse_filters() except ValueError: return jsonify({"error": "invalid filter parameter"}), 400 async with get_session() as session: svc = GalleryService(session) try: f = await svc.facets(**filters) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify( { "total": f.total, "platforms": f.platforms, "untagged": f.untagged, "no_artist": f.no_artist, "date_min": f.date_min.isoformat() if f.date_min else None, "date_max": f.date_max.isoformat() if f.date_max else None, } ) @gallery_bp.route("/jump", methods=["GET"]) async def jump(): try: year = int(request.args["year"]) month = int(request.args["month"]) filters, sort = _parse_filters() except (KeyError, ValueError): return jsonify({"error": "year and month query params required"}), 400 async with get_session() as session: svc = GalleryService(session) try: cursor = await svc.jump_cursor( year=year, month=month, sort=sort, **filters, ) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify({"cursor": cursor}) @gallery_bp.route("/image/", methods=["GET"]) async def image_detail(image_id: int): async with get_session() as session: svc = GalleryService(session) payload = await svc.get_image_with_tags(image_id) if payload is None: return jsonify({"error": "not found"}), 404 return jsonify(payload)