Initial commit: Card Grader deployed to hippofam.com/cards
PWA card-grading app, deployed behind Nginx Proxy Manager on Unraid with basic auth. Includes CARD_GRADER_BASE_PATH support for running under a sub-path, and Docker/compose config for the Unraid deployment.
This commit is contained in:
Barely Removable
commit
c7bd71a3e1
19 changed files with 3618 additions and 0 deletions
410
app.py
Normal file
410
app.py
Normal file
|
|
@ -0,0 +1,410 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Card Grader — local web app.
|
||||
|
||||
Stdlib only, except the optional `anthropic` and `pillow` packages (see
|
||||
README.md). Run it with: python3 app.py
|
||||
Then open http://localhost:8778
|
||||
|
||||
Data lives in grades.db next to this file. Back that one file up and you have
|
||||
backed up your whole grading history.
|
||||
|
||||
This is the single-purpose extraction of the grading feature out of a larger
|
||||
Pokemon card flipping tracker: no inventory, no price lookups, no catalog —
|
||||
just "how would this card likely grade," estimated from photos, with a local
|
||||
history of what you've checked. See cardimage.py for the actual mechanism:
|
||||
two of PSA's four categories (centering, edge whitening) are measured
|
||||
directly from the pixels rather than eyeballed by the model.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import urllib.parse
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
import cardimage
|
||||
import store
|
||||
import vision
|
||||
|
||||
# A photo arrives base64-encoded inside JSON (avoids hand-rolling multipart
|
||||
# parsing on top of the stdlib server). Cap it so a stray upload can't wedge
|
||||
# the process reading an unbounded body.
|
||||
MAX_UPLOAD_BYTES = 12 * 1024 * 1024
|
||||
# More angles help — front, back, corner close-ups — but past a handful the
|
||||
# extra photos cost tokens without adding evidence.
|
||||
MAX_GRADE_IMAGES = 6
|
||||
# Small enough to sit in a history list without bloating the database; big
|
||||
# enough to still recognise the card at a glance.
|
||||
THUMBNAIL_TARGET_PX = 320
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
STATIC_DIR = os.path.join(BASE_DIR, "static")
|
||||
PORT = int(os.environ.get("PORT", "8778"))
|
||||
|
||||
# Set when this app is reverse-proxied under a sub-path (e.g. "/cards")
|
||||
# rather than served at its own domain root. The four files that reference
|
||||
# absolute URLs (index.html, manifest.json, sw.js, app.js) get this spliced
|
||||
# in for every "__BASE__" placeholder — see _static_templated. Everything
|
||||
# else (routing included) works the same either way, since nginx forwards
|
||||
# the full "/cards/..." path through unchanged and _route() below strips it
|
||||
# right back off before matching.
|
||||
BASE_PATH = (os.environ.get("CARD_GRADER_BASE_PATH") or "").rstrip("/")
|
||||
TEMPLATED_STATIC = {"index.html", "manifest.json", "sw.js", "app.js"}
|
||||
|
||||
# Set CARD_GRADER_LOCK=1 when exposing this beyond your own machine. It stops
|
||||
# visitors rewriting your stored settings — including swapping the model to
|
||||
# the most expensive one, or replacing your API key. Grading still works
|
||||
# normally; only the settings write is refused.
|
||||
SETTINGS_LOCKED = os.environ.get("CARD_GRADER_LOCK", "").strip() not in ("", "0")
|
||||
|
||||
CONTENT_TYPES = {
|
||||
".html": "text/html; charset=utf-8",
|
||||
".css": "text/css; charset=utf-8",
|
||||
".js": "application/javascript; charset=utf-8",
|
||||
".svg": "image/svg+xml",
|
||||
".png": "image/png",
|
||||
".ico": "image/x-icon",
|
||||
".json": "application/manifest+json",
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------- helpers
|
||||
|
||||
|
||||
def _decode_uploaded_images(raw_images):
|
||||
"""Decode base64 uploads. Returns (images, error_message).
|
||||
|
||||
The format is decided from the bytes, not the filename — phones hand
|
||||
over names this code has no business trusting (no extension, a
|
||||
content:// URI, .HEIC), and an unreadable *name* is no reason to reject
|
||||
a perfectly readable *photo*. See cardimage.normalize_upload.
|
||||
"""
|
||||
images, total = [], 0
|
||||
for entry in raw_images or []:
|
||||
raw = (entry or {}).get("image_base64") or ""
|
||||
filename = (entry or {}).get("filename") or "upload"
|
||||
if not raw:
|
||||
return None, "No image received"
|
||||
try:
|
||||
image_bytes = base64.b64decode(raw, validate=True)
|
||||
except Exception:
|
||||
return None, "Image data was not valid base64"
|
||||
total += len(image_bytes)
|
||||
if total > MAX_UPLOAD_BYTES:
|
||||
return None, "Images are {:.1f} MB combined; the limit is {} MB.".format(
|
||||
total / 1024 / 1024, MAX_UPLOAD_BYTES // 1024 // 1024)
|
||||
image_bytes, filename, error = cardimage.normalize_upload(image_bytes, filename)
|
||||
if error:
|
||||
return None, error
|
||||
images.append((image_bytes, filename))
|
||||
return images, None
|
||||
|
||||
|
||||
def _public_settings():
|
||||
"""Settings safe to hand to a browser.
|
||||
|
||||
The stored API key never leaves the server. Once this is reachable by
|
||||
anyone but you — which is the whole point of putting it behind a tunnel
|
||||
for friends — returning the raw key would hand every visitor the ability
|
||||
to spend your Anthropic credits anywhere they like. They get a boolean
|
||||
saying whether one is configured, which is all the UI needs.
|
||||
"""
|
||||
s = store.get_settings()
|
||||
return {
|
||||
"vision_model": s.get("vision_model"),
|
||||
"vision_effort": s.get("vision_effort"),
|
||||
"server_key_configured": bool(s.get("anthropic_api_key")),
|
||||
"settings_locked": SETTINGS_LOCKED,
|
||||
}
|
||||
|
||||
|
||||
def _make_thumbnail(image_bytes):
|
||||
"""Small base64 JPEG for the history list, or None if Pillow is missing."""
|
||||
if not cardimage.available():
|
||||
return None
|
||||
try:
|
||||
from PIL import Image
|
||||
img = Image.open(io.BytesIO(image_bytes))
|
||||
img.load()
|
||||
if img.mode not in ("RGB", "L"):
|
||||
img = img.convert("RGB")
|
||||
factor = THUMBNAIL_TARGET_PX / float(max(img.size))
|
||||
if factor < 1:
|
||||
img = img.resize(
|
||||
(max(1, int(img.width * factor)), max(1, int(img.height * factor))),
|
||||
Image.LANCZOS)
|
||||
buffer = io.BytesIO()
|
||||
img.convert("RGB").save(buffer, format="JPEG", quality=78)
|
||||
return "data:image/jpeg;base64," + base64.standard_b64encode(
|
||||
buffer.getvalue()).decode("ascii")
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- handler
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.1"
|
||||
server_version = "CardGrader"
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
if str(args[1] if len(args) > 1 else "").startswith(("4", "5")):
|
||||
sys.stderr.write(" {} {}\n".format(self.address_string(), fmt % args))
|
||||
|
||||
# ------------------------------------------------------------ helpers
|
||||
|
||||
def _send(self, code, body, content_type="application/json; charset=utf-8",
|
||||
cache_control="no-store"):
|
||||
if isinstance(body, str):
|
||||
body = body.encode("utf-8")
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", cache_control)
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def _json(self, obj, code=200):
|
||||
self._send(code, json.dumps(obj, default=str))
|
||||
|
||||
def _error(self, message, code=400):
|
||||
self._json({"error": message}, code)
|
||||
|
||||
def _body(self):
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
if not length:
|
||||
return {}
|
||||
try:
|
||||
return json.loads(self.rfile.read(length).decode("utf-8"))
|
||||
except ValueError:
|
||||
return {}
|
||||
|
||||
def _route(self):
|
||||
path = urllib.parse.urlparse(self.path).path
|
||||
# nginx forwards the full "/cards/..." URI through unchanged (same
|
||||
# pattern as this box's other proxied apps, each of which handles
|
||||
# its own URL-base) — strip it back off so route matching below
|
||||
# doesn't need to know whether it's mounted at the domain root or
|
||||
# under a sub-path.
|
||||
if BASE_PATH and (path == BASE_PATH or path.startswith(BASE_PATH + "/")):
|
||||
path = path[len(BASE_PATH):]
|
||||
return path.rstrip("/") or "/"
|
||||
|
||||
def _static(self, relative):
|
||||
safe = os.path.normpath(relative).lstrip(os.sep)
|
||||
path = os.path.join(STATIC_DIR, safe)
|
||||
if not path.startswith(STATIC_DIR) or not os.path.isfile(path):
|
||||
return self._error("Not found", 404)
|
||||
ext = os.path.splitext(path)[1]
|
||||
if os.path.basename(path) in TEMPLATED_STATIC:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
body = handle.read().replace("__BASE__", BASE_PATH).encode("utf-8")
|
||||
else:
|
||||
with open(path, "rb") as handle:
|
||||
body = handle.read()
|
||||
# "no-cache" (revalidate every time), NOT "no-store". They sound
|
||||
# interchangeable and are not: Chrome refuses to register a service
|
||||
# worker whose script came back no-store, so the whole install-as-an-
|
||||
# app path silently dies. no-cache still guarantees you never get a
|
||||
# stale file, since the browser revalidates before using it.
|
||||
self._send(200, body, CONTENT_TYPES.get(ext, "application/octet-stream"),
|
||||
cache_control="no-cache")
|
||||
|
||||
# ---------------------------------------------------------------- GET
|
||||
|
||||
def do_GET(self):
|
||||
route = self._route()
|
||||
try:
|
||||
if route == "/" or route == "/index.html":
|
||||
return self._static("index.html")
|
||||
# Served from the root deliberately: a service worker's scope is
|
||||
# its own directory, so one living at /static/sw.js could only
|
||||
# ever control /static/ — not the app. Same file, root URL.
|
||||
if route == "/sw.js":
|
||||
return self._static("sw.js")
|
||||
if route.startswith("/static/"):
|
||||
return self._static(route[len("/static/"):])
|
||||
if route == "/api/settings":
|
||||
return self._json(_public_settings())
|
||||
if route == "/api/vision-models":
|
||||
return self._json(vision.price_guide())
|
||||
if route == "/api/history":
|
||||
return self._json({"grades": store.list_grades()})
|
||||
if route.startswith("/api/history/"):
|
||||
grade = store.get_grade(int(route.split("/")[3]))
|
||||
if not grade:
|
||||
return self._error("No such grade", 404)
|
||||
return self._json({"grade": grade})
|
||||
return self._error("Not found", 404)
|
||||
except (ValueError, IndexError):
|
||||
return self._error("Bad request path", 400)
|
||||
except Exception as exc:
|
||||
return self._error("Server error: {}".format(exc), 500)
|
||||
|
||||
# --------------------------------------------------------------- POST
|
||||
|
||||
def do_POST(self):
|
||||
route = self._route()
|
||||
try:
|
||||
body = self._body()
|
||||
if route == "/api/settings":
|
||||
if SETTINGS_LOCKED:
|
||||
return self._error(
|
||||
"Settings are locked on this server. Use your own API "
|
||||
"key in this browser instead.", 403)
|
||||
store.save_settings(body)
|
||||
return self._json(_public_settings())
|
||||
if route == "/api/grade":
|
||||
return self._grade_card(body)
|
||||
return self._error("Not found", 404)
|
||||
except (ValueError, IndexError):
|
||||
return self._error("Bad request", 400)
|
||||
except Exception as exc:
|
||||
return self._error("Server error: {}".format(exc), 500)
|
||||
|
||||
def do_PATCH(self):
|
||||
route = self._route()
|
||||
try:
|
||||
if route.startswith("/api/history/"):
|
||||
grade_id = int(route.split("/")[3])
|
||||
body = self._body()
|
||||
if store.get_grade(grade_id) is None:
|
||||
return self._error("No such grade", 404)
|
||||
grade = store.update_grade_label(grade_id, body.get("label"))
|
||||
return self._json({"grade": grade})
|
||||
return self._error("Not found", 404)
|
||||
except (ValueError, IndexError):
|
||||
return self._error("Bad request", 400)
|
||||
except Exception as exc:
|
||||
return self._error("Server error: {}".format(exc), 500)
|
||||
|
||||
def do_DELETE(self):
|
||||
route = self._route()
|
||||
try:
|
||||
if route.startswith("/api/history/"):
|
||||
store.delete_grade(int(route.split("/")[3]))
|
||||
return self._json({"ok": True})
|
||||
return self._error("Not found", 404)
|
||||
except (ValueError, IndexError):
|
||||
return self._error("Bad request", 400)
|
||||
except Exception as exc:
|
||||
return self._error("Server error: {}".format(exc), 500)
|
||||
|
||||
# ------------------------------------------------------------- grade
|
||||
|
||||
def _grade_card(self, body):
|
||||
"""Estimate the PSA grade for a card, from uploaded photo(s).
|
||||
|
||||
Always returns the estimate. Saves it to history too, unless the
|
||||
caller explicitly opts out with save=false — a one-off sanity check
|
||||
before you've decided a card is worth logging.
|
||||
"""
|
||||
settings = store.get_settings()
|
||||
|
||||
raw_images = body.get("images") or []
|
||||
if not isinstance(raw_images, list) or not raw_images:
|
||||
return self._error("Send at least one image.", 400)
|
||||
if len(raw_images) > MAX_GRADE_IMAGES:
|
||||
return self._error("Send at most {} images.".format(MAX_GRADE_IMAGES), 400)
|
||||
images, error = _decode_uploaded_images(raw_images)
|
||||
if error:
|
||||
return self._error(error, 400)
|
||||
|
||||
model = body.get("model") if body.get("model") in vision.MODELS else settings.get("vision_model")
|
||||
|
||||
# A key sent with the request wins over the server's own. That is what
|
||||
# makes this shareable: hand a friend the URL and they can bring their
|
||||
# own Anthropic key rather than spending yours. Never stored — it
|
||||
# lives in their browser and is used for this one call.
|
||||
caller_key = (body.get("api_key") or "").strip() or None
|
||||
|
||||
print("[grade] calling vision model={} on {} image(s){}…".format(
|
||||
model, len(images), " (caller key)" if caller_key else ""), flush=True)
|
||||
t0 = time.time()
|
||||
try:
|
||||
result = vision.grade_card(
|
||||
images,
|
||||
caller_key or settings.get("anthropic_api_key") or None,
|
||||
model=model,
|
||||
effort=settings.get("vision_effort"),
|
||||
)
|
||||
except vision.VisionError as exc:
|
||||
return self._error(str(exc), 502)
|
||||
print("[grade] done in {:.1f}s".format(time.time() - t0), flush=True)
|
||||
|
||||
grade = {
|
||||
"image_count": len(images),
|
||||
"closeups": result["closeups"],
|
||||
"card_type": result["card_type"],
|
||||
"card_note": result["card_note"],
|
||||
"edge_measurements": result["edge_measurements"],
|
||||
"centering_measurement": result["centering_measurement"],
|
||||
"estimated_grade": result["estimated_grade"],
|
||||
"grade_low": result["grade_low"],
|
||||
"grade_high": result["grade_high"],
|
||||
"confidence": result["confidence"],
|
||||
"categories": result["categories"],
|
||||
"limitations": result["limitations"],
|
||||
"note": result["note"],
|
||||
"usage": result["usage"],
|
||||
"estimated_cost": vision.estimate_cost(result["usage"]),
|
||||
}
|
||||
|
||||
grade_id = None
|
||||
if body.get("save", True):
|
||||
thumbnail = _make_thumbnail(images[0][0])
|
||||
grade_id = store.save_grade(grade, thumbnail=thumbnail, label=body.get("label"))
|
||||
|
||||
return self._json({"grade": grade, "grade_id": grade_id}, 201)
|
||||
|
||||
|
||||
def lan_ip():
|
||||
"""Best-effort LAN address, so a phone can reach this."""
|
||||
import socket
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
try:
|
||||
sock.connect(("8.8.8.8", 80))
|
||||
return sock.getsockname()[0]
|
||||
except Exception:
|
||||
return None
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
class Server(ThreadingHTTPServer):
|
||||
allow_reuse_address = True
|
||||
daemon_threads = True
|
||||
|
||||
|
||||
def main():
|
||||
store.init()
|
||||
try:
|
||||
server = Server(("0.0.0.0", PORT), Handler)
|
||||
except OSError as exc:
|
||||
print("\n Could not start on port {}: {}".format(PORT, exc))
|
||||
print(" Something else is using it. Free it with:")
|
||||
print(" lsof -ti :{} | xargs kill\n".format(PORT))
|
||||
raise SystemExit(1)
|
||||
ip = lan_ip()
|
||||
|
||||
print("\n Card Grader")
|
||||
print(" " + "-" * 42)
|
||||
print(" On this Mac: http://localhost:{}".format(PORT))
|
||||
if ip:
|
||||
print(" On your phone: http://{}:{} (same Wi-Fi)".format(ip, PORT))
|
||||
print(" Database: {}".format(store.DB_PATH))
|
||||
print("\n Ctrl-C to stop.\n")
|
||||
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print("\n Stopped.")
|
||||
server.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue