| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import shutil |
| import sys |
| from pathlib import Path |
|
|
| from app.camera import aim, snapshot |
| from app.config import load_settings |
| from app.db import get_receipt, list_line_items, list_receipts, open_db |
| from app.pipeline import process_file |
| from app.schemas import ProcessResult |
|
|
|
|
| def _print_result(result: ProcessResult) -> None: |
| payload = result.model_dump(mode="json") |
| json.dump(payload, sys.stdout, indent=2) |
| sys.stdout.write("\n") |
|
|
|
|
| def cmd_scan(args: argparse.Namespace) -> int: |
| settings = load_settings() |
| if getattr(args, "aim_down", False) and not args.image: |
| aim(settings, "down") |
| if args.image: |
| image = Path(args.image) |
| else: |
| image = snapshot(settings) |
| if args.inbox: |
| dest = settings.inbox_dir / image.name |
| shutil.copy2(image, dest) |
| image = dest |
| result = process_file(image, settings) |
| _print_result(result) |
| return 0 if result.status.value not in {"failed"} else 1 |
|
|
|
|
| def cmd_query(args: argparse.Namespace) -> int: |
| settings = load_settings() |
| con = open_db(settings) |
| try: |
| rows = list_receipts(con, status=args.status, limit=args.limit) |
| out = [] |
| for row in rows: |
| item = dict(row) |
| if args.category and item.get("category") != args.category: |
| continue |
| if args.vendor and (item.get("vendor") or "").lower().find(args.vendor.lower()) < 0: |
| continue |
| item["line_items"] = [dict(x) for x in list_line_items(con, int(row["id"]))] |
| out.append(item) |
| json.dump(out, sys.stdout, indent=2, default=str) |
| sys.stdout.write("\n") |
| finally: |
| con.close() |
| return 0 |
|
|
|
|
| def cmd_show(args: argparse.Namespace) -> int: |
| settings = load_settings() |
| con = open_db(settings) |
| try: |
| row = get_receipt(con, args.id) |
| if row is None: |
| print(f"not found: {args.id}", file=sys.stderr) |
| return 1 |
| payload = dict(row) |
| payload["line_items"] = [dict(x) for x in list_line_items(con, args.id)] |
| json.dump(payload, sys.stdout, indent=2, default=str) |
| sys.stdout.write("\n") |
| finally: |
| con.close() |
| return 0 |
|
|
|
|
| def cmd_snapshot(_args: argparse.Namespace) -> int: |
| settings = load_settings() |
| path = snapshot(settings) |
| print(path) |
| return 0 |
|
|
|
|
| def cmd_ui(_args: argparse.Namespace) -> int: |
| from app.ui import main |
|
|
| main() |
| return 0 |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser(prog="keys-scan") |
| sub = parser.add_subparsers(dest="cmd", required=True) |
|
|
| scan = sub.add_parser("scan", help="process one image (Lamp camera if omitted)") |
| scan.add_argument("--image", help="path to jpeg/png/pdf; omit to snapshot Lamp camera") |
| scan.add_argument("--inbox", action="store_true", help="copy into inbox/ first") |
| scan.add_argument("--aim-down", action="store_true", help="HAL servo down before snapshot") |
| scan.set_defaults(func=cmd_scan) |
|
|
| query = sub.add_parser("query", help="list stored receipts") |
| query.add_argument("--status") |
| query.add_argument("--category") |
| query.add_argument("--vendor") |
| query.add_argument("--limit", type=int, default=50) |
| query.set_defaults(func=cmd_query) |
|
|
| show = sub.add_parser("show", help="show one receipt") |
| show.add_argument("id", type=int) |
| show.set_defaults(func=cmd_show) |
|
|
| snap = sub.add_parser("snapshot", help="HAL camera snapshot only") |
| snap.set_defaults(func=cmd_snapshot) |
|
|
| ui = sub.add_parser("ui", help="Gradio review UI") |
| ui.set_defaults(func=cmd_ui) |
| return parser |
|
|
|
|
| def main(argv: list[str] | None = None) -> None: |
| parser = build_parser() |
| args = parser.parse_args(argv) |
| raise SystemExit(args.func(args)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|