602bce8ce3
Storing originals for Regrade grew the database ~16MB per grade at the upload cap. Photos now expire on a configurable window (default 7 days, counted from last grade so regrading resets it) and the file is VACUUMed so the space is actually returned — clearing the column alone only moves pages to the freelist, which would have made the whole feature a no-op on disk. Grades, thumbnails and measurements are never pruned; a pruned card's Regrade falls back to the existing re-pick path, now with a banner saying why.
555 lines
23 KiB
Python
555 lines
23 KiB
Python
#!/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 threading
|
|
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). This caps the DECODED image bytes.
|
|
MAX_UPLOAD_BYTES = 12 * 1024 * 1024
|
|
# Hard ceiling on the request body itself, enforced before a single byte is
|
|
# read. MAX_UPLOAD_BYTES alone cannot do this job: it is checked only after
|
|
# the whole body has already been pulled into memory, so a client declaring
|
|
# a multi-gigabyte Content-Length would be obliged first and rejected
|
|
# afterwards. nginx caps this too, but only on the proxied path — the
|
|
# container also listens on the LAN (see docker-compose.yml), so the app has
|
|
# to enforce its own limit rather than inherit one.
|
|
# Sized for the largest legitimate request: MAX_UPLOAD_BYTES of image is
|
|
# ~4/3 that as base64, plus JSON overhead. Kept in step with nginx's
|
|
# client_max_body_size in nginx/card-grader.conf.
|
|
MAX_BODY_BYTES = 20 * 1024 * 1024
|
|
# How often to expire stored source photos (see store.prune_source_images).
|
|
# Six-hourly rather than daily so a long-running container doesn't hold a
|
|
# week's worth of extra images for up to another day past the window, and
|
|
# so the first sweep after a restart isn't the only one that ever runs.
|
|
PRUNE_INTERVAL_SECONDS = 6 * 60 * 60
|
|
# 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"
|
|
# Without this a client that opens a connection and then stalls holds a
|
|
# worker thread forever — keep-alive means the handler sits waiting on
|
|
# the next request line indefinitely. Generous enough for a phone
|
|
# uploading photos over a slow link, finite enough that a dead or
|
|
# deliberately-slow connection lets go.
|
|
timeout = 120
|
|
|
|
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):
|
|
"""Parsed JSON body, or None if the request was refused outright.
|
|
|
|
Returns None (having already sent a response) when the body is over
|
|
the limit, so callers must stop rather than carry on with an empty
|
|
dict — which is what an unparseable body still yields.
|
|
"""
|
|
try:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
except ValueError:
|
|
self._reject_body("Malformed Content-Length.")
|
|
return None
|
|
if length < 0:
|
|
self._reject_body("Malformed Content-Length.")
|
|
return None
|
|
if length > MAX_BODY_BYTES:
|
|
self._reject_body(
|
|
"That request is {:.1f} MB; the limit is {} MB.".format(
|
|
length / 1024 / 1024, MAX_BODY_BYTES // 1024 // 1024))
|
|
return None
|
|
if not length:
|
|
return {}
|
|
try:
|
|
return json.loads(self.rfile.read(length).decode("utf-8"))
|
|
except ValueError:
|
|
return {}
|
|
|
|
def _reject_body(self, message):
|
|
"""413 for an oversized//malformed body, then hang up.
|
|
|
|
The body is deliberately never read, so the socket still holds
|
|
whatever the client is sending — reusing it under keep-alive would
|
|
parse that leftover payload as the next request. Closing is the only
|
|
safe way to refuse without reading.
|
|
"""
|
|
self.close_connection = True
|
|
self._json({"error": message}, 413)
|
|
|
|
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 body is None:
|
|
return # _body already sent 413 and closed
|
|
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)
|
|
if route.startswith("/api/history/") and route.endswith("/regrade"):
|
|
grade_id = int(route.split("/")[3])
|
|
return self._regrade_card(grade_id, 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 body is None:
|
|
return # _body already sent 413 and closed
|
|
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 _estimate(self, images, body, settings):
|
|
"""Run vision.grade_card and shape its result into a storable dict.
|
|
|
|
Shared by fresh grading and Regrade — the only difference between
|
|
those two call sites is where `images` comes from (a fresh upload vs.
|
|
photos already on file) and what happens to the result afterward
|
|
(insert a new row vs. overwrite an existing one), not how the
|
|
estimate itself gets produced. Raises vision.VisionError on failure;
|
|
callers turn that into an HTTP response themselves, since a 502 from
|
|
a first grade and a 502 from a regrade warrant slightly different
|
|
wording.
|
|
"""
|
|
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()
|
|
result = vision.grade_card(
|
|
images,
|
|
caller_key or settings.get("anthropic_api_key") or None,
|
|
model=model,
|
|
effort=settings.get("vision_effort"),
|
|
)
|
|
print("[grade] done in {:.1f}s".format(time.time() - t0), flush=True)
|
|
|
|
return {
|
|
"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"],
|
|
"aspect_measurement": result["aspect_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"]),
|
|
}
|
|
|
|
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)
|
|
|
|
try:
|
|
grade = self._estimate(images, body, settings)
|
|
except vision.VisionError as exc:
|
|
return self._error(str(exc), 502)
|
|
|
|
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"),
|
|
source_images=images)
|
|
|
|
return self._json({"grade": grade, "grade_id": grade_id}, 201)
|
|
|
|
def _regrade_card(self, grade_id, body):
|
|
"""Re-run grading on an existing history row, in place.
|
|
|
|
Uses the photo(s) already stored for this card unless the caller
|
|
supplies fresh ones — which happens for a card graded before this
|
|
feature existed (or saved without images some other way), where
|
|
there's nothing on file to re-run with. Either way the row is
|
|
updated rather than duplicated, and freshly-supplied photos get
|
|
stored too, so the NEXT regrade on this card is instant.
|
|
"""
|
|
existing = store.get_grade(grade_id)
|
|
if not existing:
|
|
return self._error("No such grade", 404)
|
|
|
|
raw_images = body.get("images")
|
|
newly_supplied = bool(raw_images)
|
|
if newly_supplied:
|
|
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)
|
|
else:
|
|
images = store.get_grade_images(grade_id)
|
|
if not images:
|
|
return self._error(
|
|
"No photos were saved for this card, so it can't be "
|
|
"regraded automatically — pick the photo(s) again.", 409)
|
|
|
|
settings = store.get_settings()
|
|
try:
|
|
grade = self._estimate(images, body, settings)
|
|
except vision.VisionError as exc:
|
|
return self._error(str(exc), 502)
|
|
|
|
thumbnail = _make_thumbnail(images[0][0])
|
|
grade = store.update_grade_result(
|
|
grade_id, grade, thumbnail=thumbnail,
|
|
source_images=images if newly_supplied else None)
|
|
|
|
return self._json({"grade": grade}, 200)
|
|
|
|
|
|
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 _prune_forever():
|
|
"""Expire stored source photos on a timer, for the life of the process.
|
|
|
|
A daemon thread rather than a cron entry so the retention window holds
|
|
on any host this runs on, without a second thing to install and keep in
|
|
step. Never lets an exception end the loop: a failed sweep should cost
|
|
one cycle, not disable pruning until the next restart.
|
|
"""
|
|
while True:
|
|
try:
|
|
pruned = store.prune_source_images()
|
|
if pruned:
|
|
print("[prune] dropped stored photos from {} grade(s) older "
|
|
"than {} days".format(pruned, store.SOURCE_IMAGE_RETENTION_DAYS),
|
|
flush=True)
|
|
except Exception as exc:
|
|
print("[prune] failed: {}".format(exc), flush=True)
|
|
time.sleep(PRUNE_INTERVAL_SECONDS)
|
|
|
|
|
|
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))
|
|
if store.SOURCE_IMAGE_RETENTION_DAYS > 0:
|
|
print(" Photos kept: {} days (grades kept forever)".format(
|
|
store.SOURCE_IMAGE_RETENTION_DAYS))
|
|
else:
|
|
print(" Photos kept: forever (pruning disabled)")
|
|
print("\n Ctrl-C to stop.\n")
|
|
|
|
# Sweep once at boot so a container that restarts often still expires
|
|
# things, then hand off to the timer.
|
|
threading.Thread(target=_prune_forever, daemon=True).start()
|
|
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\n Stopped.")
|
|
server.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|