bonamin
8ba8c95ecd
Includes all work to date: - local_backend: FastAPI backend with products, orders, tables, shifts, cloud sync - manager_dashboard: React manager UI with product/category management, reports, settings - waiter_pwa: React PWA for waiter devices - Category reparent endpoint and UI - Waiter domain: local_ip sent on heartbeat, waiter_domain persisted from cloud response - QR code modal in AppInfoTab for waiter domain - Product form: number input spinners removed, category pre-selected on new product - Category row: count badge moved to far right Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
343 lines
14 KiB
Python
343 lines
14 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from models.product import (
|
|
Category, Product, ProductOption, ProductQuickOption,
|
|
ProductIngredient, ProductPreferenceSet, ProductPreferenceChoice,
|
|
)
|
|
from models.table import Table, TableGroup
|
|
from models.user import User
|
|
from routers.deps import require_manager
|
|
|
|
router = APIRouter()
|
|
|
|
EXPORT_VERSION = 1
|
|
|
|
|
|
def _serialize_product(p: Product) -> dict:
|
|
quick_options = [
|
|
{"name": qo.name, "price": qo.price, "allow_multiple": qo.allow_multiple,
|
|
"sort_order": qo.sort_order, "is_favorite": qo.is_favorite,
|
|
"favorite_sort_order": qo.favorite_sort_order, "is_compact": qo.is_compact}
|
|
for qo in p.quick_options
|
|
]
|
|
options = []
|
|
for opt in p.options:
|
|
sub = json.loads(opt.sub_choices) if opt.sub_choices else []
|
|
options.append({"name": opt.name, "extra_cost": opt.extra_cost,
|
|
"allow_multiple": opt.allow_multiple, "sub_choices": sub,
|
|
"is_favorite": opt.is_favorite, "favorite_sort_order": opt.favorite_sort_order})
|
|
ingredients = [
|
|
{"name": ing.name, "extra_cost": ing.extra_cost,
|
|
"is_favorite": ing.is_favorite, "favorite_sort_order": ing.favorite_sort_order}
|
|
for ing in p.ingredients
|
|
]
|
|
preference_sets = []
|
|
for ps in p.preference_sets:
|
|
shared = json.loads(ps.shared_subset) if ps.shared_subset else None
|
|
default_index = None
|
|
choices = []
|
|
for i, ch in enumerate(ps.choices):
|
|
if ch.id == ps.default_choice_id:
|
|
default_index = i
|
|
sub = json.loads(ch.sub_choices) if ch.sub_choices else []
|
|
choices.append({"name": ch.name, "extra_cost": ch.extra_cost,
|
|
"sub_choices": sub, "disables_subset": ch.disables_subset})
|
|
preference_sets.append({
|
|
"name": ps.name, "choices": choices,
|
|
"default_choice_index": default_index, "shared_subset": shared,
|
|
"is_favorite": ps.is_favorite, "favorite_sort_order": ps.favorite_sort_order,
|
|
})
|
|
return {
|
|
"name": p.name, "base_price": p.base_price, "is_available": p.is_available,
|
|
"lifecycle_status": p.lifecycle_status, "sort_order": p.sort_order,
|
|
"printer_zone_id": None, # always stripped on export
|
|
"quick_options": quick_options, "options": options,
|
|
"ingredients": ingredients, "preference_sets": preference_sets,
|
|
}
|
|
|
|
|
|
def _serialize_category(cat: Category) -> dict:
|
|
products = [_serialize_product(p) for p in cat.products if p.lifecycle_status != "archived"]
|
|
return {
|
|
"name": cat.name, "color": cat.color, "sort_order": cat.sort_order,
|
|
"parent_name": cat.parent.name if cat.parent else None,
|
|
"general_sort_order": cat.general_sort_order, "auto_expanded": cat.auto_expanded,
|
|
"products": products,
|
|
}
|
|
|
|
|
|
@router.get("/export/catalog")
|
|
def export_catalog(db: Session = Depends(get_db), user: User = Depends(require_manager)):
|
|
categories = db.query(Category).order_by(Category.sort_order).all()
|
|
orphan_products = (
|
|
db.query(Product)
|
|
.filter(Product.category_id == None, Product.lifecycle_status != "archived")
|
|
.order_by(Product.sort_order)
|
|
.all()
|
|
)
|
|
data = {
|
|
"categories": [_serialize_category(c) for c in categories],
|
|
"uncategorized_products": [_serialize_product(p) for p in orphan_products],
|
|
}
|
|
payload = {
|
|
"xenia_export_version": EXPORT_VERSION,
|
|
"bundle": "catalog",
|
|
"exported_at": datetime.now(timezone.utc).isoformat(),
|
|
"data": data,
|
|
}
|
|
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
return JSONResponse(
|
|
content=payload,
|
|
headers={"Content-Disposition": f'attachment; filename="xenia-catalog-{today}.json"'},
|
|
)
|
|
|
|
|
|
@router.get("/export/tables")
|
|
def export_tables(db: Session = Depends(get_db), user: User = Depends(require_manager)):
|
|
groups = db.query(TableGroup).order_by(TableGroup.sort_order).all()
|
|
table_groups = []
|
|
for g in groups:
|
|
tables = [
|
|
{"number": t.number, "label": t.label, "is_active": t.is_active,
|
|
"floor_x": t.floor_x, "floor_y": t.floor_y}
|
|
for t in sorted(g.tables, key=lambda t: t.number)
|
|
]
|
|
table_groups.append({
|
|
"name": g.name, "prefix": g.prefix, "sort_order": g.sort_order,
|
|
"color": g.color, "tables": tables,
|
|
})
|
|
ungrouped = (
|
|
db.query(Table)
|
|
.filter(Table.group_id == None)
|
|
.order_by(Table.number)
|
|
.all()
|
|
)
|
|
ungrouped_tables = [
|
|
{"number": t.number, "label": t.label, "is_active": t.is_active,
|
|
"floor_x": t.floor_x, "floor_y": t.floor_y}
|
|
for t in ungrouped
|
|
]
|
|
payload = {
|
|
"xenia_export_version": EXPORT_VERSION,
|
|
"bundle": "tables",
|
|
"exported_at": datetime.now(timezone.utc).isoformat(),
|
|
"data": {"table_groups": table_groups, "ungrouped_tables": ungrouped_tables},
|
|
}
|
|
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
return JSONResponse(
|
|
content=payload,
|
|
headers={"Content-Disposition": f'attachment; filename="xenia-tables-{today}.json"'},
|
|
)
|
|
|
|
|
|
@router.post("/import/catalog")
|
|
def import_catalog(payload: dict, db: Session = Depends(get_db), user: User = Depends(require_manager)):
|
|
if payload.get("bundle") != "catalog":
|
|
raise HTTPException(status_code=400, detail="Wrong bundle type. Expected 'catalog'.")
|
|
if payload.get("xenia_export_version", 0) != EXPORT_VERSION:
|
|
raise HTTPException(status_code=400, detail="Unsupported export version.")
|
|
|
|
data = payload.get("data", {})
|
|
categories_data = data.get("categories", [])
|
|
|
|
def _upsert_category(cat_data: dict, parent_id=None) -> Category:
|
|
existing = db.query(Category).filter(
|
|
Category.name == cat_data["name"],
|
|
Category.parent_id == parent_id,
|
|
).first()
|
|
if existing:
|
|
existing.color = cat_data.get("color")
|
|
existing.sort_order = cat_data.get("sort_order", 0)
|
|
existing.general_sort_order = cat_data.get("general_sort_order", 0)
|
|
existing.auto_expanded = cat_data.get("auto_expanded", False)
|
|
db.flush()
|
|
return existing
|
|
else:
|
|
cat = Category(
|
|
name=cat_data["name"],
|
|
color=cat_data.get("color"),
|
|
sort_order=cat_data.get("sort_order", 0),
|
|
parent_id=parent_id,
|
|
general_sort_order=cat_data.get("general_sort_order", 0),
|
|
auto_expanded=cat_data.get("auto_expanded", False),
|
|
)
|
|
db.add(cat)
|
|
db.flush()
|
|
return cat
|
|
|
|
def _upsert_product(prod_data: dict, category_id=None):
|
|
existing = db.query(Product).filter(Product.name == prod_data["name"]).first()
|
|
if existing:
|
|
existing.base_price = prod_data["base_price"]
|
|
existing.is_available = prod_data.get("is_available", True)
|
|
existing.lifecycle_status = prod_data.get("lifecycle_status", "active")
|
|
existing.sort_order = prod_data.get("sort_order", 0)
|
|
existing.category_id = category_id
|
|
existing.printer_zone_id = None
|
|
db.flush()
|
|
product = existing
|
|
else:
|
|
product = Product(
|
|
name=prod_data["name"],
|
|
base_price=prod_data["base_price"],
|
|
is_available=prod_data.get("is_available", True),
|
|
lifecycle_status=prod_data.get("lifecycle_status", "active"),
|
|
sort_order=prod_data.get("sort_order", 0),
|
|
category_id=category_id,
|
|
printer_zone_id=None,
|
|
)
|
|
db.add(product)
|
|
db.flush()
|
|
|
|
# Replace sub-items (safe: sub-items have no direct order history references)
|
|
for qo in list(product.quick_options):
|
|
db.delete(qo)
|
|
for opt in list(product.options):
|
|
db.delete(opt)
|
|
for ing in list(product.ingredients):
|
|
db.delete(ing)
|
|
for ps in list(product.preference_sets):
|
|
db.delete(ps)
|
|
db.flush()
|
|
|
|
for qo in prod_data.get("quick_options", []):
|
|
db.add(ProductQuickOption(product_id=product.id, **qo))
|
|
|
|
for opt in prod_data.get("options", []):
|
|
sub_json = json.dumps(opt.get("sub_choices", []))
|
|
db.add(ProductOption(
|
|
product_id=product.id, name=opt["name"], extra_cost=opt.get("extra_cost", 0.0),
|
|
allow_multiple=opt.get("allow_multiple", False), sub_choices=sub_json,
|
|
is_favorite=opt.get("is_favorite", False),
|
|
favorite_sort_order=opt.get("favorite_sort_order", 0),
|
|
))
|
|
|
|
for ing in prod_data.get("ingredients", []):
|
|
db.add(ProductIngredient(product_id=product.id, **ing))
|
|
|
|
for ps_data in prod_data.get("preference_sets", []):
|
|
shared_json = json.dumps(ps_data["shared_subset"]) if ps_data.get("shared_subset") else None
|
|
ps = ProductPreferenceSet(
|
|
product_id=product.id, name=ps_data["name"],
|
|
shared_subset=shared_json,
|
|
is_favorite=ps_data.get("is_favorite", False),
|
|
favorite_sort_order=ps_data.get("favorite_sort_order", 0),
|
|
)
|
|
db.add(ps)
|
|
db.flush()
|
|
created_choices = []
|
|
for ch in ps_data.get("choices", []):
|
|
sub_json = json.dumps(ch.get("sub_choices", []))
|
|
choice = ProductPreferenceChoice(
|
|
set_id=ps.id, name=ch["name"], extra_cost=ch.get("extra_cost", 0.0),
|
|
sub_choices=sub_json, disables_subset=ch.get("disables_subset", False),
|
|
)
|
|
db.add(choice)
|
|
db.flush()
|
|
created_choices.append(choice)
|
|
idx = ps_data.get("default_choice_index")
|
|
if idx is not None and 0 <= idx < len(created_choices):
|
|
ps.default_choice_id = created_choices[idx].id
|
|
|
|
# First pass: top-level categories (parent_name is None)
|
|
for cat_data in categories_data:
|
|
if cat_data.get("parent_name") is None:
|
|
cat = _upsert_category(cat_data, parent_id=None)
|
|
for prod_data in cat_data.get("products", []):
|
|
_upsert_product(prod_data, category_id=cat.id)
|
|
|
|
# Second pass: sub-categories (parent must already exist from first pass)
|
|
for cat_data in categories_data:
|
|
if cat_data.get("parent_name") is not None:
|
|
parent = db.query(Category).filter(
|
|
Category.name == cat_data["parent_name"],
|
|
Category.parent_id == None,
|
|
).first()
|
|
parent_id = parent.id if parent else None
|
|
cat = _upsert_category(cat_data, parent_id=parent_id)
|
|
for prod_data in cat_data.get("products", []):
|
|
_upsert_product(prod_data, category_id=cat.id)
|
|
|
|
# Uncategorized products (no category)
|
|
for prod_data in data.get("uncategorized_products", []):
|
|
_upsert_product(prod_data, category_id=None)
|
|
|
|
db.commit()
|
|
return {"ok": True, "message": "Catalog imported successfully."}
|
|
|
|
|
|
@router.post("/import/tables")
|
|
def import_tables(payload: dict, db: Session = Depends(get_db), user: User = Depends(require_manager)):
|
|
if payload.get("bundle") != "tables":
|
|
raise HTTPException(status_code=400, detail="Wrong bundle type. Expected 'tables'.")
|
|
if payload.get("xenia_export_version", 0) != EXPORT_VERSION:
|
|
raise HTTPException(status_code=400, detail="Unsupported export version.")
|
|
|
|
data = payload.get("data", {})
|
|
for group_data in data.get("table_groups", []):
|
|
existing_group = db.query(TableGroup).filter(TableGroup.name == group_data["name"]).first()
|
|
if existing_group:
|
|
existing_group.prefix = group_data.get("prefix")
|
|
existing_group.sort_order = group_data.get("sort_order", 0)
|
|
existing_group.color = group_data.get("color")
|
|
db.flush()
|
|
group = existing_group
|
|
else:
|
|
group = TableGroup(
|
|
name=group_data["name"],
|
|
prefix=group_data.get("prefix"),
|
|
sort_order=group_data.get("sort_order", 0),
|
|
color=group_data.get("color"),
|
|
)
|
|
db.add(group)
|
|
db.flush()
|
|
|
|
for table_data in group_data.get("tables", []):
|
|
existing_table = db.query(Table).filter(
|
|
Table.number == table_data["number"],
|
|
Table.group_id == group.id,
|
|
).first()
|
|
if existing_table:
|
|
existing_table.label = table_data.get("label")
|
|
existing_table.is_active = table_data.get("is_active", True)
|
|
existing_table.floor_x = table_data.get("floor_x")
|
|
existing_table.floor_y = table_data.get("floor_y")
|
|
else:
|
|
db.add(Table(
|
|
number=table_data["number"],
|
|
label=table_data.get("label"),
|
|
group_id=group.id,
|
|
is_active=table_data.get("is_active", True),
|
|
floor_x=table_data.get("floor_x"),
|
|
floor_y=table_data.get("floor_y"),
|
|
))
|
|
|
|
# Ungrouped tables (no zone)
|
|
for table_data in data.get("ungrouped_tables", []):
|
|
existing_table = db.query(Table).filter(
|
|
Table.number == table_data["number"],
|
|
Table.group_id == None,
|
|
).first()
|
|
if existing_table:
|
|
existing_table.label = table_data.get("label")
|
|
existing_table.is_active = table_data.get("is_active", True)
|
|
existing_table.floor_x = table_data.get("floor_x")
|
|
existing_table.floor_y = table_data.get("floor_y")
|
|
else:
|
|
db.add(Table(
|
|
number=table_data["number"],
|
|
label=table_data.get("label"),
|
|
group_id=None,
|
|
is_active=table_data.get("is_active", True),
|
|
floor_x=table_data.get("floor_x"),
|
|
floor_y=table_data.get("floor_y"),
|
|
))
|
|
|
|
db.commit()
|
|
return {"ok": True, "message": "Tables imported successfully."}
|