""" Book Catalog - book-centered gallery with grouped photos and inline title editing. """ import base64 import hashlib import hmac import json import os import re import sqlite3 import time from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Optional from urllib.parse import quote import httpx import jwt from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Query, Request from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from pydantic import BaseModel try: import importlib psycopg = importlib.import_module("psycopg") except ImportError: # pragma: no cover - local sqlite mode can run without psycopg psycopg = None load_dotenv() # --- Config ----------------------------------------------------------------- DB_PATH = Path(__file__).parent.parent / "books.db" IMAGES_BASE = Path(os.getenv("IMAGES_PATH", r"C:\Users\ignac\OneDrive\Pictures\selected_libros")) OPTIMIZED_IMAGES_BASE = Path(os.getenv("OPTIMIZED_IMAGES_PATH", str(Path(__file__).parent.parent / "optimized_images"))) DATABASE_URL = os.getenv("DATABASE_URL", "").strip() DATABASE_ADAPTER = os.getenv("DATABASE_ADAPTER", "").strip().lower() FOLDER_PRIORITY = {"front": 0, "root": 1, "side": 2} AUTH_MODE = os.getenv("AUTH_MODE", "").strip().lower() APP_PASSWORD = os.getenv("APP_PASSWORD", "").strip() APP_SECRET_KEY = os.getenv("APP_SECRET_KEY", "change-me-in-production").strip() or "change-me-in-production" SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "bibliothek_session").strip() or "bibliothek_session" SESSION_TTL_SECONDS = int(os.getenv("SESSION_TTL_SECONDS", "2592000")) COOKIE_SECURE = os.getenv("COOKIE_SECURE", "false").strip().lower() in {"1", "true", "yes", "on"} SUPABASE_URL = os.getenv("SUPABASE_URL", "").strip().rstrip("/") SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "").strip() SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "").strip() SUPABASE_AUTH_REDIRECT_URL = os.getenv("SUPABASE_AUTH_REDIRECT_URL", "").strip() OWNER_EMAILS = { email.strip().lower() for email in os.getenv("OWNER_EMAILS", "ignaciolagosruiz@gmail.com").split(",") if email.strip() } IMAGE_BUCKET = os.getenv("IMAGE_BUCKET", "catalog-images").strip() or "catalog-images" IMAGE_ACCESS_MODE = os.getenv("IMAGE_ACCESS_MODE", "local").strip().lower() or "local" REMOTE_IMAGE_BASE_URL = os.getenv("REMOTE_IMAGE_BASE_URL", "").strip().rstrip("/") SIGNED_URL_TTL_SECONDS = int(os.getenv("SIGNED_URL_TTL_SECONDS", "604800")) if DATABASE_ADAPTER in {"sqlite", "postgres", "rest"}: DB_MODE = DATABASE_ADAPTER elif os.getenv("VERCEL") == "1" and SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY: DB_MODE = "rest" elif DATABASE_URL: DB_MODE = "postgres" else: DB_MODE = "sqlite" SEARCH_OPERATOR = "ILIKE" if DB_MODE == "postgres" else "LIKE" if not AUTH_MODE: if SUPABASE_URL and SUPABASE_ANON_KEY: AUTH_MODE = "supabase" elif APP_PASSWORD: AUTH_MODE = "password" else: AUTH_MODE = "none" REVIEW_FIELDS = [ "title", "subtitle", "author", "translator", "publisher", "year", "edition", "language", "series", "volume", "condition", "special_features", "other_text", "raw_ocr_text", ] REVIEW_STATES = {"pending", "verified", "flagged", "corrected"} app = FastAPI(title="Book Catalog Gallery") SIGNED_URL_CACHE: dict[str, tuple[float, str]] = {} JWKS_CLIENT = jwt.PyJWKClient(f"{SUPABASE_URL}/auth/v1/.well-known/jwks.json") if SUPABASE_URL else None # --- Database ---------------------------------------------------------------- class HybridRow(dict): def __init__(self, columns, values): super().__init__(zip(columns, values)) self._values = tuple(values) def __getitem__(self, key): if isinstance(key, int): return self._values[key] return super().__getitem__(key) def pg_row_factory(cursor): columns = [column.name for column in cursor.description] def make_row(values): return HybridRow(columns, values) return make_row def normalize_database_url(url: str) -> str: if url.startswith("postgres://"): return url.replace("postgres://", "postgresql://", 1) return url class CompatConnection: def __init__(self): self.backend = DB_MODE if self.backend == "postgres": if psycopg is None: raise RuntimeError("psycopg is required when DATABASE_URL is configured") self.conn = psycopg.connect(normalize_database_url(DATABASE_URL), row_factory=pg_row_factory) else: self.conn = sqlite3.connect(str(DB_PATH)) self.conn.row_factory = sqlite3.Row def _convert_query(self, query: str) -> str: if self.backend == "postgres": return query.replace("?", "%s") return query def execute(self, query: str, params=()): cursor = self.conn.cursor() cursor.execute(self._convert_query(query), params or ()) return cursor def executemany(self, query: str, seq_of_params): cursor = self.conn.cursor() cursor.executemany(self._convert_query(query), seq_of_params) return cursor def commit(self): self.conn.commit() def close(self): self.conn.close() def get_db(): return CompatConnection() def rest_headers(prefer_representation: bool = False) -> dict[str, str]: if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY: raise RuntimeError("Supabase REST mode requires SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY") headers = { "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}", "apikey": SUPABASE_SERVICE_ROLE_KEY, "Content-Type": "application/json", } if prefer_representation: headers["Prefer"] = "return=representation" return headers def rest_fetch_rows(select: str = "*", extra_params: Optional[dict[str, str]] = None) -> list[dict]: params = { "select": select, "order": "id.asc", "limit": "1000", } if extra_params: params.update(extra_params) response = httpx.get( f"{SUPABASE_URL}/rest/v1/book_images", headers=rest_headers(), params=params, timeout=60.0, ) response.raise_for_status() return response.json() def rest_fetch_row(image_id: int, select: str = "*") -> Optional[dict]: response = httpx.get( f"{SUPABASE_URL}/rest/v1/book_images", headers=rest_headers(), params={ "select": select, "id": f"eq.{image_id}", "limit": "1", }, timeout=30.0, ) response.raise_for_status() payload = response.json() return payload[0] if payload else None def rest_update_row(image_id: int, values: dict) -> Optional[dict]: response = httpx.patch( f"{SUPABASE_URL}/rest/v1/book_images", headers=rest_headers(prefer_representation=True), params={"id": f"eq.{image_id}"}, json=values, timeout=30.0, ) response.raise_for_status() payload = response.json() return payload[0] if payload else None def rest_update_rows(image_ids: list[int], values: dict) -> list[dict]: if not image_ids: return [] ids = ",".join(str(image_id) for image_id in image_ids) response = httpx.patch( f"{SUPABASE_URL}/rest/v1/book_images", headers=rest_headers(prefer_representation=True), params={"id": f"in.({ids})"}, json=values, timeout=60.0, ) response.raise_for_status() return response.json() def ensure_book_image_columns(): if DB_MODE != "sqlite": return conn = sqlite3.connect(str(DB_PATH)) cols = {row[1] for row in conn.execute("PRAGMA table_info(book_images)").fetchall()} if "description" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN description TEXT DEFAULT ""') if "review_status" not in cols: conn.execute("ALTER TABLE book_images ADD COLUMN review_status TEXT DEFAULT 'pending'") if "review_updated_at" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN review_updated_at TEXT DEFAULT ""') if "image_url" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN image_url TEXT DEFAULT ""') if "thumb_url" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN thumb_url TEXT DEFAULT ""') if "storage_path" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN storage_path TEXT DEFAULT ""') if "thumb_storage_path" not in cols: conn.execute('ALTER TABLE book_images ADD COLUMN thumb_storage_path TEXT DEFAULT ""') conn.execute( "UPDATE book_images SET review_status = 'pending' WHERE review_status IS NULL OR TRIM(review_status) = ''" ) conn.commit() conn.close() ensure_book_image_columns() # --- Helpers ----------------------------------------------------------------- def clean_text(value) -> str: if value is None: return "" if isinstance(value, list): value = ", ".join(str(item).strip() for item in value if str(item).strip()) return str(value).strip() def session_signature(payload: str) -> str: return hmac.new(APP_SECRET_KEY.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() def create_session_token() -> str: expires_at = int(time.time()) + SESSION_TTL_SECONDS payload = f"shared:{expires_at}" return f"{payload}:{session_signature(payload)}" def is_valid_session_token(token: str) -> bool: try: payload, signature = token.rsplit(":", 1) _, expires_at_text = payload.split(":", 1) expires_at = int(expires_at_text) except ValueError: return False expected = session_signature(payload) if not hmac.compare_digest(signature, expected): return False return expires_at >= int(time.time()) def request_is_authenticated(request: Request) -> bool: if AUTH_MODE != "password" or not APP_PASSWORD: return True token = clean_text(request.cookies.get(SESSION_COOKIE_NAME)) return bool(token) and is_valid_session_token(token) def extract_bearer_token(request: Request) -> str: auth_header = clean_text(request.headers.get("Authorization")) if auth_header.lower().startswith("bearer "): return auth_header[7:].strip() return "" def supabase_issuer() -> str: return f"{SUPABASE_URL}/auth/v1" if SUPABASE_URL else "" def verify_supabase_token_locally(token: str) -> dict: if not token or not SUPABASE_URL or JWKS_CLIENT is None: raise HTTPException(status_code=401, detail="Authentication required") header = jwt.get_unverified_header(token) alg = clean_text(header.get("alg")).upper() if alg.startswith("HS"): raise RuntimeError("Legacy symmetric JWTs require fallback validation") signing_key = JWKS_CLIENT.get_signing_key_from_jwt(token) claims = jwt.decode( token, signing_key.key, algorithms=[alg], issuer=supabase_issuer(), options={"verify_aud": False, "require": ["exp", "iss", "sub"]}, ) return claims def verify_supabase_token_via_userinfo(token: str) -> dict: if not SUPABASE_URL or not SUPABASE_ANON_KEY: raise HTTPException(status_code=401, detail="Supabase auth is not configured") response = httpx.get( f"{SUPABASE_URL}/auth/v1/user", headers={ "Authorization": f"Bearer {token}", "apikey": SUPABASE_ANON_KEY, }, timeout=15.0, ) if response.status_code != 200: raise HTTPException(status_code=401, detail="Invalid session") payload = response.json() return { "sub": payload.get("id"), "email": clean_text(payload.get("email")), "role": clean_text(payload.get("role") or payload.get("app_metadata", {}).get("role")), "claims": payload, } def verify_supabase_access_token(token: str) -> dict: if not token: raise HTTPException(status_code=401, detail="Authentication required") try: claims = verify_supabase_token_locally(token) return { "sub": claims.get("sub"), "email": clean_text(claims.get("email")), "role": clean_text(claims.get("role")), "claims": claims, } except HTTPException: raise except Exception: # noqa: BLE001 return verify_supabase_token_via_userinfo(token) def current_supabase_user(request: Request) -> dict: user = getattr(request.state, "user", None) if not user: raise HTTPException(status_code=401, detail="Authentication required") return user def is_owner_email(email: str) -> bool: return clean_text(email).lower() in OWNER_EMAILS def app_config() -> dict: return { "authMode": AUTH_MODE, "supabaseUrl": SUPABASE_URL, "supabaseAnonKey": SUPABASE_ANON_KEY, "authRedirectUrl": SUPABASE_AUTH_REDIRECT_URL, "inviteEnabled": bool(SUPABASE_SERVICE_ROLE_KEY), } def quote_storage_path(path: str) -> str: return quote(path.lstrip("/"), safe="/") def get_public_image_url(path: str) -> str: if REMOTE_IMAGE_BASE_URL: return f"{REMOTE_IMAGE_BASE_URL}/{quote_storage_path(path)}" if not SUPABASE_URL or not IMAGE_BUCKET: return "" return f"{SUPABASE_URL}/storage/v1/object/public/{IMAGE_BUCKET}/{quote_storage_path(path)}" def get_signed_image_url(path: str) -> str: if not SUPABASE_URL or not IMAGE_BUCKET or not SUPABASE_SERVICE_ROLE_KEY: return "" cache_key = f"{IMAGE_BUCKET}:{path}" cached = SIGNED_URL_CACHE.get(cache_key) if cached and cached[0] > time.time() + 300: return cached[1] response = httpx.post( f"{SUPABASE_URL}/storage/v1/object/sign/{IMAGE_BUCKET}/{quote_storage_path(path)}", headers={ "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}", "apikey": SUPABASE_SERVICE_ROLE_KEY, "Content-Type": "application/json", }, json={"expiresIn": SIGNED_URL_TTL_SECONDS}, timeout=20.0, ) response.raise_for_status() payload = response.json() signed = clean_text(payload.get("signedURL") or payload.get("signedUrl") or payload.get("signed_url")) if not signed: return "" signed_url = signed if signed.startswith("http") else f"{SUPABASE_URL}/storage/v1{signed}" SIGNED_URL_CACHE[cache_key] = (time.time() + SIGNED_URL_TTL_SECONDS, signed_url) return signed_url def resolve_hosted_url(path: str) -> str: cleaned = clean_text(path) if not cleaned: return "" if IMAGE_ACCESS_MODE == "local_optimized": return f"/media/{quote_storage_path(cleaned)}" if IMAGE_ACCESS_MODE == "signed": return get_signed_image_url(cleaned) if IMAGE_ACCESS_MODE == "public" or REMOTE_IMAGE_BASE_URL: return get_public_image_url(cleaned) return "" def known_text(value) -> str: text = clean_text(value) if not text or text.lower() == "unknown": return "" return text def normalize_review_status(value) -> str: text = clean_text(value).lower() return text if text in REVIEW_STATES else "pending" def review_timestamp() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def normalize_title(value) -> str: title = known_text(value) if not title: return "" title = re.sub(r"[^\w\s]", " ", title.lower(), flags=re.UNICODE) title = re.sub(r"\s+", " ", title).strip() return title def encode_group_key(normalized_title: str, image_id: int) -> str: if normalized_title: token = base64.urlsafe_b64encode(normalized_title.encode("utf-8")).decode("ascii") return f"title-{token.rstrip('=')}" return f"image-{image_id}" def decode_group_key(group_key: str) -> dict: if group_key.startswith("image-"): try: return {"type": "image", "image_id": int(group_key.split("-", 1)[1])} except ValueError as exc: raise HTTPException(status_code=400, detail="Invalid image group key") from exc if group_key.startswith("title-"): token = group_key.split("-", 1)[1] padding = "=" * (-len(token) % 4) try: normalized_title = base64.urlsafe_b64decode((token + padding).encode("ascii")).decode("utf-8") except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=400, detail="Invalid title group key") from exc return {"type": "title", "normalized_title": normalized_title} raise HTTPException(status_code=400, detail="Unknown group key format") def photo_label(folder: str) -> str: return { "front": "Front cover", "root": "Title page / inside page", "side": "Spine", }.get(folder, "Book photo") def extract_message_text(content) -> str: if isinstance(content, str): return content.strip() if content is None: return "" if isinstance(content, list): parts = [] for chunk in content: if isinstance(chunk, str): text = chunk.strip() else: text = clean_text(getattr(chunk, "text", "")) if not text and isinstance(chunk, dict): text = clean_text(chunk.get("text") or chunk.get("content")) if text: parts.append(text) return "\n".join(parts).strip() return clean_text(content) def attach_image_urls(item: dict) -> dict: image_url = clean_text(item.get("image_url")) thumb_url = clean_text(item.get("thumb_url")) storage_path = clean_text(item.get("storage_path")) thumb_storage_path = clean_text(item.get("thumb_storage_path")) or storage_path display_url = image_url or resolve_hosted_url(storage_path) preview_url = thumb_url or resolve_hosted_url(thumb_storage_path) or display_url if not display_url: display_url = f"/photos/{item.get('source_folder', '')}/{item.get('source_filename', '')}" if not preview_url: preview_url = display_url item["display_url"] = display_url item["thumb_url"] = preview_url return item def serialize_review_image(row: dict) -> dict: item = dict(row) folder = clean_text(item.get("source_folder")) item["photo_label"] = photo_label(folder) item["review_status"] = normalize_review_status(item.get("review_status")) item["review_updated_at"] = clean_text(item.get("review_updated_at")) item["group_key"] = encode_group_key(normalize_title(item.get("title")), item["id"]) attach_image_urls(item) return item def review_queue_item(row: dict) -> dict: item = serialize_review_image(row) return { "id": item["id"], "title": known_text(item.get("title")) or f"Image #{item['id']}", "author": known_text(item.get("author")), "language": known_text(item.get("language")), "source_folder": item.get("source_folder", ""), "source_filename": item.get("source_filename", ""), "display_url": item.get("display_url", ""), "thumb_url": item.get("thumb_url", ""), "photo_label": item["photo_label"], "review_status": item["review_status"], "review_updated_at": item["review_updated_at"], } def photo_sort_key(row: dict) -> tuple: return (FOLDER_PRIORITY.get(clean_text(row.get("source_folder")), 9), row["id"]) def pick_cover_image(rows: list[dict]) -> dict: return sorted(rows, key=photo_sort_key)[0] def most_common_value(rows: list[dict], field: str) -> str: values = [known_text(row.get(field)) for row in rows] values = [value for value in values if value] if not values: return "" return Counter(values).most_common(1)[0][0] def year_sort_value(value) -> int: text = clean_text(value) match = re.search(r"(1[6-9]\d{2}|20\d{2}|2100)", text) if match: return int(match.group(1)) return 9999 def image_brief(row: dict, with_urls: bool = True) -> dict: item = { "id": row["id"], "source_folder": row["source_folder"], "source_filename": row["source_filename"], "image_url": row.get("image_url", ""), "thumb_url": row.get("thumb_url", ""), "storage_path": row.get("storage_path", ""), "thumb_storage_path": row.get("thumb_storage_path", ""), "photo_label": photo_label(row["source_folder"]), } return attach_image_urls(item) if with_urls else item def summarize_group(rows: list[dict], with_urls: bool = True, sample_limit: int = 4) -> dict: ordered_rows = sorted(rows, key=photo_sort_key) cover = pick_cover_image(ordered_rows) normalized_title = normalize_title(ordered_rows[0].get("title")) editable_title = most_common_value(ordered_rows, "title") display_title = editable_title or f"Untitled #{cover['id']}" described_count = sum(1 for row in ordered_rows if clean_text(row.get("description"))) review_counts = Counter(normalize_review_status(row.get("review_status")) for row in ordered_rows) folders = sorted( {clean_text(row.get("source_folder")) for row in ordered_rows if clean_text(row.get("source_folder"))}, key=lambda folder: FOLDER_PRIORITY.get(folder, 9), ) return { "group_key": encode_group_key(normalized_title, cover["id"]), "title": display_title, "editable_title": editable_title, "author": most_common_value(ordered_rows, "author"), "publisher": most_common_value(ordered_rows, "publisher"), "year": most_common_value(ordered_rows, "year"), "language": most_common_value(ordered_rows, "language"), "image_count": len(ordered_rows), "described_count": described_count, "review_counts": { "pending": review_counts["pending"], "verified": review_counts["verified"], "flagged": review_counts["flagged"], "corrected": review_counts["corrected"], }, "needs_review_count": review_counts["pending"] + review_counts["flagged"], "done_review_count": review_counts["verified"] + review_counts["corrected"], "folders": folders, "cover_image": image_brief(cover, with_urls=with_urls), "sample_images": [image_brief(row, with_urls=with_urls) for row in ordered_rows[:sample_limit]], "min_image_id": min(row["id"] for row in ordered_rows), } def group_rows(rows: list[dict]) -> dict[str, list[dict]]: grouped = defaultdict(list) for row in rows: item = dict(row) key = encode_group_key(normalize_title(item.get("title")), item["id"]) grouped[key].append(item) return grouped def build_groups(rows: list[dict], with_urls: bool = True, sample_limit: int = 4) -> list[dict]: return [summarize_group(group, with_urls=with_urls, sample_limit=sample_limit) for group in group_rows(rows).values()] def sort_groups(groups: list[dict], sort: str, order: str) -> list[dict]: reverse = order == "desc" def text_key(value) -> str: return known_text(value).lower() or "~~~~" if sort == "title": key_func = lambda group: (text_key(group["title"]), group["min_image_id"]) elif sort == "author": key_func = lambda group: (text_key(group["author"]), text_key(group["title"]), group["min_image_id"]) elif sort == "year": key_func = lambda group: (year_sort_value(group["year"]), text_key(group["title"]), group["min_image_id"]) elif sort == "image_count": key_func = lambda group: (group["image_count"], text_key(group["title"]), group["min_image_id"]) else: key_func = lambda group: group["min_image_id"] return sorted(groups, key=key_func, reverse=reverse) def build_filters( search: Optional[str], folder: Optional[str], language: Optional[str], has_title: Optional[bool], ) -> tuple[str, list]: conditions = [] params = [] if search: conditions.append( f"(title {SEARCH_OPERATOR} ? OR author {SEARCH_OPERATOR} ? OR publisher {SEARCH_OPERATOR} ? OR raw_ocr_text {SEARCH_OPERATOR} ?)" ) term = f"%{search}%" params.extend([term, term, term, term]) if folder: conditions.append("source_folder = ?") params.append(folder) if language: conditions.append(f"language {SEARCH_OPERATOR} ?") params.append(f"%{language}%") if has_title is True: conditions.append("title != 'Unknown' AND title != ''") elif has_title is False: conditions.append("(title = 'Unknown' OR title = '')") where = " WHERE " + " AND ".join(conditions) if conditions else "" return where, params def filter_rows_python( rows: list[dict], search: Optional[str], folder: Optional[str], language: Optional[str], has_title: Optional[bool], ) -> list[dict]: filtered = [] needle = clean_text(search).lower() for row in rows: if folder and clean_text(row.get("source_folder")) != folder: continue if language and language.lower() not in clean_text(row.get("language")).lower(): continue title_known = bool(known_text(row.get("title"))) if has_title is True and not title_known: continue if has_title is False and title_known: continue if needle: haystack = " ".join( [ clean_text(row.get("title")), clean_text(row.get("author")), clean_text(row.get("publisher")), clean_text(row.get("raw_ocr_text")), ] ).lower() if needle not in haystack: continue filtered.append(dict(row)) return filtered def postgrest_escape_like(value: str) -> str: text = clean_text(value) return text.replace("%", "").replace(",", " ").replace("(", " ").replace(")", " ") def rest_filter_params( search: Optional[str], folder: Optional[str], language: Optional[str], has_title: Optional[bool], ) -> dict[str, str]: params: dict[str, str] = {} clauses: list[str] = [] if search: needle = postgrest_escape_like(search) clauses.append( f"or(title.ilike.*{needle}*,author.ilike.*{needle}*,publisher.ilike.*{needle}*,raw_ocr_text.ilike.*{needle}*)" ) if folder: clauses.append(f"source_folder.eq.{folder}") if language: clauses.append(f"language.ilike.*{postgrest_escape_like(language)}*") if has_title is True: clauses.append("title.not.eq.Unknown") clauses.append("title.not.eq.") elif has_title is False: clauses.append("or(title.eq.Unknown,title.eq.)") if clauses: params["and"] = f"({','.join(clauses)})" return params def sort_images_python(rows: list[dict], sort: str, order: str) -> list[dict]: reverse = order == "desc" def text_key(value) -> str: return clean_text(value).lower() if sort == "title": key_func = lambda row: (text_key(row.get("title")), row.get("id", 0)) elif sort == "author": key_func = lambda row: (text_key(row.get("author")), text_key(row.get("title")), row.get("id", 0)) elif sort == "year": key_func = lambda row: (year_sort_value(row.get("year")), text_key(row.get("title")), row.get("id", 0)) elif sort == "source_folder": key_func = lambda row: (FOLDER_PRIORITY.get(clean_text(row.get("source_folder")), 9), row.get("id", 0)) elif sort == "source_filename": key_func = lambda row: (text_key(row.get("source_filename")), row.get("id", 0)) else: key_func = lambda row: row.get("id", 0) return sorted(rows, key=key_func, reverse=reverse) def rows_for_group_key(rows: list[dict], group_key: str) -> list[dict]: decoded = decode_group_key(group_key) if decoded["type"] == "image": return [row for row in rows if row["id"] == decoded["image_id"]] normalized_title = decoded["normalized_title"] return [row for row in rows if normalize_title(row.get("title")) == normalized_title] def build_group_detail(rows: list[dict]) -> dict: ordered_rows = sorted(rows, key=photo_sort_key) summary = summarize_group(ordered_rows) primary_id = summary["cover_image"]["id"] images = [] for row in ordered_rows: item = dict(row) images.append(serialize_review_image(item)) summary["images"] = images summary["primary_image_id"] = primary_id summary["primary_image_index"] = next( (index for index, image in enumerate(images) if image["id"] == primary_id), 0, ) return summary PUBLIC_PATHS = {"/login", "/logout", "/healthz"} PUBLIC_PREFIXES = ("/_vercel", "/favicon", "/.well-known") @app.middleware("http") async def password_gate(request: Request, call_next): path = request.url.path if path == "/" and AUTH_MODE in {"supabase", "none"}: return await call_next(request) if path in PUBLIC_PATHS or any(path.startswith(prefix) for prefix in PUBLIC_PREFIXES): return await call_next(request) if AUTH_MODE == "password": if request_is_authenticated(request): return await call_next(request) if path.startswith("/api/") or path.startswith("/photos/") or path.startswith("/media/"): return JSONResponse({"detail": "Authentication required"}, status_code=401) return HTMLResponse(LOGIN_HTML, status_code=401) if AUTH_MODE == "supabase": if path.startswith("/api/") or path.startswith("/photos/") or path.startswith("/media/"): token = extract_bearer_token(request) try: request.state.user = verify_supabase_access_token(token) except HTTPException as exc: return JSONResponse({"detail": exc.detail}, status_code=exc.status_code) return await call_next(request) return await call_next(request) class LoginRequest(BaseModel): password: str @app.post("/login") def login(req: LoginRequest): if AUTH_MODE != "password": raise HTTPException(status_code=404, detail="Password login is disabled") if APP_PASSWORD and req.password != APP_PASSWORD: raise HTTPException(status_code=401, detail="Incorrect password") response = JSONResponse({"ok": True}) response.set_cookie( SESSION_COOKIE_NAME, create_session_token(), httponly=True, max_age=SESSION_TTL_SECONDS, secure=COOKIE_SECURE, samesite="lax", path="/", ) return response @app.post("/logout") def logout(): response = JSONResponse({"ok": True}) response.delete_cookie(SESSION_COOKIE_NAME, path="/") return response @app.get("/api/auth/me") def auth_me(request: Request): if AUTH_MODE == "supabase": user = current_supabase_user(request) return { "authenticated": True, "email": user.get("email", ""), "user_id": user.get("sub", ""), "is_owner": is_owner_email(user.get("email", "")), "auth_mode": AUTH_MODE, } return {"authenticated": True, "auth_mode": AUTH_MODE, "is_owner": True} class InviteMemberRequest(BaseModel): email: str @app.post("/api/invite-member") def invite_member(req: InviteMemberRequest, request: Request): if AUTH_MODE != "supabase": raise HTTPException(status_code=400, detail="Supabase Auth is not enabled") if not SUPABASE_SERVICE_ROLE_KEY or not SUPABASE_URL: raise HTTPException(status_code=500, detail="Supabase service role key is missing") user = current_supabase_user(request) requester_email = clean_text(user.get("email")).lower() if not is_owner_email(requester_email): raise HTTPException(status_code=403, detail="Only owners can invite members") invite_email = clean_text(req.email).lower() if not invite_email or "@" not in invite_email: raise HTTPException(status_code=400, detail="Please enter a valid email") params = {} if SUPABASE_AUTH_REDIRECT_URL: params["redirect_to"] = SUPABASE_AUTH_REDIRECT_URL response = httpx.post( f"{SUPABASE_URL}/auth/v1/invite", params=params, headers={ "Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}", "apikey": SUPABASE_SERVICE_ROLE_KEY, "Content-Type": "application/json", }, json={ "email": invite_email, "data": { "role": "member", "invited_by": requester_email, }, }, timeout=20.0, ) if response.status_code == 422: raise HTTPException(status_code=409, detail="That email is already invited or already has access") if response.status_code >= 400: detail = clean_text(response.text) or "Failed to send invite" raise HTTPException(status_code=400, detail=detail[:300]) return {"ok": True, "email": invite_email} @app.get("/healthz") def healthz(): return {"ok": True, "db_mode": DB_MODE, "auth_mode": AUTH_MODE} # --- API Routes -------------------------------------------------------------- @app.get("/api/images") def list_images( search: Optional[str] = Query(None), folder: Optional[str] = Query(None), language: Optional[str] = Query(None), has_title: Optional[bool] = Query(None), sort: str = Query("id"), order: str = Query("asc"), page: int = Query(1, ge=1), per_page: int = Query(30, ge=1, le=100), ): if DB_MODE == "rest": extra_params = rest_filter_params(search, folder, language, has_title) rows = rest_fetch_rows( "id,title,subtitle,author,publisher,year,language,source_folder,source_filename,description,review_status,review_updated_at,image_url,thumb_url,storage_path,thumb_storage_path,raw_ocr_text", extra_params=extra_params, ) rows = sort_images_python(rows, sort, order) total = len(rows) offset = (page - 1) * per_page images = [serialize_review_image(row) for row in rows[offset: offset + per_page]] return { "images": images, "total": total, "page": page, "per_page": per_page, "pages": max(1, (total + per_page - 1) // per_page), } conn = get_db() where, params = build_filters(search, folder, language, has_title) total = conn.execute(f"SELECT COUNT(*) FROM book_images{where}", params).fetchone()[0] allowed_sorts = {"id", "title", "author", "year", "source_folder", "source_filename"} sort_col = sort if sort in allowed_sorts else "id" order_dir = "DESC" if order == "desc" else "ASC" offset = (page - 1) * per_page query = f""" SELECT id, title, subtitle, author, publisher, year, language, source_folder, source_filename, description, review_status, review_updated_at, image_url, thumb_url, storage_path, thumb_storage_path FROM book_images{where} ORDER BY {sort_col} {order_dir} LIMIT ? OFFSET ? """ rows = conn.execute(query, params + [per_page, offset]).fetchall() conn.close() images = [] for row in rows: images.append(serialize_review_image(dict(row))) return { "images": images, "total": total, "page": page, "per_page": per_page, "pages": max(1, (total + per_page - 1) // per_page), } @app.get("/api/images/{image_id}") def get_image_detail(image_id: int): if DB_MODE == "rest": row = rest_fetch_row(image_id) if not row: raise HTTPException(status_code=404, detail="Image not found") image = serialize_review_image(row) group_key = image["group_key"] all_rows = rest_fetch_rows("id,title,author,source_folder,source_filename,image_url,thumb_url,storage_path,thumb_storage_path") related = [] for related_row in sorted(rows_for_group_key(all_rows, group_key), key=photo_sort_key): if related_row["id"] == image_id: continue related.append( { "id": related_row["id"], "title": related_row["title"], "author": related_row["author"], "folder": related_row["source_folder"], "filename": related_row["source_filename"], "reason": "same book", } ) image["related"] = related[:20] return image conn = get_db() row = conn.execute("SELECT * FROM book_images WHERE id = ?", (image_id,)).fetchone() if not row: conn.close() raise HTTPException(status_code=404, detail="Image not found") image = serialize_review_image(dict(row)) group_key = image["group_key"] all_rows = [ dict(r) for r in conn.execute( "SELECT id, title, author, source_folder, source_filename, image_url, thumb_url, storage_path, thumb_storage_path FROM book_images" ).fetchall() ] conn.close() related = [] for related_row in sorted(rows_for_group_key(all_rows, group_key), key=photo_sort_key): if related_row["id"] == image_id: continue related.append( { "id": related_row["id"], "title": related_row["title"], "author": related_row["author"], "folder": related_row["source_folder"], "filename": related_row["source_filename"], "reason": "same book", } ) image["related"] = related[:20] return image @app.get("/api/books") def list_books( search: Optional[str] = Query(None), folder: Optional[str] = Query(None), language: Optional[str] = Query(None), has_title: Optional[bool] = Query(None), sort: str = Query("title"), order: str = Query("asc"), page: int = Query(1, ge=1), per_page: int = Query(24, ge=1, le=100), ): if DB_MODE == "rest": extra_params = rest_filter_params(search, folder, language, has_title) rows = rest_fetch_rows( "id,title,author,publisher,year,language,source_folder,source_filename,description,review_status,review_updated_at,image_url,thumb_url,storage_path,thumb_storage_path", extra_params=extra_params, ) grouped_map = group_rows(rows) groups = sort_groups([summarize_group(group, with_urls=False, sample_limit=0) for group in grouped_map.values()], sort, order) total = len(groups) offset = (page - 1) * per_page paged = [summarize_group(grouped_map[group["group_key"]], with_urls=True, sample_limit=0) for group in groups[offset: offset + per_page]] return { "books": paged, "total": total, "page": page, "per_page": per_page, "pages": max(1, (total + per_page - 1) // per_page), } conn = get_db() where, params = build_filters(search, folder, language, has_title) rows = [ dict(r) for r in conn.execute( f""" SELECT id, title, author, publisher, year, language, source_folder, source_filename, description, review_status, review_updated_at, image_url, thumb_url, storage_path, thumb_storage_path FROM book_images{where} """, params, ).fetchall() ] conn.close() groups = sort_groups(build_groups(rows, sample_limit=0), sort, order) total = len(groups) offset = (page - 1) * per_page return { "books": groups[offset: offset + per_page], "total": total, "page": page, "per_page": per_page, "pages": max(1, (total + per_page - 1) // per_page), } @app.get("/api/books/{group_key}") def get_book_group(group_key: str): if DB_MODE == "rest": rows = rest_fetch_rows() grouped_rows = rows_for_group_key(rows, group_key) if not grouped_rows: raise HTTPException(status_code=404, detail="Book group not found") return build_group_detail(grouped_rows) conn = get_db() rows = [dict(r) for r in conn.execute("SELECT * FROM book_images").fetchall()] conn.close() grouped_rows = rows_for_group_key(rows, group_key) if not grouped_rows: raise HTTPException(status_code=404, detail="Book group not found") return build_group_detail(grouped_rows) class UpdateGroupTitleRequest(BaseModel): title: str @app.patch("/api/book-groups/{group_key}/title") def update_book_group_title(group_key: str, req: UpdateGroupTitleRequest): new_title = clean_text(req.title) if not new_title or new_title.lower() == "unknown": raise HTTPException(status_code=400, detail="Please enter a real title") if DB_MODE == "rest": rows = rest_fetch_rows("id,title") grouped_rows = rows_for_group_key(rows, group_key) if not grouped_rows: raise HTTPException(status_code=404, detail="Book group not found") image_ids = [row["id"] for row in grouped_rows] rest_update_rows(image_ids, {"title": new_title}) focus_image_id = grouped_rows[0]["id"] return { "ok": True, "updated": len(grouped_rows), "title": new_title, "group_key": encode_group_key(normalize_title(new_title), focus_image_id), } conn = get_db() rows = [dict(r) for r in conn.execute("SELECT id, title FROM book_images").fetchall()] grouped_rows = rows_for_group_key(rows, group_key) if not grouped_rows: conn.close() raise HTTPException(status_code=404, detail="Book group not found") updates = [(new_title, row["id"]) for row in grouped_rows] conn.executemany("UPDATE book_images SET title = ? WHERE id = ?", updates) conn.commit() conn.close() focus_image_id = grouped_rows[0]["id"] return { "ok": True, "updated": len(grouped_rows), "title": new_title, "group_key": encode_group_key(normalize_title(new_title), focus_image_id), } @app.get("/api/review/dashboard") def get_review_dashboard(): if DB_MODE == "rest": rows = rest_fetch_rows() counts = Counter(normalize_review_status(row.get("review_status")) for row in rows) next_row = next((row for row in rows if normalize_review_status(row.get("review_status")) == "pending"), None) flagged_rows = [row for row in rows if normalize_review_status(row.get("review_status")) == "flagged"] return { "counts": { "total": len(rows), "pending": counts["pending"], "verified": counts["verified"], "flagged": counts["flagged"], "corrected": counts["corrected"], "done": counts["verified"] + counts["corrected"], "remaining": counts["pending"] + counts["flagged"], }, "next_image": serialize_review_image(next_row) if next_row else None, "flagged": [review_queue_item(row) for row in flagged_rows], } conn = get_db() rows = [dict(r) for r in conn.execute("SELECT * FROM book_images ORDER BY id").fetchall()] conn.close() counts = Counter(normalize_review_status(row.get("review_status")) for row in rows) next_row = next( (row for row in rows if normalize_review_status(row.get("review_status")) == "pending"), None, ) flagged_rows = [row for row in rows if normalize_review_status(row.get("review_status")) == "flagged"] return { "counts": { "total": len(rows), "pending": counts["pending"], "verified": counts["verified"], "flagged": counts["flagged"], "corrected": counts["corrected"], "done": counts["verified"] + counts["corrected"], "remaining": counts["pending"] + counts["flagged"], }, "next_image": serialize_review_image(next_row) if next_row else None, "flagged": [review_queue_item(row) for row in flagged_rows], } def update_review_state(image_id: int, review_state: str) -> dict: if DB_MODE == "rest": row = rest_fetch_row(image_id) if not row: raise HTTPException(status_code=404, detail="Image not found") updated = rest_update_row(image_id, {"review_status": review_state, "review_updated_at": review_timestamp()}) return serialize_review_image(updated or row) conn = get_db() row = conn.execute("SELECT * FROM book_images WHERE id = ?", (image_id,)).fetchone() if not row: conn.close() raise HTTPException(status_code=404, detail="Image not found") updated_at = review_timestamp() conn.execute( "UPDATE book_images SET review_status = ?, review_updated_at = ? WHERE id = ?", (review_state, updated_at, image_id), ) conn.commit() updated = conn.execute("SELECT * FROM book_images WHERE id = ?", (image_id,)).fetchone() conn.close() return serialize_review_image(dict(updated)) @app.post("/api/review/{image_id}/verify") def verify_review_image(image_id: int): return {"ok": True, "image": update_review_state(image_id, "verified")} @app.post("/api/review/{image_id}/flag") def flag_review_image(image_id: int): return {"ok": True, "image": update_review_state(image_id, "flagged")} class ReviewCorrectionRequest(BaseModel): title: str = "" subtitle: str = "" author: str = "" translator: str = "" publisher: str = "" year: str = "" edition: str = "" language: str = "" series: str = "" volume: str = "" condition: str = "" special_features: str = "" other_text: str = "" raw_ocr_text: str = "" @app.patch("/api/review/{image_id}") def save_review_correction(image_id: int, req: ReviewCorrectionRequest): if DB_MODE == "rest": row = rest_fetch_row(image_id) if not row: raise HTTPException(status_code=404, detail="Image not found") payload = {field: clean_text(getattr(req, field)) for field in REVIEW_FIELDS} payload["review_status"] = "corrected" payload["review_updated_at"] = review_timestamp() updated = rest_update_row(image_id, payload) return {"ok": True, "image": serialize_review_image(updated or row)} conn = get_db() row = conn.execute("SELECT * FROM book_images WHERE id = ?", (image_id,)).fetchone() if not row: conn.close() raise HTTPException(status_code=404, detail="Image not found") payload = {field: clean_text(getattr(req, field)) for field in REVIEW_FIELDS} assignments = ", ".join(f"{field} = ?" for field in REVIEW_FIELDS) values: list[object] = [payload[field] for field in REVIEW_FIELDS] values.extend(["corrected", review_timestamp(), image_id]) conn.execute( f"UPDATE book_images SET {assignments}, review_status = ?, review_updated_at = ? WHERE id = ?", values, ) conn.commit() updated = conn.execute("SELECT * FROM book_images WHERE id = ?", (image_id,)).fetchone() conn.close() return {"ok": True, "image": serialize_review_image(dict(updated))} @app.get("/api/stats") def get_stats(): if DB_MODE == "rest": rows = rest_fetch_rows("id,title,author,language,source_folder,source_filename,description,review_status,image_url,thumb_url,storage_path,thumb_storage_path") total = len(rows) with_title = sum(1 for row in rows if known_text(row.get("title"))) with_author = sum(1 for row in rows if known_text(row.get("author"))) with_desc = sum(1 for row in rows if clean_text(row.get("description"))) review_counts = Counter(normalize_review_status(row.get("review_status")) for row in rows) folder_counts = Counter(clean_text(row.get("source_folder")) for row in rows if clean_text(row.get("source_folder"))) language_counts = Counter(known_text(row.get("language")) for row in rows if known_text(row.get("language"))) return { "total": total, "book_groups": len(group_rows(rows)), "with_title": with_title, "with_author": with_author, "with_description": with_desc, "review": { "pending": review_counts["pending"], "verified": review_counts["verified"], "flagged": review_counts["flagged"], "corrected": review_counts["corrected"], "done": review_counts["verified"] + review_counts["corrected"], }, "folders": [{"folder": folder, "count": count} for folder, count in sorted(folder_counts.items())], "languages": [ {"language": language, "count": count} for language, count in sorted(language_counts.items(), key=lambda item: (-item[1], item[0])) ], } conn = get_db() rows = [ dict(r) for r in conn.execute( "SELECT id, title, author, language, source_folder, source_filename, description, review_status FROM book_images" ).fetchall() ] total = len(rows) with_title = sum(1 for row in rows if known_text(row.get("title"))) with_author = sum(1 for row in rows if known_text(row.get("author"))) with_desc = sum(1 for row in rows if clean_text(row.get("description"))) review_counts = Counter(normalize_review_status(row.get("review_status")) for row in rows) folders = conn.execute( "SELECT source_folder, COUNT(*) FROM book_images GROUP BY source_folder ORDER BY source_folder" ).fetchall() languages = conn.execute( "SELECT language, COUNT(*) as cnt FROM book_images WHERE language != '' AND language != 'Unknown' GROUP BY language ORDER BY cnt DESC" ).fetchall() conn.close() return { "total": total, "book_groups": len(group_rows(rows)), "with_title": with_title, "with_author": with_author, "with_description": with_desc, "review": { "pending": review_counts["pending"], "verified": review_counts["verified"], "flagged": review_counts["flagged"], "corrected": review_counts["corrected"], "done": review_counts["verified"] + review_counts["corrected"], }, "folders": [{"folder": row[0], "count": row[1]} for row in folders], "languages": [{"language": row[0], "count": row[1]} for row in languages], } @app.get("/api/languages") def get_languages(): if DB_MODE == "rest": rows = rest_fetch_rows("language") return sorted({known_text(row.get("language")) for row in rows if known_text(row.get("language"))}) conn = get_db() rows = conn.execute( "SELECT DISTINCT language FROM book_images WHERE language != '' AND language != 'Unknown' ORDER BY language" ).fetchall() conn.close() return [row[0] for row in rows] # --- Serve book images ------------------------------------------------------- @app.get("/photos/{folder}/{filename}") def serve_photo(folder: str, filename: str): if folder == "root": path = IMAGES_BASE / filename else: path = IMAGES_BASE / folder / filename if not path.exists(): raise HTTPException(status_code=404, detail="Image file not found") return FileResponse(str(path), media_type="image/jpeg") @app.get("/media/{asset_path:path}") def serve_optimized_media(asset_path: str): base = OPTIMIZED_IMAGES_BASE.resolve() path = (base / asset_path).resolve() if path != base and base not in path.parents: raise HTTPException(status_code=400, detail="Invalid asset path") if not path.exists() or not path.is_file(): raise HTTPException(status_code=404, detail="Optimized asset not found") return FileResponse(str(path), media_type="image/jpeg") # --- AI description generation ---------------------------------------------- class GenerateRequest(BaseModel): image_id: int @app.post("/api/generate-description") def generate_description(req: GenerateRequest): db_conn = None if DB_MODE == "rest": image = rest_fetch_row(req.image_id) if not image: raise HTTPException(status_code=404, detail="Image not found") else: db_conn = get_db() row = db_conn.execute("SELECT * FROM book_images WHERE id = ?", (req.image_id,)).fetchone() if not row: db_conn.close() raise HTTPException(status_code=404, detail="Image not found") image = dict(row) info_parts = [] for key, label in [ ("title", "Title"), ("subtitle", "Subtitle"), ("author", "Author"), ("publisher", "Publisher"), ("year", "Year"), ("edition", "Edition"), ("language", "Language"), ("translator", "Translator"), ("series", "Series"), ("volume", "Volume"), ("condition", "Condition"), ("special_features", "Special features"), ("other_text", "Other details"), ]: value = image.get(key, "") if value and value != "Unknown": info_parts.append(f"{label}: {value}") book_info = "\n".join(info_parts) ocr_text = (image["raw_ocr_text"] or "")[:1500] photo_type = image["source_folder"] photo_desc = { "root": "title page or inner page", "front": "front cover", "side": "spine", }.get(photo_type, "book photo") try: from mistralai.client import Mistral client = Mistral(api_key=os.getenv("MISTRAL_API_KEY")) response = client.chat.complete( model="mistral-small-latest", messages=[ { "role": "user", "content": f"""You are writing a brief, appealing description for an antiquarian/second-hand book listing. This photo shows the **{photo_desc}** of a book. Based on the OCR text and extracted metadata below, write: 1. A one-line description of what this photo shows (e.g. \"Title page of a 1902 English edition published in Buenos Aires\") 2. A 2-3 sentence selling description highlighting what makes this book interesting or collectible (age, rarity, publisher, condition, historical significance, etc.) Keep it professional but warm - like a knowledgeable bookseller. Write in English. EXTRACTED METADATA: {book_info} RAW OCR TEXT FROM PHOTO: {ocr_text}""", } ], max_tokens=400, ) description = extract_message_text(response.choices[0].message.content) if not description: raise RuntimeError("Mistral returned an empty description") except Exception as exc: # noqa: BLE001 parts = [] if image["title"] and image["title"] != "Unknown": parts.append(f'"{image["title"]}"') if image["author"] and image["author"] != "Unknown": parts.append(f"by {image['author']}") if image["publisher"]: parts.append(f"published by {image['publisher']}") if image["year"]: parts.append(f"({image['year']})") description = ( f"Photo of {photo_desc}. " + " ".join(parts) if parts else f"Book photo ({photo_desc})" ) description += f"\n\n[AI generation failed: {str(exc)[:100]}]" if DB_MODE == "rest": rest_update_row(req.image_id, {"description": description}) else: assert db_conn is not None db_conn.execute("UPDATE book_images SET description = ? WHERE id = ?", (description, req.image_id)) db_conn.commit() db_conn.close() return {"description": description} class BatchGenerateRequest(BaseModel): start_id: int = 1 count: int = 10 @app.post("/api/generate-descriptions-batch") def generate_descriptions_batch(req: BatchGenerateRequest): if DB_MODE == "rest": rows = rest_fetch_rows("id,description") rows = [row for row in rows if not clean_text(row.get("description")) and row.get("id", 0) >= req.start_id] rows = rows[: req.count] else: conn = get_db() rows = conn.execute( "SELECT id FROM book_images WHERE (description IS NULL OR description = '') AND id >= ? ORDER BY id LIMIT ?", (req.start_id, req.count), ).fetchall() conn.close() results = [] for row in rows: image_id = row["id"] if isinstance(row, dict) else row[0] try: generate_description(GenerateRequest(image_id=image_id)) results.append({"id": image_id, "status": "ok"}) time.sleep(1.5) except Exception as exc: # noqa: BLE001 results.append({"id": image_id, "status": f"error: {exc}"}) if DB_MODE == "rest": remaining = sum(1 for row in rest_fetch_rows("description") if not clean_text(row.get("description"))) else: conn = get_db() remaining = conn.execute( "SELECT COUNT(*) FROM book_images WHERE description IS NULL OR description = ''" ).fetchone()[0] conn.close() return {"generated": len(results), "results": results, "remaining": remaining} # --- Chat endpoint ----------------------------------------------------------- class ChatRequest(BaseModel): question: str @app.post("/api/chat") def chat_about_books(req: ChatRequest): if DB_MODE == "rest": books = [ row for row in rest_fetch_rows("id,title,author,publisher,year,language,series,source_folder") if clean_text(row.get("title")) and clean_text(row.get("title")).lower() != "unknown" ] books = sorted(books, key=lambda row: clean_text(row.get("title")).lower()) else: conn = get_db() books = conn.execute( "SELECT id, title, author, publisher, year, language, series, source_folder FROM book_images WHERE title != 'Unknown' AND title != '' ORDER BY title" ).fetchall() conn.close() catalog_lines = [] for book in books: if isinstance(book, dict): book_id = book.get("id") title = book.get("title") author = book.get("author") publisher = book.get("publisher") year = book.get("year") language = book.get("language") series = book.get("series") else: book_id, title, author, publisher, year, language, series = book[0:7] parts = [f"#{book_id}"] if title: parts.append(f'"{title}"') if author and author != "Unknown": parts.append(f"by {author}") if publisher: parts.append(f"({publisher})") if year: parts.append(f"[{year}]") if language and language != "Unknown": parts.append(f"lang:{language}") if series: parts.append(f"series:{series}") catalog_lines.append(" ".join(parts)) catalog_text = "\n".join(catalog_lines) try: from mistralai.client import Mistral client = Mistral(api_key=os.getenv("MISTRAL_API_KEY")) response = client.chat.complete( model="mistral-small-latest", messages=[ { "role": "system", "content": f"""You are a helpful librarian/bookseller assistant. Answer questions about this book photo collection. Be specific - cite book titles, authors, and image IDs (#N) when relevant. CATALOG ({len(books)} identified images): {catalog_text}""", }, {"role": "user", "content": req.question}, ], max_tokens=2000, ) answer = extract_message_text(response.choices[0].message.content) return {"answer": answer or "AI chat returned an empty response."} except Exception as exc: # noqa: BLE001 return {"answer": f"AI chat unavailable: {exc}. Try searching in the gallery instead."} # --- Frontend ---------------------------------------------------------------- @app.get("/", response_class=HTMLResponse) def serve_frontend(): config_json = json.dumps(app_config()).replace(" Bibliothek - Family Access
Private family link

Bibliothek

Enter the shared family password to browse the catalog, review OCR, and help clean up the collection.

""" FRONTEND_HTML = r""" Bibliothek - Book Catalog
Private family library
Bibliothek
Sign in with your invited email and I will send you a private magic link. Only approved family members can access the catalog.
📚

Bibliothek

Book-centered catalog for desktop and phone
0/0 described
Batch descriptions still run per photo, but the gallery now groups matching titles into books.
Verification game

Swipe right if the OCR is right. Swipe left if it needs fixing.

Loading the review queue...
How many books are in Spanish? Which books are from before 1900? List books with multiple photos What authors appear most often? Which titles still need descriptions?
Ask me anything about the collection and I will answer from the catalog.
Loading...
Book photo
"""