"""
def _lan_ip() -> str:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
sock.connect(("192.0.2.1", 1))
return sock.getsockname()[0]
except OSError:
return "127.0.0.1"
finally:
sock.close()
def _as_text(raw: object) -> str:
if raw is None:
return ""
if isinstance(raw, list):
return "\n".join(str(part) for part in raw)
return str(raw)
def pretty_json(raw: object) -> str:
text = _as_text(raw).strip() or "{}"
return json.dumps(json.loads(text), indent=2)
def parse_rid(value: object) -> int | None:
if value is None or value == "":
return None
try:
return int(float(str(value)))
except (TypeError, ValueError):
return None
def _table_rows(data: object) -> list[list[object]]:
if data is None:
return []
if hasattr(data, "values"):
return [list(row) for row in data.values]
return [list(row) for row in data]
def _image_preview(path: str) -> str | None:
if not path:
return None
suffix = Path(path).suffix.lower()
if suffix in {".jpg", ".jpeg", ".png", ".webp", ".tif", ".tiff"}:
return path
return None
def _save_upload(settings: Settings, name: str, data: bytes) -> Path:
safe = Path(name).name or "upload.bin"
dest = settings.inbox_dir / safe
n = 1
while dest.exists():
dest = settings.inbox_dir / f"{dest.stem}-{n}{dest.suffix}"
n += 1
dest.write_bytes(data)
return dest
_JOBS: dict[str, dict[str, object]] = {}
_JOBS_LOCK = threading.Lock()
def build_app(settings: Settings) -> FastAPI:
settings.ensure_dirs()
api = FastAPI(title="keys-automatic-receipt-doc-scanner")
@api.post("/api/inbox")
async def api_inbox(file: UploadFile = File(...)) -> JSONResponse:
data = await file.read()
dest = _save_upload(settings, file.filename or "iphone.jpg", data)
job_id = uuid.uuid4().hex
with _JOBS_LOCK:
_JOBS[job_id] = {"state": "processing"}
def _run() -> None:
try:
result = process_file(dest, settings)
extract = result.extract
payload: dict[str, object] = {
"state": "done",
"status": result.status.value,
"receipt_id": result.receipt_id,
"vendor": extract.vendor if extract else None,
"total": str(extract.total) if extract and extract.total is not None else None,
"category": extract.category if extract else None,
"error": result.error,
}
except Exception as exc:
payload = {"state": "error", "error": str(exc)}
with _JOBS_LOCK:
_JOBS[job_id] = payload
threading.Thread(target=_run, daemon=True, name="inbox-scan").start()
return JSONResponse(
{
"ok": True,
"job_id": job_id,
"path": str(dest),
"bytes": len(data),
"processing": True,
}
)
@api.get("/api/health")
async def api_health() -> JSONResponse:
from app.route import resolve_brain
try:
brain = resolve_brain(settings)
payload = {
"ok": True,
"route": settings.llm_route,
"kind": brain.kind,
"base_url": brain.base_url,
"model": brain.model,
"source": brain.source,
}
except Exception as exc:
payload = {"ok": False, "error": str(exc), "route": settings.llm_route}
return JSONResponse(payload)
@api.get("/api/jobs/{job_id}")
async def job_status(job_id: str) -> JSONResponse:
with _JOBS_LOCK:
job = _JOBS.get(job_id)
if job is None:
return JSONResponse({"state": "error", "error": "unknown job"}, status_code=404)
return JSONResponse(job)
@api.get("/phone", response_class=HTMLResponse)
async def phone() -> str:
return PHONE_HTML
def health() -> str:
llm = build_llm(settings)
embed = build_embed(settings)
return (
f"LLM {settings.llm_backend}/{settings.llm_model} @ {settings.llm_base_url} "
f"vision={llm.accepts_images} health={llm.health()}\n"
f"Embed {settings.embed_backend}/{settings.embed_model} dim={settings.embed_dim} "
f"health={embed.health()}\n"
f"Camera {settings.camera_url} idle={settings.idle_seconds}s"
)
def inbox_list() -> str:
files = sorted(p.name for p in settings.inbox_dir.iterdir() if p.is_file())
return "\n".join(files) or "(empty inbox)"
def ingest(files: list) -> str:
if not files:
return "no files"
names = []
for item in files:
src = Path(item if isinstance(item, str) else item.name)
dest = settings.inbox_dir / src.name
shutil.copy2(src, dest)
names.append(dest.name)
return "queued: " + ", ".join(names)
def review_table() -> list[list[str]]:
con = open_db(settings)
try:
rows = list_receipts(con, limit=40)
return [
[
str(r["id"]),
r["status"],
r["doc_kind"] or "",
r["category"] or "",
r["vendor"] or "",
r["receipt_date"] or "",
"" if r["total_cents"] is None else f"{r['total_cents']/100:.2f}",
]
for r in rows
]
finally:
con.close()
def _empty_load() -> tuple:
return (
None,
"",
"receipt",
"other",
"",
"",
"",
"",
"",
"{}",
[],
"not found",
)
def load_one(receipt_id: object) -> tuple:
rid = parse_rid(receipt_id)
if rid is None:
return _empty_load()
con = open_db(settings)
try:
row = get_receipt(con, rid)
if row is None:
return _empty_load()
try:
payload = pretty_json(row["extract_json"] or "{}")
except json.JSONDecodeError:
payload = row["extract_json"] or "{}"
extract = None
try:
extract = ReceiptExtract.model_validate_json(payload)
except Exception:
extract = None
lines = [
[
item.description,
"" if item.qty is None else item.qty,
"" if item.unit_price is None else str(item.unit_price),
"" if item.amount is None else str(item.amount),
item.sku or "",
]
for item in (extract.line_items if extract else [])
]
if not lines:
lines = [
[
r["description"],
r["qty"] if r["qty"] is not None else "",
"" if r["unit_price_cents"] is None else f"{r['unit_price_cents']/100:.2f}",
"" if r["amount_cents"] is None else f"{r['amount_cents']/100:.2f}",
r["sku"] or "",
]
for r in list_line_items(con, rid)
]
return (
_image_preview(row["source_path"] or ""),
str(rid),
(extract.doc_kind if extract else row["doc_kind"]) or "receipt",
(extract.category if extract else row["category"]) or "other",
(extract.vendor if extract else row["vendor"]) or "",
(
extract.date.isoformat()
if extract and extract.date
else (row["receipt_date"] or "")
),
"" if extract is None or extract.tax is None else str(extract.tax),
"" if extract is None or extract.total is None else str(extract.total),
(extract.currency if extract else row["currency"]) or "",
payload,
lines,
f"loaded #{rid}",
)
finally:
con.close()
def load_from_table(data: object, evt: gr.SelectData) -> tuple:
rows = _table_rows(data)
index = evt.index
row_i = index[0] if isinstance(index, (list, tuple)) else index
if row_i is None or row_i < 0 or row_i >= len(rows) or not rows[row_i]:
return _empty_load()
return load_one(rows[row_i][0])
def save_extract(receipt_id: object, raw: object) -> tuple[str, list[list[str]]]:
rid = parse_rid(receipt_id)
if rid is None:
return "pick a receipt (click a row or enter id)", review_table()
try:
extract = ReceiptExtract.model_validate_json(pretty_json(raw))
except Exception as exc:
return f"invalid JSON: {exc}", review_table()
con = open_db(settings)
try:
update_extract(con, rid, extract, ReceiptStatus.needs_review)
finally:
con.close()
return f"saved JSON for #{rid}", review_table()
def save_fields(
receipt_id: object,
doc_kind: str,
category: str,
vendor: str,
receipt_date: str,
tax: str,
total: str,
currency: str,
lines: object,
) -> tuple[str, str, list[list[str]]]:
rid = parse_rid(receipt_id)
if rid is None:
return "pick a receipt first", "{}", review_table()
def _num(val: object) -> str | None:
text = str(val).strip()
return None if text in {"", "None", "null"} else text
items = []
for row in _table_rows(lines):
if not row or not str(row[0]).strip():
continue
items.append(
{
"description": str(row[0]).strip(),
"qty": _num(row[1] if len(row) > 1 else None),
"unit_price": _num(row[2] if len(row) > 2 else None),
"amount": _num(row[3] if len(row) > 3 else None),
"sku": (str(row[4]).strip() or None) if len(row) > 4 else None,
}
)
payload = {
"doc_kind": doc_kind or "receipt",
"category": category or "other",
"vendor": vendor.strip() or None,
"date": receipt_date.strip() or None,
"tax": tax.strip() or None,
"total": total.strip() or None,
"currency": currency.strip() or None,
"line_items": items,
}
raw = json.dumps(payload, indent=2)
try:
extract = ReceiptExtract.model_validate(payload)
except Exception as exc:
return f"invalid fields: {exc}", raw, review_table()
con = open_db(settings)
try:
update_extract(con, rid, extract, ReceiptStatus.needs_review)
finally:
con.close()
return f"saved receipt #{rid}", pretty_json(extract.model_dump_json()), review_table()
def confirm(receipt_id: object) -> tuple[str, list[list[str]]]:
rid = parse_rid(receipt_id)
if rid is None:
return "pick a receipt first", review_table()
con = open_db(settings)
try:
update_receipt_status(con, rid, ReceiptStatus.confirmed)
finally:
con.close()
return f"confirmed #{rid}", review_table()
def delete_one(receipt_id: object) -> tuple:
rid = parse_rid(receipt_id)
empty = _empty_load()
if rid is None:
return (*empty[:-1], "pick a receipt first", review_table())
con = open_db(settings)
try:
found = delete_receipt(con, rid)
finally:
con.close()
if not found:
return (*empty[:-1], f"not found #{rid}", review_table())
return (*empty[:-1], f"deleted #{rid}", review_table())
def catalog_table() -> list[list[str]]:
con = open_db(settings)
try:
return [
[str(r["id"]), r["sku"] or "", r["vendor"] or "", r["description"]]
for r in list_catalog(con)
]
finally:
con.close()
def add_sku(sku: str, vendor: str, description: str) -> str:
if not description.strip():
return "description required"
con = open_db(settings)
try:
add_catalog_item(con, sku=sku or None, vendor=vendor or None, description=description)
finally:
con.close()
return "added"
def process_now(path: str) -> str:
if not path:
return "no path"
result = process_file(Path(path), settings)
return result.model_dump_json(indent=2)
with gr.Blocks(title="Receipt Studio", theme=gr.themes.Soft(primary_hue="amber"), css=CSS) as demo:
gr.Markdown(
"# Receipt Studio\n"
"Lamp camera · phone upload · Gemma 4 12B Unified on the GPU box (not on the Lamp)."
)
with gr.Tab("Inbox"):
gr.Markdown(
f"Drop files here or on your phone: `http://{_lan_ip()}:{settings.ui_port}/phone` "
f"(bind `0.0.0.0` / `RECEIPT_UI_SHARE_LAN=true`). Syncthing can also land in `inbox/`."
)
files = gr.File(label="Photos / PDFs", file_count="multiple", type="filepath")
ingest_btn = gr.Button("Queue in inbox")
ingest_out = gr.Textbox(label="Queued")
listing = gr.Textbox(label="Inbox", lines=8)
refresh = gr.Button("Refresh inbox")
ingest_btn.click(ingest, inputs=[files], outputs=[ingest_out]).then(
inbox_list, outputs=[listing]
)
refresh.click(inbox_list, outputs=[listing])
demo.load(inbox_list, outputs=[listing])
with gr.Tab("Review"):
gr.Markdown(
"Click a row to load. Edit **Kind / Category / Vendor / Date / Tax / Total** "
"(or the JSON), then **Save fields + lines**. "
"**Delete** removes a bad scan from the database and disk."
)
table = gr.Dataframe(
headers=["id", "status", "kind", "category", "vendor", "date", "total"],
datatype=["str"] * 7,
interactive=False,
wrap=True,
)
refresh_r = gr.Button("Refresh queue")
with gr.Row():
img = gr.Image(label="Scan", type="filepath")
with gr.Column():
rid = gr.Textbox(label="Receipt id")
kind = gr.Dropdown(choices=list(DOC_KINDS), label="Kind", value="receipt")
category = gr.Dropdown(choices=list(CATEGORIES), label="Category", value="other")
vendor = gr.Textbox(label="Vendor")
receipt_date = gr.Textbox(label="Date (YYYY-MM-DD)")
tax = gr.Textbox(label="Tax")
total = gr.Textbox(label="Total")
currency = gr.Textbox(label="Currency")
raw = gr.Textbox(
label="Extract JSON (editable)",
lines=18,
max_lines=40,
interactive=True,
)
lines = gr.Dataframe(
headers=["description", "qty", "unit_price", "amount", "sku"],
datatype=["str", "str", "str", "str", "str"],
label="Line items (editable)",
interactive=True,
wrap=True,
)
with gr.Row():
load_btn = gr.Button("Load id")
save_fields_btn = gr.Button("Save fields + lines")
save_json_btn = gr.Button("Save JSON")
ok_btn = gr.Button("Confirm")
del_btn = gr.Button("Delete", variant="stop")
msg = gr.Textbox(label="Status")
load_outputs = [
img,
rid,
kind,
category,
vendor,
receipt_date,
tax,
total,
currency,
raw,
lines,
msg,
]
refresh_r.click(review_table, outputs=[table])
table.select(load_from_table, inputs=[table], outputs=load_outputs)
load_btn.click(load_one, inputs=[rid], outputs=load_outputs)
save_fields_btn.click(
save_fields,
inputs=[rid, kind, category, vendor, receipt_date, tax, total, currency, lines],
outputs=[msg, raw, table],
)
save_json_btn.click(save_extract, inputs=[rid, raw], outputs=[msg, table])
ok_btn.click(confirm, inputs=[rid], outputs=[msg, table])
del_btn.click(delete_one, inputs=[rid], outputs=load_outputs + [table])
demo.load(review_table, outputs=[table])
with gr.Tab("Catalog"):
cat = gr.Dataframe(headers=["id", "sku", "vendor", "description"])
sku = gr.Textbox(label="SKU")
vendor = gr.Textbox(label="Vendor")
desc = gr.Textbox(label="Description")
add_btn = gr.Button("Add SKU")
add_msg = gr.Textbox()
add_btn.click(add_sku, inputs=[sku, vendor, desc], outputs=[add_msg]).then(
catalog_table, outputs=[cat]
)
demo.load(catalog_table, outputs=[cat])
with gr.Tab("Settings"):
gr.Markdown(
"Gemma 4 12B Unified does **not** load on the Lamp (6 GB). "
"This UI talks to the GPU box URLs in `.env`."
)
gr.Textbox(value=health, label="Backends", every=15)
path = gr.Textbox(label="Process this path now")
run = gr.Button("Process")
run_out = gr.Textbox(lines=16)
run.click(process_now, inputs=[path], outputs=[run_out])
return gr.mount_gradio_app(api, demo, path="/")
def main() -> None:
settings = load_settings()
if settings.ui_share_lan:
settings.ui_host = "0.0.0.0"
start_inbox_watcher(settings)
import uvicorn
uvicorn.run(
build_app(settings),
host=settings.ui_host,
port=settings.ui_port,
log_level="info",
)
if __name__ == "__main__":
main()