dodox
d97aac0e17
Backend: receipt upload -> LLM vision extraction (OpenAI-compatible, provider-agnostic), item review/edit, per-group splitting against Cospend projects/members, highlight+upload via WebDAV, public share link, bill creation via Cospend's OCS API (verified against real source, not just doc summaries). Frontend: capture -> review -> group -> summary flow as an installable PWA. install.sh / run.sh (venv + npm, tmux session) instead of Docker, per the ~/Projects/gain pattern.
292 lines
10 KiB
Python
292 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import uuid
|
|
from datetime import date as date_cls
|
|
from datetime import datetime
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
|
|
from . import cospend_client, highlight, llm_client, nc_client
|
|
from .config import Config
|
|
from .db import get_conn
|
|
|
|
bp = Blueprint("api", __name__)
|
|
|
|
|
|
# ---------------------------------------------------------------- receipts
|
|
|
|
@bp.post("/receipts")
|
|
def create_receipt():
|
|
"""multipart/form-data with an `image` file. Extracts items via LLM."""
|
|
file = request.files.get("image")
|
|
if file is None:
|
|
return jsonify(error="missing 'image' file"), 400
|
|
|
|
image_bytes = file.read()
|
|
receipt_id = str(uuid.uuid4())
|
|
image_path = f"{Config.UPLOAD_DIR}/{receipt_id}.jpg"
|
|
with open(image_path, "wb") as f:
|
|
f.write(image_bytes)
|
|
|
|
try:
|
|
extracted = llm_client.extract_receipt(image_bytes, mime_type=file.mimetype or "image/jpeg")
|
|
except Exception as exc: # noqa: BLE001 - surfaced to the client as-is
|
|
return jsonify(error=f"extraction failed: {exc}"), 502
|
|
|
|
items = extracted["items"]
|
|
store_name = extracted["store_name"]
|
|
# Fall back to today if the receipt's date wasn't readable - the user
|
|
# can still override it in the review step.
|
|
receipt_date = extracted["date"] or date_cls.today().isoformat()
|
|
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO receipts (id, image_path, items_json, store_name, receipt_date)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(receipt_id, image_path, json.dumps(items), store_name, receipt_date),
|
|
)
|
|
|
|
return jsonify(id=receipt_id, items=items, store_name=store_name, date=receipt_date), 201
|
|
|
|
|
|
@bp.get("/receipts/<receipt_id>")
|
|
def get_receipt(receipt_id: str):
|
|
receipt = _load_receipt(receipt_id)
|
|
if receipt is None:
|
|
return jsonify(error="not found"), 404
|
|
return jsonify(receipt)
|
|
|
|
|
|
@bp.patch("/receipts/<receipt_id>/items")
|
|
def update_items(receipt_id: str):
|
|
"""Body: {"items": [...]} — full replacement, for the review/edit step."""
|
|
body = request.get_json(force=True) or {}
|
|
items = body.get("items")
|
|
if not isinstance(items, list):
|
|
return jsonify(error="'items' must be a list"), 400
|
|
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"UPDATE receipts SET items_json = ? WHERE id = ?",
|
|
(json.dumps(items), receipt_id),
|
|
)
|
|
if cur.rowcount == 0:
|
|
return jsonify(error="not found"), 404
|
|
|
|
return jsonify(id=receipt_id, items=items)
|
|
|
|
|
|
@bp.patch("/receipts/<receipt_id>/meta")
|
|
def update_meta(receipt_id: str):
|
|
"""Body: {"store_name"?: str, "date"?: "YYYY-MM-DD"} - user override for
|
|
what the receipt reader guessed (or didn't find)."""
|
|
body = request.get_json(force=True) or {}
|
|
fields, values = [], []
|
|
if "store_name" in body:
|
|
fields.append("store_name = ?")
|
|
values.append(body["store_name"])
|
|
if "date" in body:
|
|
if not re.match(r"^\d{4}-\d{2}-\d{2}$", str(body["date"] or "")):
|
|
return jsonify(error="date must be YYYY-MM-DD"), 400
|
|
fields.append("receipt_date = ?")
|
|
values.append(body["date"])
|
|
if not fields:
|
|
return jsonify(error="nothing to update"), 400
|
|
|
|
values.append(receipt_id)
|
|
with get_conn() as conn:
|
|
cur = conn.execute(f"UPDATE receipts SET {', '.join(fields)} WHERE id = ?", values)
|
|
if cur.rowcount == 0:
|
|
return jsonify(error="not found"), 404
|
|
|
|
receipt = _load_receipt(receipt_id)
|
|
return jsonify(receipt)
|
|
|
|
|
|
def _load_receipt(receipt_id: str) -> dict | None:
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT * FROM receipts WHERE id = ?", (receipt_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
return {
|
|
"id": row["id"],
|
|
"items": json.loads(row["items_json"]),
|
|
"store_name": row["store_name"],
|
|
"date": row["receipt_date"],
|
|
"status": row["status"],
|
|
"created_at": row["created_at"],
|
|
}
|
|
|
|
|
|
# -------------------------------------------------------------- cospend
|
|
|
|
@bp.get("/cospend/projects")
|
|
def cospend_projects():
|
|
try:
|
|
projects = cospend_client.get_projects()
|
|
except Exception as exc: # noqa: BLE001
|
|
return jsonify(error=f"cospend request failed: {exc}"), 502
|
|
return jsonify(projects=projects, default_project_id=Config.COSPEND_DEFAULT_PROJECT_ID or None)
|
|
|
|
|
|
@bp.get("/cospend/projects/<project_id>/members")
|
|
def cospend_members(project_id: str):
|
|
try:
|
|
members = cospend_client.get_members(project_id)
|
|
except Exception as exc: # noqa: BLE001
|
|
return jsonify(error=f"cospend request failed: {exc}"), 502
|
|
return jsonify(members)
|
|
|
|
|
|
# ------------------------------------------------------------------ groups
|
|
|
|
@bp.post("/receipts/<receipt_id>/groups")
|
|
def create_group(receipt_id: str):
|
|
"""Body: {name, cospend_project_id, payer_member_id, member_ids: [...], item_ids: [...]}"""
|
|
receipt = _load_receipt(receipt_id)
|
|
if receipt is None:
|
|
return jsonify(error="not found"), 404
|
|
|
|
body = request.get_json(force=True) or {}
|
|
name = body.get("name") or "WG"
|
|
project_id = body.get("cospend_project_id")
|
|
payer_member_id = body.get("payer_member_id")
|
|
member_ids = body.get("member_ids") or []
|
|
item_ids = body.get("item_ids") or []
|
|
|
|
if not project_id or not payer_member_id or not member_ids or not item_ids:
|
|
return (
|
|
jsonify(
|
|
error="cospend_project_id, payer_member_id, member_ids and item_ids are required"
|
|
),
|
|
400,
|
|
)
|
|
|
|
group_id = str(uuid.uuid4())
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO groups
|
|
(id, receipt_id, name, cospend_project_id, payer_member_id,
|
|
member_ids_json, item_ids_json)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
group_id,
|
|
receipt_id,
|
|
name,
|
|
project_id,
|
|
payer_member_id,
|
|
json.dumps(member_ids),
|
|
json.dumps(item_ids),
|
|
),
|
|
)
|
|
conn.execute("UPDATE receipts SET status = 'grouped' WHERE id = ?", (receipt_id,))
|
|
|
|
return jsonify(id=group_id), 201
|
|
|
|
|
|
@bp.get("/receipts/<receipt_id>/groups")
|
|
def list_groups(receipt_id: str):
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM groups WHERE receipt_id = ? ORDER BY created_at", (receipt_id,)
|
|
).fetchall()
|
|
return jsonify([_group_dict(r) for r in rows])
|
|
|
|
|
|
def _group_dict(row) -> dict:
|
|
return {
|
|
"id": row["id"],
|
|
"receipt_id": row["receipt_id"],
|
|
"name": row["name"],
|
|
"cospend_project_id": row["cospend_project_id"],
|
|
"payer_member_id": row["payer_member_id"],
|
|
"member_ids": json.loads(row["member_ids_json"]),
|
|
"item_ids": json.loads(row["item_ids_json"]),
|
|
"status": row["status"],
|
|
"share_url": row["share_url"],
|
|
"cospend_bill_id": row["cospend_bill_id"],
|
|
"error": row["error"],
|
|
}
|
|
|
|
|
|
@bp.post("/receipts/<receipt_id>/groups/<group_id>/submit")
|
|
def submit_group(receipt_id: str, group_id: str):
|
|
"""Highlights selected items, uploads to NC, shares, creates the Cospend bill."""
|
|
receipt = _load_receipt(receipt_id)
|
|
if receipt is None:
|
|
return jsonify(error="receipt not found"), 404
|
|
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM groups WHERE id = ? AND receipt_id = ?", (group_id, receipt_id)
|
|
).fetchone()
|
|
if row is None:
|
|
return jsonify(error="group not found"), 404
|
|
group = _group_dict(row)
|
|
|
|
items_by_id = {item["id"]: item for item in receipt["items"]}
|
|
selected = [items_by_id[i] for i in group["item_ids"] if i in items_by_id]
|
|
if not selected:
|
|
return jsonify(error="no matching items on this receipt"), 400
|
|
|
|
total = round(sum(item["price"] for item in selected), 2)
|
|
bboxes = [item["bbox"] for item in selected if item.get("bbox")]
|
|
|
|
with open(_receipt_image_path(receipt_id), "rb") as f:
|
|
original_bytes = f.read()
|
|
|
|
try:
|
|
highlighted = highlight.highlight_items(original_bytes, bboxes)
|
|
|
|
# Timestamp for human sorting/browsing in NC, group id suffix so two
|
|
# submits in the same second (or a retry) never collide/overwrite.
|
|
safe_name = re.sub(r"[^A-Za-z0-9_-]+", "-", group["name"]).strip("-") or "group"
|
|
filename = f"{datetime.now():%Y-%m-%d_%H%M%S}_{safe_name}_{group_id[:8]}.jpg"
|
|
rel_path = nc_client.upload_file(filename, highlighted)
|
|
share_url = nc_client.create_public_share(rel_path)
|
|
|
|
item_lines = "\n".join(f"- {i['label']}: {i['price']}" for i in selected)
|
|
|
|
# Descriptive title: store name when we have one, else just the
|
|
# group name - either way followed by the share link, since
|
|
# Cospend's UI only recognizes a bill as having an attachment when
|
|
# the link is in `what` (the title); a link in `comment` is just
|
|
# plain text and isn't picked up.
|
|
store_name = receipt.get("store_name")
|
|
label = f"{store_name} ({group['name']})" if store_name else group["name"]
|
|
bill_date = receipt.get("date") or date_cls.today().isoformat()
|
|
|
|
bill_id = cospend_client.create_bill(
|
|
project_id=group["cospend_project_id"],
|
|
what=f"{label} {share_url}",
|
|
amount=total,
|
|
payer_id=int(group["payer_member_id"]),
|
|
ower_ids=[int(m) for m in group["member_ids"]],
|
|
date=bill_date,
|
|
comment=item_lines,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"UPDATE groups SET status = 'failed', error = ? WHERE id = ?",
|
|
(str(exc), group_id),
|
|
)
|
|
return jsonify(error=f"submit failed: {exc}"), 502
|
|
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""UPDATE groups SET status = 'submitted', share_url = ?, cospend_bill_id = ?
|
|
WHERE id = ?""",
|
|
(share_url, str(bill_id), group_id),
|
|
)
|
|
|
|
return jsonify(share_url=share_url, bill_id=bill_id, amount=total)
|
|
|
|
|
|
def _receipt_image_path(receipt_id: str) -> str:
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT image_path FROM receipts WHERE id = ?", (receipt_id,)).fetchone()
|
|
return row["image_path"]
|