ninja_hippo/card-grader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
card-grader/app.py
Barely Removable 2b3bbaf40f Add a structured authenticity signal with a distinct banner, instead of burying it in card_note 
The Charizard grade that prompted this had the model writing 'possible
non-standard/proxy print' directly into card_note -- the only free-text
field available -- since there was nowhere else for that observation to go.
That's why it got no visual treatment: the UI renders card_note as a plain
grey subtitle, identical to any ordinary card name.

Added authenticity as its own top-level field (flag: none/worth_checking/
likely_not_genuine, plus an observation) alongside the four grading
categories but explicitly separate from them -- this is about whether the
card is a genuine product at all, not its condition, and PSA authenticates
before it grades, so a flagged card's estimated_grade goes to null rather
than a number.

Prompted with concrete tells for both classes of concern: fan-made/proxy
prints (fictional sets, non-existent number combos, home-printer texture,
'proxy' watermarks) and counterfeits of real cards (colour/font/holo
mismatches against genuine copies). Trimming (already covered under CORNERS
from an earlier fix) now also sets this field when flagged, so the UI has
one place to check regardless of which specific issue triggered it.

UI: a full banner (not a hint line) above the slab -- critical red for
likely_not_genuine, warning amber for worth_checking -- plus a matching dot
next to the card name in History so it's visible without opening the card.
Both colours reuse --critical/--warning rather than the brand red, since
this is exactly the 'distrust this' signal those are already reserved for.

Migration tested against the live schema shape; existing rows get
authenticity=None and render with no banner, as before.
2026-08-23 08:19:58 -07:00

628 lines
26 KiB
Python
Raw Blame History

#!/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")
# When set, only this ONE username (from the reverse proxy's basic auth, see
# Handler._username) may write server settings — everyone else sees the same
# read-only view CARD_GRADER_LOCK produces, regardless of CARD_GRADER_LOCK's
# own value. This is strictly narrower than the blanket lock: it names one
# person rather than locking out or opening up to everyone at once. Blank
# (the default) falls back to CARD_GRADER_LOCK's all-or-nothing behaviour.
ADMIN_USERNAME = os.environ.get("CARD_GRADER_ADMIN_USER", "").strip()
def _settings_writable(username):
if ADMIN_USERNAME:
return username == ADMIN_USERNAME
return not SETTINGS_LOCKED
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(username):
"""Settings safe to hand to a browser.
Stored API keys never leave 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 credits anywhere they like. They get a boolean per provider
saying whether one is configured, which is all the UI needs.
"""
s = store.get_settings()
writable = _settings_writable(username)
return {
"vision_model": s.get("vision_model"),
"vision_effort": s.get("vision_effort"),
"anthropic_key_configured": bool(s.get("anthropic_api_key")),
"openai_key_configured": bool(s.get("openai_api_key")),
"is_admin": writable,
# Kept for the frontend's existing "locked" UI treatment — now means
# "not writable by YOU", whatever the reason, rather than a single
# global flag.
"settings_locked": not writable,
}
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 _username(self):
"""Who the reverse proxy authenticated, or None.
Two sources, because which one is available depends on the proxy's
config: an explicit X-Auth-User (nginx's $remote_user) if it's been
set, otherwise the Basic credentials nginx forwards upstream by
default. Only the username is ever read — the password half is
discarded without being looked at, since the proxy already validated
it and this app has no business re-checking it.
For ATTRIBUTION ONLY. Anything on the LAN can reach this app
directly (see docker-compose.yml) and set either header freely, so
this must never gate access to anything — it answers "who ran up
this bill", not "who is allowed in".
"""
header = (self.headers.get("X-Auth-User") or "").strip()
if header:
return header[:64]
auth = (self.headers.get("Authorization") or "").strip()
if auth.lower().startswith("basic "):
try:
decoded = base64.b64decode(auth[6:], validate=True).decode("utf-8", "replace")
except Exception:
return None
name = decoded.split(":", 1)[0].strip()
return name[:64] or None
return None
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(self._username()))
if route == "/api/vision-models":
return self._json(vision.price_guide())
if route == "/api/usage":
data = store.usage_by_user()
data["you"] = self._username()
return self._json(data)
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":
username = self._username()
if not _settings_writable(username):
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(username))
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")
# Which stored server key applies depends on which model this grade
# actually runs on — Sonnet needs the Anthropic key, GPT-5.6 Sol
# needs the OpenAI one, and they are never interchangeable.
provider = vision.MODELS.get(model, {}).get("provider", "anthropic")
server_key = settings.get("openai_api_key" if provider == "openai" else "anthropic_api_key")
# 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 key (matching whichever provider `model` needs) 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 server_key or None,
model=model,
effort=settings.get("vision_effort"),
)
print("[grade] done in {:.1f}s".format(time.time() - t0), flush=True)
# Surfaced so the caller can log who paid — the whole point of the
# split in the spend ledger.
self._last_used_own_key = bool(caller_key)
return {
"image_count": len(images),
"closeups": result["closeups"],
"card_type": result["card_type"],
"card_note": result["card_note"],
"authenticity": result["authenticity"],
"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)
# Logged even when save=false: the call was still billed, and a
# ledger that only counted saved cards would under-report spend.
store.log_grade_event(grade, username=self._username(), grade_id=grade_id,
own_key=getattr(self, "_last_used_own_key", False),
kind="grade")
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])
store.log_grade_event(grade, username=self._username(), grade_id=grade_id,
own_key=getattr(self, "_last_used_own_key", False),
kind="regrade")
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()
Reference in a new issue View git blame Copy permalink