"""Sample-data backend for the AI review dashboard: runs, review rows and the approve gate.
npm run build && python server_stub.py # page + API on http://127.0.0.1:8000
python server_stub.py & npm run dev # or Vite on :5173, proxying /api here
It stands in for the project's backend until that exists. A real backend serves the
same endpoints (templates/contract.md) with the same rules: bands from bands.py, and
approval refused while any flagged field is unverified. The page never decides that.
Standard library only.
"""
import json
import re
import threading
import time
from html import escape
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse
from bands import BANDS, field_band, needs_review
HERE = Path(__file__).resolve().parent
API = "/api" # same as API_BASE in index.html
REASON = {"green": None, "amber": "Slightly unclear, sample check", "red": "Unclear print, check every word",
"blank": "Required field is empty", "unscored": "Not found in the transcription"}
# (field, value, word scores, mandatory). Swap in the fields the project's extraction returns.
SAMPLE_FIELDS = [
("Invoice number", "INV-1001", [1.0], True),
("Invoice date", "2026-10-01", [0.99, 0.97], True),
("Vendor", "Contoso Supplies", [0.98, 0.92], True),
("PO number", "PO-4471", [0.86], False),
("Due date", "", [], True),
("Total", "2,450.00", [0.99], True),
("Tax", "196.00", [], False),
]
RUNS, ROWS = {}, {}
_next_row = [1]
def put_run(job_id, document, fields, status="done", pages_done=1, pages_total=1, model="sample-model"):
"""Add or replace a run. fields: [(field, value, word scores, mandatory)]."""
RUNS[job_id] = {"job_id": job_id, "document": document, "status": status, "model": model,
"pages_done": pages_done, "pages_total": pages_total, "approved": False, "error": None}
ROWS[job_id] = []
for field, value, words, mandatory in fields:
band = field_band(words, value, mandatory)
ROWS[job_id].append({"row_id": _next_row[0], "field": field, "value": value, "band": band,
"score": min(words) if words else None, "reason": REASON[band],
"verified": False, "corrected_value": None})
_next_row[0] += 1
return RUNS[job_id]
def reset():
RUNS.clear()
ROWS.clear()
put_run("job-1002", "invoice-1002.pdf", SAMPLE_FIELDS, status="running", pages_done=1, pages_total=3)
put_run("job-1001", "invoice-1001.pdf", SAMPLE_FIELDS, pages_total=2, pages_done=2)
def run_view(job_id):
run, rows = RUNS[job_id], ROWS[job_id]
done = run["status"] == "done"
return {**run, "summary": {b: sum(r["band"] == b for r in rows) for b in BANDS} if done else None}
def unverified(job_id):
return [r for r in ROWS[job_id] if needs_review(r["band"]) and not r["verified"]]
# ---- routes: (method, path regex, handler(match groups..., body) -> (status, payload)) ----
def list_runs(body):
return 200, {"runs": [run_view(j) for j in RUNS]}
def get_run(job_id, body):
return 200, run_view(job_id)
def review(job_id, body):
return 200, {"rows": ROWS[job_id] if RUNS[job_id]["status"] == "done" else []}
def patch_row(row_id, body):
row = next(r for rows in ROWS.values() for r in rows if r["row_id"] == int(row_id))
row.update({k: body[k] for k in ("corrected_value", "verified") if k in body})
return 200, row
def verify_remaining(job_id, body):
left = unverified(job_id)
for r in left:
r["verified"] = True
return 200, {"verified": len(left)}
def approve(job_id, body):
left = unverified(job_id)
if RUNS[job_id]["status"] != "done":
return 409, {"detail": "The run hasn't finished extracting."}
if left: # the approve gate
return 409, {"detail": f"{len(left)} flagged field(s) still need verifying."}
RUNS[job_id]["approved"] = True
return 200, {"ok": True}
def document(job_id, body):
"""A stand-in for the source document. A real backend returns the PDF or page image."""
lines = "".join(f'{escape(f)}: {escape(v)}'
for i, (f, v, *_) in enumerate(SAMPLE_FIELDS))
svg = (f'')
return 200, svg
ROUTES = [
("GET", r"/runs", list_runs),
("GET", r"/runs/([\w-]+)", get_run),
("GET", r"/runs/([\w-]+)/review", review),
("GET", r"/runs/([\w-]+)/document", document),
("PATCH", r"/review/(\d+)", patch_row),
("POST", r"/runs/([\w-]+)/verify-remaining", verify_remaining),
("POST", r"/runs/([\w-]+)/approve", approve),
("POST", r"/set-dark", lambda body: (200, {})),
]
class Handler(SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=str(HERE / "dist"), **kwargs)
def log_message(self, *args):
pass
def do_GET(self):
self.api("GET") if self.path.startswith(API + "/") else super().do_GET()
def do_POST(self):
self.api("POST")
def do_PATCH(self):
self.api("PATCH")
def api(self, method):
path = urlparse(self.path).path[len(API):]
size = int(self.headers.get("Content-Length") or 0)
body = json.loads(self.rfile.read(size) or b"{}") if size else {}
status, payload = 404, {"detail": f"No route for {method} {path}"}
for m, pattern, fn in ROUTES:
match = re.fullmatch(pattern, path)
if m == method and match:
try:
status, payload = fn(*match.groups(), body)
except (KeyError, StopIteration):
status, payload = 404, {"detail": "Not found"}
break
svg = isinstance(payload, str)
data = payload.encode() if svg else json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "image/svg+xml" if svg else "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def make_server(host="127.0.0.1", port=8000):
reset()
return ThreadingHTTPServer((host, port), Handler)
def simulate_extraction():
"""Demo only: the running sample run reads one page every 3 seconds, then finishes."""
while True:
time.sleep(3)
for run in RUNS.values():
if run["status"] == "running":
run["pages_done"] += 1
if run["pages_done"] >= run["pages_total"]:
run["status"] = "done"
if __name__ == "__main__":
server = make_server()
threading.Thread(target=simulate_extraction, daemon=True).start()
print("Sample backend on http://127.0.0.1:8000 (Ctrl+C to stop)")
server.serve_forever()