odysseus-gallery

#!/usr/bin/env python3
"""odysseus-gallery — shell wrapper for the photo / image gallery.

Read-only by default (upload + edit operations need multipart payloads
that are awkward at the shell). Filters images by tag, prompt, favorite,
album, and surfaces EXIF metadata.

    odysseus-gallery list --limit 20 --favorites
    odysseus-gallery show IMAGE_ID
    odysseus-gallery albums | jq -r '.[] | .name'
    odysseus-gallery search "sunset"          # matches prompt + tags
    odysseus-gallery delete IMAGE_ID          # soft-delete (is_active=False)
"""

from __future__ import annotations
import sys
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
quiet_logs()

import argparse, json, logging, os, sys
from pathlib import Path

try:
    from core.database import SessionLocal, GalleryImage, GalleryAlbum
    quiet_logs()
except ModuleNotFoundError as e:
    sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
    sys.exit(2)


def _preview_text(value, limit: int = 200) -> str:
    """Truncated preview tolerant of non-string values. A gallery row whose
    ``prompt`` is a non-string would crash ``(value or "")[:200]`` with a
    TypeError; coerce non-strings to ""."""
    text = value if isinstance(value, str) else ""
    return text[:limit]


def _text_field(value) -> str:
    return value if isinstance(value, str) else ""


def _serialize_image(i: "GalleryImage") -> dict:
    return {
        "id": i.id,
        "filename": _text_field(i.filename),
        "prompt": _preview_text(i.prompt),
        "model": _text_field(i.model),
        "size": _text_field(i.size),
        "tags": _text_field(i.tags),
        "favorite": bool(i.favorite),
        "album_id": _text_field(i.album_id),
        "session_id": _text_field(i.session_id),
        "width": i.width,
        "height": i.height,
        "file_size": i.file_size,
        "taken_at": i.taken_at.isoformat() if i.taken_at else "",
        "camera_make": _text_field(i.camera_make),
        "camera_model": _text_field(i.camera_model),
        "created_at": i.created_at.isoformat() if i.created_at else "",
    }


def _album_image_count(album) -> int:
    images = getattr(album, "images", None)
    try:
        return len(images) if images is not None else 0
    except TypeError:
        return 0


def cmd_list(args):
    db = SessionLocal()
    try:
        q = db.query(GalleryImage).filter(GalleryImage.is_active == True)  # noqa: E712
        if args.favorites:
            q = q.filter(GalleryImage.favorite == True)  # noqa: E712
        if args.album:
            al = db.query(GalleryAlbum).filter(GalleryAlbum.name == args.album).first()
            if not al:
                fail(f"no album named {args.album!r}")
            q = q.filter(GalleryImage.album_id == al.id)
        if args.tag:
            q = q.filter(GalleryImage.tags.ilike(f"%{args.tag}%"))
        q = q.order_by(GalleryImage.created_at.desc()).limit(args.limit)
        emit([_serialize_image(i) for i in q.all()], args)
    finally:
        db.close()


def cmd_show(args):
    db = SessionLocal()
    try:
        i = db.get(GalleryImage, args.id)
        if not i:
            fail(f"no image with id {args.id!r}")
        out = _serialize_image(i)
        out["prompt_full"] = _text_field(i.prompt)
        out["ai_tags"] = _text_field(i.ai_tags)
        out["gps_lat"] = i.gps_lat or ""
        out["gps_lng"] = i.gps_lng or ""
        out["file_hash"] = _text_field(i.file_hash)
        emit(out, args)
    finally:
        db.close()


def cmd_albums(args):
    db = SessionLocal()
    try:
        rows = db.query(GalleryAlbum).order_by(GalleryAlbum.name.asc()).all()
        emit([
            {"id": a.id, "name": a.name, "image_count": _album_image_count(a)}
            for a in rows
        ], args)
    finally:
        db.close()


def cmd_search(args):
    db = SessionLocal()
    try:
        like = f"%{args.query}%"
        rows = db.query(GalleryImage).filter(
            GalleryImage.is_active == True,  # noqa: E712
        ).filter(
            (GalleryImage.prompt.ilike(like)) |
            (GalleryImage.tags.ilike(like)) |
            (GalleryImage.ai_tags.ilike(like))
        ).order_by(GalleryImage.created_at.desc()).limit(args.limit).all()
        emit([_serialize_image(i) for i in rows], args)
    finally:
        db.close()


def cmd_delete(args):
    db = SessionLocal()
    try:
        i = db.get(GalleryImage, args.id)
        if not i:
            fail(f"no image with id {args.id!r}")
        i.is_active = False
        db.commit()
        emit({"ok": True, "id": i.id, "soft_deleted": True}, args)
    finally:
        db.close()


def _build_parser():
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--pretty", action="store_true")

    p = argparse.ArgumentParser(prog="odysseus-gallery", parents=[common])
    sub = p.add_subparsers(dest="cmd", required=True)

    pl = sub.add_parser("list", parents=[common])
    pl.add_argument("--limit", type=int, default=50)
    pl.add_argument("--favorites", action="store_true")
    pl.add_argument("--album", help="filter by album name")
    pl.add_argument("--tag", help="substring match against tags column")
    pl.set_defaults(func=cmd_list)

    psh = sub.add_parser("show", parents=[common])
    psh.add_argument("id")
    psh.set_defaults(func=cmd_show)

    pa = sub.add_parser("albums", parents=[common])
    pa.set_defaults(func=cmd_albums)

    ps = sub.add_parser("search", parents=[common])
    ps.add_argument("query")
    ps.add_argument("--limit", type=int, default=50)
    ps.set_defaults(func=cmd_search)

    pd = sub.add_parser("delete", parents=[common])
    pd.add_argument("id")
    pd.set_defaults(func=cmd_delete)

    return p


if __name__ == "__main__":
    sys.exit(run(_build_parser()))