220 lines
8.7 KiB
Python
220 lines
8.7 KiB
Python
"""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
|
|
from pricing import PRICING_AS_OF, PRICING_SOURCE, cost_for_call
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
API = "/api" # same as API_BASE in index.html
|
|
REASON = {"green": None, "amber": "Slightly unclear, verify it", "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 = {}, {}
|
|
ROI_RUNS = [] # [{job_id, document, created_at, model, input_tokens, output_tokens, thinking_tokens}]
|
|
_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)
|
|
ROI_RUNS.clear()
|
|
now = time.time()
|
|
for i, (model, tokens) in enumerate([("gemini-2.5-flash", (21_000, 1_800, 900)),
|
|
("gemini-2.5-flash", (18_500, 1_600, 0)),
|
|
("gemini-2.5-pro", (240_000, 3_000, 2_500)),
|
|
("gemini-2.5-flash-lite", (22_000, 1_700, 0)),
|
|
("gemini-2.5-flash", (None, None, None))]): # usage not recorded
|
|
ROI_RUNS.append({"job_id": f"job-{900 + i}", "document": f"invoice-{900 + i}.pdf",
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - i * 12 * 864e2)),
|
|
"model": model, **dict(zip(("input_tokens", "output_tokens", "thinking_tokens"), tokens))})
|
|
|
|
|
|
reset() # sample data from import on: the desktop bridge imports this module without a server
|
|
|
|
|
|
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 roi(body):
|
|
"""Cost per run. total_cost is null when the run recorded no token usage: unpriced, never zero.
|
|
An unknown model is an error (500 naming the known models), never a cost of zero."""
|
|
try:
|
|
runs = [{**r, "total_cost": None if r["input_tokens"] is None else
|
|
cost_for_call(r["model"], r["input_tokens"], r["output_tokens"], r["thinking_tokens"] or 0)}
|
|
for r in ROI_RUNS]
|
|
except ValueError as e:
|
|
return 500, {"detail": str(e)}
|
|
return 200, {"runs": runs, "pricing": {"source": PRICING_SOURCE, "as_of": PRICING_AS_OF}}
|
|
|
|
|
|
def document(job_id, body):
|
|
"""A stand-in for the source document. A real backend returns the PDF or page image."""
|
|
lines = "".join(f'<text x="48" y="{150 + i * 46}" font-size="20">{escape(f)}: {escape(v)}</text>'
|
|
for i, (f, v, *_) in enumerate(SAMPLE_FIELDS))
|
|
svg = (f'<svg xmlns="http://www.w3.org/2000/svg" width="640" height="820" font-family="Georgia,serif">'
|
|
f'<rect width="100%" height="100%" fill="#fff"/><text x="48" y="80" font-size="28">'
|
|
f'{escape(RUNS[job_id]["document"])} (sample)</text>{lines}</svg>')
|
|
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),
|
|
("GET", r"/roi", roi),
|
|
("POST", r"/set-dark", lambda body: (200, {})),
|
|
]
|
|
|
|
|
|
def dispatch(method, path, body):
|
|
"""One API request -> (status, payload). The web handler and the desktop bridge both call this."""
|
|
for m, pattern, fn in ROUTES:
|
|
match = re.fullmatch(pattern, path)
|
|
if m == method and match:
|
|
try:
|
|
return fn(*match.groups(), body)
|
|
except (KeyError, StopIteration):
|
|
return 404, {"detail": "Not found"}
|
|
return 404, {"detail": f"No route for {method} {path}"}
|
|
|
|
|
|
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 = dispatch(method, path, body)
|
|
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()
|