3270a9fe69
- _body() read the full declared Content-Length into memory before any size check, so MAX_UPLOAD_BYTES could only reject an upload already held in RAM. nginx caps this on the proxied path, but the container also listens on the LAN, so the app now enforces its own 20MB ceiling and closes the connection rather than reading. - SQLite ran with the default rollback journal and 5s lock timeout, chosen when a row was a few KB; rows now carry the original photos, so two people grading at once could block each other's History. WAL + 30s. - No socket timeout meant a stalled keep-alive connection held a worker thread indefinitely. - An edge excluded as a different material was dropped from the UI with no explanation, presenting three sides as though they were all four.
520 lines
21 KiB
Python
520 lines
21 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 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
|
|
# 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 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()
|