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

feat(local_backend): reservations, print analytics, workday summary, zone-PIN management

Browse Source 
- New reservations module: model, schema, router (CRUD + status updates + upcoming alerts)
  and background task for auto-expiring stale reservations
- Reports: print_products, print_categories, print_tables analytics endpoints
  plus meta_products and business_day_summary for workday close/view flow
- printer_service: configurable font sizes/weights, donut/bar chart print layout helpers,
  analytics print blocks per printer
- tables/schemas: surfaced color, zone, and other new fields on Table, Product, User, Printer
- demo_seed.py for quick dev DB population; wipe_database.py utility

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Vasileios Kousaridas
2026-06-07 20:25:22 +03:00
parent aed71a18d8
commit 79bd3b0f41
25 changed files with 2017 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

497
local_backend/services/printer_service.py
View File

@@ -57,6 +57,7 @@ _DIVIDER_CHARS = {
_PRINT_SETTING_KEYS = [
"print.ticket_mode",
"print.divider_style",
"print.dot_style",
"print.font_order_number",
"print.font_meta",
"print.font_item_name",
@@ -73,6 +74,7 @@ _PRINT_SETTING_KEYS = [
_PRINT_SETTING_DEFAULTS = {
"print.ticket_mode": "detailed",
"print.divider_style": "dash",
"print.dot_style": "dot_space:0:0",
"print.font_order_number": "48:1:0",
"print.font_meta": "0:0:0",
"print.font_item_name": "16:1:0",
@@ -86,6 +88,16 @@ _PRINT_SETTING_DEFAULTS = {
"print.beep_pattern": "double",
}
# dot_style → (fill_unit, needs_padding)
# fill_unit is repeated to fill available space; needs_padding forces a space
# before and after the dot region so it never touches the name or multiplier.
_DOT_STYLE_UNITS = {
"dot_space": ". ",
"dot": ".",
"dash_space": "- ",
"underscore": "_",
}
# Beep patterns: (n1=on×100ms, n2=off×100ms, n3=count)
# Using ESC BEL n1 n2 n3 (0x1B 0x07 n1 n2 n3)
_BEEP_PATTERNS = {
@@ -124,26 +136,79 @@ def _load_print_settings(db: Session) -> dict:
return settings
def _divider(p: Network, style: str = "dash"):
def load_divider_style(db: Session) -> str:
"""Public helper — returns the configured divider style string."""
row = db.query(PosSettings).filter(PosSettings.key == "print.divider_style").first()
return row.value if row else _PRINT_SETTING_DEFAULTS["print.divider_style"]
def _divider(p: Network, style: str = "dash", line_width: int = LINE_WIDTH):
char = _DIVIDER_CHARS.get(style, "-")
p._raw(b'\x1b\x61\x00')
if char:
p._raw(_gr(char * LINE_WIDTH + "\n"))
p._raw(_gr(char * line_width + "\n"))
else:
p._raw(b'\n')
def _item_line(name: str, qty: int, line_width: int = LINE_WIDTH) -> str:
"""Build a dot-leader line ending with 'xN'.
line_width must reflect the effective width at the chosen font size
(double-width fonts halve the available char count to 24)."""
suffix = f"x{qty}"
available = line_width - len(name) - len(suffix)
if available < 2:
# Name alone is too long — put qty on same line with a single space
return f"{name} {suffix}"
dots = (". " * ((available // 2) + 1))[:available]
return f"{name}{dots}{suffix}"
def _item_line(name: str, qty: int, line_width: int = LINE_WIDTH, dot_style: str = "dot_space") -> str:
"""Simple single-font dot-leader string — used by receipt and synopsis printers."""
suffix = f" x{qty}"
available = line_width - len(name) - 1 - len(suffix)
if available < 1:
return f"{name} x{qty}"
unit = _DOT_STYLE_UNITS.get(dot_style, ". ")
dots = (unit * (available // len(unit) + 1))[:available]
return f"{name} {dots}{suffix}"
def _print_item_line(
p: Network,
name: str, qty: int,
sz_name: int, b_name: bool,
dot_fill: str, dot_sz: int, dot_bold: bool,
line_width: int = LINE_WIDTH,
):
"""Print one item line as three segments with independent fonts:
[name font] name· [dot font] ····· [name font] ·xN\\n
Column widths are computed in physical printer columns (each double-width
char occupies 2 columns) so mixed font sizes stay aligned."""
suffix = f" x{qty}"
# Physical column width of each character under each font
name_char_w = 2 if sz_name in (32, 48) else 1
dot_char_w = 2 if dot_sz in (32, 48) else 1
# Columns consumed by name (with trailing space) and suffix
name_cols = (len(name) + 1) * name_char_w # +1 for the mandatory gap space
suffix_cols = len(suffix) * name_char_w
available_cols = line_width - name_cols - suffix_cols
if available_cols < dot_char_w:
# No room for even one dot — fall back to plain line in name font
_apply_font(p, sz_name, b_name)
_raw_text(p, f"{name} x{qty}\n")
_reset_font(p)
return
# How many dot characters fit in the available columns
n_dots = available_cols // dot_char_w
unit = _DOT_STYLE_UNITS.get(dot_fill, ". ")
dots = (unit * (n_dots // len(unit) + 1))[:n_dots]
# Emit: [name] [space] in name font
_apply_font(p, sz_name, b_name)
_raw_text(p, name + " ")
# Emit: dots in dot font
_apply_font(p, dot_sz, dot_bold)
_raw_text(p, dots)
# Emit: [space][xN]\n back in name font
_apply_font(p, sz_name, b_name)
_raw_text(p, suffix + "\n")
_reset_font(p)
def _apply_font(p: Network, size: int, bold: bool):
@@ -228,7 +293,7 @@ def send_test_print(ip: str, port: int, name: str) -> Tuple[bool, str]:
return False, str(e)
def send_test_order_print(ip: str, port: int, db: Session) -> Tuple[bool, str]:
def send_test_order_print(ip: str, port: int, db: Session, line_width: int = LINE_WIDTH) -> Tuple[bool, str]:
"""Print a fake order using the current font/layout settings — for settings preview."""
if _is_spoof_mode(db):
logger.info("Spoof printing ON — dropping test order print")
@@ -343,7 +408,7 @@ def send_test_order_print(ip: str, port: int, db: Session) -> Tuple[bool, str]:
try:
p = _get_printer(ip, port)
_print_kitchen_ticket(p, _Order(), items, _PatchedDB())
_print_kitchen_ticket(p, _Order(), items, _PatchedDB(), line_width)
p.close()
return True, ""
except Exception as e:
@@ -415,7 +480,7 @@ def _parse_options(item: OrderItem) -> dict:
return result
def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db: Session):
def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db: Session, line_width: int = LINE_WIDTH):
cfg = _load_print_settings(db)
mode = cfg.get("print.ticket_mode", "detailed")
div = cfg.get("print.divider_style", "dash")
@@ -443,24 +508,28 @@ def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db:
header = f"Παρ. #{order.id} | Τρ. {table_name} | {now_str} | {waiter_nick}"
_raw_text(p, (header.upper() if c_ord else header) + "\n")
_reset_font(p)
_divider(p, div)
_divider(p, div, line_width)
# ── DETAILED header ──────────────────────────────────────────────────────
else:
_print_line(p, f"Παραγγελια #{order.id}", sz_ord, b_ord, c_ord,
align=b'\x1b\x61\x01')
_divider(p, div)
_divider(p, div, line_width)
p._raw(b'\x1b\x61\x00')
_apply_font(p, sz_meta, b_meta)
_raw_text(p, ("ΤΡΑΠΕΖΙ:" if c_meta else "Τραπεζι:") + f" Τραπεζι {table_name}\n")
_raw_text(p, ("ΗΜΕΡΟΜΗΝΙΑ:" if c_meta else "Ημερομηνια:") + f" {now_str}\n")
_raw_text(p, ("ΣΕΡΒΙΤΟΡΟΣ:" if c_meta else "Σερβιτορος:") + f" {waiter_nick}\n")
_reset_font(p)
_divider(p, div)
_divider(p, div, line_width)
# ── Items ────────────────────────────────────────────────────────────────
# Double-width fonts halve the effective character width
item_line_width = LINE_WIDTH // 2 if sz_item in (32, 48) else LINE_WIDTH
# Dot-leader style: "STYLE:SIZE:BOLD"
raw_dot = cfg.get("print.dot_style", "dot_space:0:0")
dot_parts = raw_dot.split(":")
dot_fill = dot_parts[0] if dot_parts[0] in _DOT_STYLE_UNITS else "dot_space"
dot_sz = int(dot_parts[1]) if len(dot_parts) > 1 and dot_parts[1].isdigit() else 0
dot_bold = (dot_parts[2] == "1") if len(dot_parts) > 2 else False
for item in items:
product = db.query(Product).filter(Product.id == item.product_id).first()
@@ -468,9 +537,12 @@ def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db:
item_name = raw_name.upper() if c_item else raw_name
p._raw(b'\x1b\x61\x00')
_apply_font(p, sz_item, b_item)
_raw_text(p, _item_line(item_name, item.quantity, item_line_width) + "\n")
_reset_font(p)
_print_item_line(
p, item_name, item.quantity,
sz_item, b_item,
dot_fill, dot_sz, dot_bold,
line_width,
)
opts = _parse_options(item)
@@ -572,7 +644,7 @@ def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db:
if not compact:
p._raw(b'\n')
_divider(p, div)
_divider(p, div, line_width)
# Order-level notes
if order.notes:
@@ -581,7 +653,7 @@ def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db:
_raw_text(p, f"Σημ: {note_text}\n")
_reset_font(p)
if not compact:
_divider(p, div)
_divider(p, div, line_width)
# Footer (detailed only)
if not compact:
@@ -658,7 +730,7 @@ def print_waiter_report(ip: str, port: int, report: dict, mode: str):
value = f"{od['total']:.2f}e"
times_part = f"{time_open} - {time_close}" if time_close else time_open
prefix = f"{times_part} - {table}"
gap = LINE_WIDTH - len(prefix) - len(value)
gap = LINE_WIDTH - len(prefix) - len(value) # reports always use default width
if gap < 3:
line = f"{prefix} {value}"
else:
@@ -673,61 +745,145 @@ def print_waiter_report(ip: str, port: int, report: dict, mode: str):
logger.error("print_waiter_report failed for %s:%s — %s", ip, port, e)
def _dot_leader_line(left: str, right: str, lw: int) -> str:
"""Build a plain dot-leader line: 'left . . . . right' fitting in lw chars."""
gap = lw - len(left) - len(right)
if gap < 2:
return f"{left} {right}"
dots = (". " * (gap // 2 + 1))[:gap]
return f"{left}{dots}{right}"
def _printer_report_block(p, block: dict, mode: str, lw: int, div: str):
"""Print a single printer's block onto an already-open printer connection."""
_divider(p, div, lw)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Εκτυπωτης: {block['printer_name']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
# Stats in double-height (no bold)
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Εκτυπωσεις: {block['print_jobs']}\n")
_raw_text(p, f"Παραγγελιες: {block['orders']}\n")
_raw_text(p, f"Αντικειμενα: {block['items']}\n")
_raw_text(p, f"Αξια: {block['total']:.2f}e\n")
p._raw(b'\x1b\x21\x00')
if mode in ("orders", "detailed") and block.get("order_data"):
_divider(p, div, lw)
for od in block["order_data"]:
# Order header line: plain font, dot leader
prefix = f"#{od['id']} {od['table']}"
value = f"{od['total']:.2f}e"
header_line = _dot_leader_line(prefix, value, lw)
# No bold — plain normal font
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x45\x00')
_raw_text(p, header_line + "\n")
if mode == "detailed":
for item in od.get("items", []):
# " qty x Name . . . . total_price"
qty = item["quantity"]
name = item["name"]
total_val = item.get("total", 0.0)
left = f" {qty}x {name}"
right = f"{total_val:.2f}e"
_raw_text(p, _dot_leader_line(left, right, lw) + "\n")
def _printer_item_breakdown(p, blocks: list, lw: int, div: str):
"""Print the product analysis section (combined across all blocks)."""
# Merge breakdown across all printer blocks
combined: dict = {}
for block in blocks:
for entry in block.get("item_breakdown", []):
name = entry["name"]
if name not in combined:
combined[name] = {"qty": 0, "value": 0.0}
combined[name]["qty"] += entry["qty"]
combined[name]["value"] = round(combined[name]["value"] + entry["value"], 2)
if not combined:
return
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x10')
_raw_text(p, "ΑΝΑΛΥΣΗ ΠΡΟΙΟΝΤΩΝ\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x61\x00')
_divider(p, div, lw)
sorted_items = sorted(combined.items(), key=lambda x: x[1]["qty"], reverse=True)
for name, data in sorted_items:
line = _dot_leader_line(name, str(data["qty"]), lw)
_raw_text(p, line + "\n")
def print_printer_report(ip: str, port: int, report: dict, mode: str):
"""Print a per-printer totals report. mode='simple'|'extensive'."""
"""Print a printer history report. report contains printer_blocks list."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping printer report print")
return
try:
lw = report.get("line_width", LINE_WIDTH)
div = report.get("div_style", "dash")
p = _get_printer(ip, port)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, "ΑΝΑΦΟΡΑ ΕΚΤΥΠΩΤΗ\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Εκτυπωτης: {report['printer_name']}\n")
_raw_text(p, f"Απο: {report['from_dt']}\n")
_raw_text(p, f"Εως: {report['to_dt']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Εργασιες εκτ.: {report['print_jobs']}\n")
_raw_text(p, f"Παραγγελιες: {report['orders']}\n")
_raw_text(p, f"Αντικειμενα: {report['items']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΣΥΝΟΛΟ: {report['total']:.2f}e\n")
period = report.get("period_label", "")
if report.get("period_is_workday"):
_raw_text(p, f"Εργασιμη: {period}\n")
else:
_raw_text(p, f"Περιοδος: {period}\n")
p._raw(b'\x1b\x21\x00')
if mode == "extensive" and report.get("order_data"):
_divider(p)
blocks = report.get("printer_blocks", [])
for block in blocks:
_printer_report_block(p, block, mode, lw, div)
# Grand total footer
if len(blocks) > 1:
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x10')
_raw_text(p, "ΣΥΝΟΛΟ ΟΛΩΝ\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x61\x00')
_raw_text(p, "ΑΝΑΛΥΤΙΚΑ\n")
_divider(p)
for od in report["order_data"]:
# Header line: "HH:MM - TABLE . . . . 9.99e"
prefix = f"{od['time']} - {od['table']}"
value = f"{od['total']:.2f}e"
gap = LINE_WIDTH - len(prefix) - len(value)
if gap < 3:
header_line = f"{prefix} {value}"
else:
dots = (". " * ((gap // 2) + 1))[:gap]
header_line = f"{prefix}{dots}{value}"
p._raw(b'\x1b\x45\x01')
_raw_text(p, header_line + "\n")
p._raw(b'\x1b\x45\x00')
# Indented items
for item in od.get("items", []):
_raw_text(p, f" {item['quantity']} x {item['name']}\n")
total_jobs = sum(b["print_jobs"] for b in blocks)
total_orders = sum(b["orders"] for b in blocks)
total_items = sum(b["items"] for b in blocks)
total_value = sum(b["total"] for b in blocks)
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Εκτυπωσεις: {total_jobs}\n")
_raw_text(p, f"Παραγγελιες: {total_orders}\n")
_raw_text(p, f"Αντικειμενα: {total_items}\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΣΥΝΟΛΟ: {total_value:.2f}e\n")
p._raw(b'\x1b\x21\x00')
elif len(blocks) == 1:
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΣΥΝΟΛΟ: {blocks[0]['total']:.2f}e\n")
p._raw(b'\x1b\x21\x00')
# Optional product breakdown section
if report.get("item_breakdown") and blocks:
_printer_item_breakdown(p, blocks, lw, div)
p._raw(b'\n\n\n')
p.cut()
@@ -736,7 +892,7 @@ def print_printer_report(ip: str, port: int, report: dict, mode: str):
logger.error("print_printer_report failed for %s:%s — %s", ip, port, e)
def print_order_receipt(ip: str, port: int, receipt: dict):
def print_order_receipt(ip: str, port: int, receipt: dict, line_width: int = LINE_WIDTH):
"""Print a manager-triggered order receipt."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping order receipt print")
@@ -748,7 +904,7 @@ def print_order_receipt(ip: str, port: int, receipt: dict):
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΠΑΡΑΓΓΕΛΙΑ #{receipt['order_id']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
_divider(p, line_width=line_width)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
@@ -758,21 +914,21 @@ def print_order_receipt(ip: str, port: int, receipt: dict):
if receipt.get("closed_at"):
_raw_text(p, f"Εκλεισε: {receipt['closed_at']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
_divider(p, line_width=line_width)
for item in receipt.get("items", []):
p._raw(b'\x1b\x21\x10')
_raw_text(p, _item_line(item["name"], item["quantity"]) + "\n")
_raw_text(p, _item_line(item["name"], item["quantity"], line_width=line_width) + "\n")
p._raw(b'\x1b\x21\x00')
_raw_text(p, f" {item['unit_price']:.2f}e x{item['quantity']} = {item['total']:.2f}e\n")
_divider(p)
_divider(p, line_width=line_width)
if receipt.get("notes"):
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Σημ: {receipt['notes']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
_divider(p, line_width=line_width)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
@@ -786,7 +942,7 @@ def print_order_receipt(ip: str, port: int, receipt: dict):
logger.error("print_order_receipt failed for %s:%s — %s", ip, port, e)
def print_order_synopsis(ip: str, port: int, synopsis: dict):
def print_order_synopsis(ip: str, port: int, synopsis: dict, line_width: int = LINE_WIDTH):
"""Print a waiter-triggered order synopsis (not a kitchen ticket)."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping order synopsis print")
@@ -798,7 +954,7 @@ def print_order_synopsis(ip: str, port: int, synopsis: dict):
p._raw(b'\x1b\x21\x30')
_raw_text(p, "ΣΥΝΟΨΗ ΠΑΡΑΓΓΕΛΙΑΣ\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
_divider(p, line_width=line_width)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
@@ -806,7 +962,7 @@ def print_order_synopsis(ip: str, port: int, synopsis: dict):
_raw_text(p, f"Σερβιτορος: {synopsis['waiter_name']}\n")
_raw_text(p, f"Ωρα: {synopsis['opened_at']}\n")
p._raw(b'\x1b\x21\x00')
_divider(p)
_divider(p, line_width=line_width)
paid_items = [i for i in synopsis.get("items", []) if i["status"] == "paid"]
active_items = [i for i in synopsis.get("items", []) if i["status"] == "active"]
@@ -816,18 +972,18 @@ def print_order_synopsis(ip: str, port: int, synopsis: dict):
_raw_text(p, "ΕΚΚΡΕΜΗ:\n")
p._raw(b'\x1b\x21\x00')
for item in active_items:
_raw_text(p, f" {_item_line(item['name'], item['quantity'])}\n")
_raw_text(p, f" {_item_line(item['name'], item['quantity'], line_width=line_width - 2)}\n")
_raw_text(p, f" {item['unit_price']:.2f}e x{item['quantity']} = {item['total']:.2f}e\n")
_divider(p)
_divider(p, line_width=line_width)
if paid_items:
p._raw(b'\x1b\x21\x10')
_raw_text(p, "ΠΛΗΡΩΜΕΝΑ:\n")
p._raw(b'\x1b\x21\x00')
for item in paid_items:
_raw_text(p, f" {_item_line(item['name'], item['quantity'])}\n")
_raw_text(p, f" {_item_line(item['name'], item['quantity'], line_width=line_width - 2)}\n")
_raw_text(p, f" {item['unit_price']:.2f}e x{item['quantity']} = {item['total']:.2f}e\n")
_divider(p)
_divider(p, line_width=line_width)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
@@ -845,6 +1001,189 @@ def print_order_synopsis(ip: str, port: int, synopsis: dict):
logger.error("print_order_synopsis failed for %s:%s — %s", ip, port, e)
# ── Analytical report prints (products / categories / tables) ─────────────────
def print_products_report(ip: str, port: int, report: dict):
"""Print a product sales report. mode='smart'(sold only) or 'full'(all products)."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping products report print")
return
try:
lw = report.get("line_width", LINE_WIDTH)
div = report.get("div_style", "dash")
mode = report.get("mode", "smart")
p = _get_printer(ip, port)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
title = "ΑΝΑΦΟΡΑ ΠΡΟΙΟΝΤΩΝ" if mode == "smart" else "ΠΛΗΡΗΣ ΑΝΑΛΥΣΗ ΠΡΟΙΟΝΤΩΝ"
_raw_text(p, title + "\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
period = report.get("period_label", "")
if report.get("period_is_workday"):
_raw_text(p, f"Εργασιμη: {period}\n")
else:
_raw_text(p, f"Περιοδος: {period}\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
items = report.get("items", [])
# Only items with sales contribute to pct denominators
sold_items = [x for x in items if x["qty"] > 0]
total_qty = sum(x["qty"] for x in sold_items)
total_rev = sum(x["revenue"] for x in sold_items)
for item in items:
if item["qty"] == 0 and mode == "smart":
continue
left = item["name"]
right = f"{item['qty']} τεμ."
_raw_text(p, _dot_leader_line(left, right, lw) + "\n")
pct_rev = round(item["revenue"] / total_rev * 100, 1) if total_rev else 0
pct_qty = round(item["qty"] / total_qty * 100, 1) if total_qty else 0
_raw_text(p, f" {item['revenue']:.2f}e | {pct_rev}% εσ. {pct_qty}% τεμ.\n")
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Συνολο τεμ.: {total_qty}\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΑΞΙΑ: {total_rev:.2f}e\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\n\n\n')
p.cut()
p.close()
except Exception as e:
logger.error("print_products_report failed for %s:%s — %s", ip, port, e)
def print_categories_report(ip: str, port: int, report: dict):
"""Print a category sales report."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping categories report print")
return
try:
lw = report.get("line_width", LINE_WIDTH)
div = report.get("div_style", "dash")
mode = report.get("mode", "smart")
p = _get_printer(ip, port)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, "ΑΝΑΦΟΡΑ ΚΑΤΗΓΟΡΙΩΝ\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
period = report.get("period_label", "")
if report.get("period_is_workday"):
_raw_text(p, f"Εργασιμη: {period}\n")
else:
_raw_text(p, f"Περιοδος: {period}\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
categories = report.get("categories", [])
total_rev = sum(c["revenue"] for c in categories)
total_qty = sum(c["units_sold"] for c in categories)
for cat in categories:
_divider(p, div, lw)
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"{cat['name']}\n")
p._raw(b'\x1b\x21\x00')
_raw_text(p, _dot_leader_line("Τεμαχια", str(cat["units_sold"]), lw) + "\n")
_raw_text(p, _dot_leader_line("Εσοδα", f"{cat['revenue']:.2f}e", lw) + "\n")
pct_rev = cat.get("pct_rev", cat.get("pct", 0))
pct_qty = cat.get("pct_qty", 0)
_raw_text(p, _dot_leader_line("% Εσοδων", f"{pct_rev}%", lw) + "\n")
_raw_text(p, _dot_leader_line("% Τεμαχιων", f"{pct_qty}%", lw) + "\n")
if mode == "full" and cat.get("products"):
for prod in cat["products"]:
left = f" {prod['name']}"
right = f"{prod['qty']} | {prod.get('pct_rev', 0)}%E {prod.get('pct_qty', 0)}%T"
_raw_text(p, _dot_leader_line(left, right, lw) + "\n")
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Συνολο τεμ.: {total_qty}\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΑΞΙΑ: {total_rev:.2f}e\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\n\n\n')
p.cut()
p.close()
except Exception as e:
logger.error("print_categories_report failed for %s:%s — %s", ip, port, e)
def print_tables_report(ip: str, port: int, report: dict):
"""Print a table performance report."""
if is_spoof_mode():
logger.info("Spoof printing ON — dropping tables report print")
return
try:
lw = report.get("line_width", LINE_WIDTH)
div = report.get("div_style", "dash")
mode = report.get("mode", "smart")
p = _get_printer(ip, port)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x30')
_raw_text(p, "ΑΝΑΦΟΡΑ ΤΡΑΠΕΖΙΩΝ\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
p._raw(b'\x1b\x61\x00')
p._raw(b'\x1b\x21\x10')
period = report.get("period_label", "")
if report.get("period_is_workday"):
_raw_text(p, f"Εργασιμη: {period}\n")
else:
_raw_text(p, f"Περιοδος: {period}\n")
p._raw(b'\x1b\x21\x00')
_divider(p, div, lw)
tables = report.get("tables", [])
total_rev = sum(t["revenue"] for t in tables)
total_orders = sum(t["order_count"] for t in tables)
for t in tables:
left = t["name"]
right = f"{t['revenue']:.2f}e"
_raw_text(p, _dot_leader_line(left, right, lw) + "\n")
if mode == "full":
avg = t["revenue"] / t["order_count"] if t["order_count"] else 0
_raw_text(p, f" {t['order_count']} παρ. / μ.ο. {avg:.2f}e\n")
if t.get("avg_duration_minutes") is not None:
_raw_text(p, f" Μ. διαρκεια: {t['avg_duration_minutes']:.0f} λεπτα\n")
_divider(p, div, lw)
p._raw(b'\x1b\x61\x01')
p._raw(b'\x1b\x21\x10')
_raw_text(p, f"Παραγγελιες: {total_orders}\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\x1b\x21\x30')
_raw_text(p, f"ΑΞΙΑ: {total_rev:.2f}e\n")
p._raw(b'\x1b\x21\x00')
p._raw(b'\n\n\n')
p.cut()
p.close()
except Exception as e:
logger.error("print_tables_report failed for %s:%s — %s", ip, port, e)
# ── Routing logic ────────────────────────────────────────────────────────────
def route_and_print(order_id: int, item_ids: List[int]):
@@ -918,7 +1257,7 @@ def _do_route_and_print(order_id: int, item_ids: List[int], db: Session) -> List
error_msg = None
try:
p = _get_printer(printer.ip_address, printer.port)
_print_kitchen_ticket(p, order, zone_items, db)
_print_kitchen_ticket(p, order, zone_items, db, printer.line_width)
p.close()
success = True
for item in zone_items: