bonamin/simple-pos-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backend overhaul: new models, routers, schemas for shifts, business day, flags, messages, settings

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Vasileios Kousaridas
2026-04-29 12:12:05 +03:00
parent 603fd45eaa
commit defc49f84f
31 changed files with 2626 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
local_backend/routers/reports.py
View File

@@ -12,6 +12,7 @@ from models.order import Order, OrderItem, OrderWaiter, PrintLog
from models.user import User
from models.table import Table
from models.printer import Printer
from models.shift import WaiterShift
from schemas.order import OrderOut
from schemas.table import TableOut
from routers.deps import require_manager
@@ -20,6 +21,12 @@ from services.printer_service import print_waiter_report, print_printer_report,
router = APIRouter()
def _dt(dt):
if dt is None:
return None
return (dt.isoformat() + "Z") if dt.tzinfo is None else dt.isoformat()
@router.get("/shift")
def shift_summary(
from_dt: Optional[str] = Query(default=None, alias="from"),
@@ -438,3 +445,213 @@ def print_printer_totals(
background_tasks.add_task(print_printer_report, printer.ip_address, printer.port, report, body.mode)
return {"status": "printing"}
# ---------------------------------------------------------------------------
# Shift history report
# ---------------------------------------------------------------------------
@router.get("/shifts")
def shifts_report(
waiter_id: Optional[int] = None,
business_day_id: Optional[int] = None,
from_dt: Optional[str] = Query(default=None, alias="from"),
to_dt: Optional[str] = Query(default=None, alias="to"),
active_only: bool = False,
db: Session = Depends(get_db),
user: User = Depends(require_manager),
):
from routers.shifts import compute_shift_total
q = db.query(WaiterShift)
if waiter_id:
q = q.filter(WaiterShift.waiter_id == waiter_id)
if business_day_id:
q = q.filter(WaiterShift.business_day_id == business_day_id)
if from_dt:
q = q.filter(WaiterShift.started_at >= datetime.fromisoformat(from_dt))
if to_dt:
q = q.filter(WaiterShift.started_at <= datetime.fromisoformat(to_dt))
if active_only:
q = q.filter(WaiterShift.ended_at == None)
shifts = q.order_by(WaiterShift.started_at.desc()).all()
waiters_db = {u.id: u for u in db.query(User).all()}
result = []
for shift in shifts:
w = waiters_db.get(shift.waiter_id)
wname = (w.full_name or w.username) if w else f"#{shift.waiter_id}"
total = compute_shift_total(shift.id, db) if shift.ended_at is None else (shift.total_collected or 0.0)
result.append({
"id": shift.id,
"waiter_id": shift.waiter_id,
"waiter_name": wname,
"business_day_id": shift.business_day_id,
"started_at": _dt(shift.started_at),
"ended_at": _dt(shift.ended_at),
"starting_cash": shift.starting_cash,
"total_collected": total,
"net_to_deliver": round(total + (shift.starting_cash or 0.0), 2),
"is_active": shift.ended_at is None,
"notes": shift.notes,
})
return {"shifts": result}
# ---------------------------------------------------------------------------
# Product performance analytics
# ---------------------------------------------------------------------------
@router.get("/products/performance")
def product_performance(
from_dt: Optional[str] = Query(default=None, alias="from"),
to_dt: Optional[str] = Query(default=None, alias="to"),
business_day_id: Optional[int] = None,
category_id: Optional[int] = None,
db: Session = Depends(get_db),
user: User = Depends(require_manager),
):
from models.product import Product
q = db.query(OrderItem).filter(OrderItem.status.in_(["active", "paid"]))
if from_dt:
q = q.filter(OrderItem.added_at >= datetime.fromisoformat(from_dt))
if to_dt:
q = q.filter(OrderItem.added_at <= datetime.fromisoformat(to_dt))
if business_day_id:
q = q.join(Order).filter(Order.business_day_id == business_day_id)
items = q.all()
products_db = {p.id: p for p in db.query(Product).all()}
summary: dict = {}
for item in items:
pid = item.product_id
product = products_db.get(pid)
if category_id and (not product or product.category_id != category_id):
continue
if pid not in summary:
summary[pid] = {
"product_id": pid,
"product_name": product.name if product else f"#{pid}",
"category_id": product.category_id if product else None,
"qty_sold": 0,
"revenue": 0.0,
"order_ids": set(),
}
summary[pid]["qty_sold"] += item.quantity
summary[pid]["revenue"] += item.unit_price * item.quantity
summary[pid]["order_ids"].add(item.order_id)
result = []
for entry in summary.values():
entry["order_count"] = len(entry.pop("order_ids"))
entry["revenue"] = round(entry["revenue"], 2)
result.append(entry)
result.sort(key=lambda x: x["qty_sold"], reverse=True)
return {"products": result}
# ---------------------------------------------------------------------------
# Table performance analytics
# ---------------------------------------------------------------------------
@router.get("/tables/performance")
def table_performance(
from_dt: Optional[str] = Query(default=None, alias="from"),
to_dt: Optional[str] = Query(default=None, alias="to"),
business_day_id: Optional[int] = None,
db: Session = Depends(get_db),
user: User = Depends(require_manager),
):
q = db.query(Order).filter(Order.status.in_(["closed", "paid"]))
if from_dt:
q = q.filter(Order.opened_at >= datetime.fromisoformat(from_dt))
if to_dt:
q = q.filter(Order.opened_at <= datetime.fromisoformat(to_dt))
if business_day_id:
q = q.filter(Order.business_day_id == business_day_id)
orders = q.all()
tables_db = {t.id: t for t in db.query(Table).all()}
summary: dict = {}
for order in orders:
tid = order.table_id
if tid not in summary:
t = tables_db.get(tid)
summary[tid] = {
"table_id": tid,
"table_name": (t.label or f"T{t.number}") if t else f"#{tid}",
"order_count": 0,
"revenue": 0.0,
"durations": [],
}
summary[tid]["order_count"] += 1
summary[tid]["revenue"] += sum(
i.unit_price * i.quantity for i in order.items if i.status in ("active", "paid")
)
if order.closed_at and order.opened_at:
summary[tid]["durations"].append(
(order.closed_at - order.opened_at).total_seconds() / 60
)
result = []
for entry in summary.values():
durations = entry.pop("durations")
entry["avg_duration_minutes"] = round(sum(durations) / len(durations), 1) if durations else None
entry["revenue"] = round(entry["revenue"], 2)
result.append(entry)
result.sort(key=lambda x: x["revenue"], reverse=True)
return {"tables": result}
# ---------------------------------------------------------------------------
# Traffic analysis (hour-of-day / day-of-week)
# ---------------------------------------------------------------------------
@router.get("/traffic")
def traffic_analysis(
from_dt: Optional[str] = Query(default=None, alias="from"),
to_dt: Optional[str] = Query(default=None, alias="to"),
business_day_id: Optional[int] = None,
db: Session = Depends(get_db),
user: User = Depends(require_manager),
):
q = db.query(Order)
if from_dt:
q = q.filter(Order.opened_at >= datetime.fromisoformat(from_dt))
if to_dt:
q = q.filter(Order.opened_at <= datetime.fromisoformat(to_dt))
if business_day_id:
q = q.filter(Order.business_day_id == business_day_id)
orders = q.all()
by_hour = {h: {"hour": h, "orders": 0, "revenue": 0.0} for h in range(24)}
day_labels = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
by_weekday = {d: {"day": d, "label": day_labels[d], "orders": 0, "revenue": 0.0} for d in range(7)}
for order in orders:
revenue = sum(
i.unit_price * i.quantity for i in order.items if i.status in ("active", "paid")
)
h = order.opened_at.hour
d = order.opened_at.weekday()
by_hour[h]["orders"] += 1
by_hour[h]["revenue"] += revenue
by_weekday[d]["orders"] += 1
by_weekday[d]["revenue"] += revenue
for h in by_hour:
by_hour[h]["revenue"] = round(by_hour[h]["revenue"], 2)
for d in by_weekday:
by_weekday[d]["revenue"] = round(by_weekday[d]["revenue"], 2)
return {
"by_hour": list(by_hour.values()),
"by_weekday": list(by_weekday.values()),
}