bonamin
72b12ddd9c
- Add tz.py with local_strftime/to_local helpers that read system.timezone from DB and convert UTC datetimes to venue local time before formatting - Fix all strftime() calls in orders.py, reports.py, printer_service.py that were formatting UTC datetimes without timezone conversion - Fix get_order endpoint returning raw dicts without Z suffix on datetimes, causing JS new Date() to treat timestamps as local instead of UTC - Fix fmtDate() in tokens.js that stripped the T separator before parsing, breaking UTC-to-local conversion for all report date displays - Make open/partially_paid table chips more visually distinct on dashboard Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1284 lines
48 KiB
Python
1284 lines
48 KiB
Python
"""
|
||
ESC/POS printer service — Jolimark TP850UE confirmed configuration.
|
||
|
||
Key findings from printer testing:
|
||
- Code page n=29 (CP737) is the only working Greek code page on this model.
|
||
- All Greek text MUST be sent as raw CP737 bytes via p._raw() — never p.text().
|
||
- Set the code page immediately after connecting, before any output.
|
||
- 80mm paper = 48 chars wide at standard font. Double-height keeps 48-char width.
|
||
"""
|
||
import json
|
||
import logging
|
||
import socket
|
||
import datetime
|
||
from typing import Tuple, List
|
||
|
||
from escpos.printer import Network
|
||
from sqlalchemy.orm import Session
|
||
|
||
from database import SessionLocal
|
||
from models.order import Order, OrderItem, PrintLog
|
||
from tz import to_local
|
||
from models.printer import Printer
|
||
from models.product import Product
|
||
from models.settings import PosSettings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
LINE_WIDTH = 48
|
||
PRINTER_TIMEOUT = 5
|
||
|
||
|
||
# ── Low-level helpers ────────────────────────────────────────────────────────
|
||
|
||
def _get_printer(ip: str, port: int) -> Network:
|
||
p = Network(ip, port, timeout=PRINTER_TIMEOUT)
|
||
p._raw(b'\x1b\x40') # ESC @ — reset printer
|
||
p._raw(b'\x1b\x74\x1d') # ESC t 29 — select CP737 (Greek) — confirmed n=29
|
||
return p
|
||
|
||
|
||
def _gr(text: str) -> bytes:
|
||
"""Encode text to CP737 bytes. Replaces unknown chars instead of crashing."""
|
||
return text.encode('cp737', errors='replace')
|
||
|
||
|
||
def _raw_text(p: Network, text: str):
|
||
"""Send text as raw CP737 bytes — the ONLY safe way to print Greek."""
|
||
p._raw(_gr(text))
|
||
|
||
|
||
_DIVIDER_CHARS = {
|
||
"dash": "-",
|
||
"equals": "=",
|
||
"star": "*",
|
||
"empty": "",
|
||
}
|
||
|
||
_PRINT_SETTING_KEYS = [
|
||
"print.ticket_mode",
|
||
"print.divider_style",
|
||
"print.dot_style",
|
||
"print.font_order_number",
|
||
"print.font_meta",
|
||
"print.font_item_name",
|
||
"print.font_quick",
|
||
"print.font_pref",
|
||
"print.font_extra",
|
||
"print.font_ingredient",
|
||
"print.font_item_note",
|
||
"print.font_order_note",
|
||
"print.beep_on_ticket",
|
||
"print.beep_pattern",
|
||
]
|
||
|
||
_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",
|
||
"print.font_quick": "0:0:0",
|
||
"print.font_pref": "0:0:0",
|
||
"print.font_extra": "0:0:0",
|
||
"print.font_ingredient": "0:0:0",
|
||
"print.font_item_note": "0:0:0",
|
||
"print.font_order_note": "0:1:0",
|
||
"print.beep_on_ticket": "true",
|
||
"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 = {
|
||
"single": (2, 2, 1), # one 200ms beep
|
||
"double": (1, 1, 2), # two short beeps
|
||
"triple": (1, 1, 3), # three short beeps
|
||
"long": (5, 2, 1), # one 500ms beep
|
||
}
|
||
|
||
# SIZE byte values (ESC ! base, no bold bit):
|
||
# 0 = normal
|
||
# 16 = double-height (bit4)
|
||
# 32 = double-width (bit5)
|
||
# 48 = double-height + double-width (bits 4+5)
|
||
# Bold applied via ESC E, caps applied in software before encoding.
|
||
|
||
def _decode_font(value: str) -> tuple[int, bool, bool]:
|
||
"""Parse 'SIZE:BOLD:CAPS' string → (esc_bang_byte, bold_flag, caps_flag)."""
|
||
try:
|
||
parts = str(value).split(":")
|
||
size = int(parts[0])
|
||
bold = len(parts) > 1 and parts[1] == "1"
|
||
caps = len(parts) > 2 and parts[2] == "1"
|
||
return size, bold, caps
|
||
except (ValueError, AttributeError):
|
||
return 0, False, False
|
||
|
||
|
||
def _load_print_settings(db: Session) -> dict:
|
||
rows = db.query(PosSettings).filter(
|
||
PosSettings.key.in_(_PRINT_SETTING_KEYS)
|
||
).all()
|
||
settings = dict(_PRINT_SETTING_DEFAULTS)
|
||
for row in rows:
|
||
settings[row.key] = row.value
|
||
return settings
|
||
|
||
|
||
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"))
|
||
else:
|
||
p._raw(b'\n')
|
||
|
||
|
||
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):
|
||
p._raw(bytes([0x1b, 0x21, size]))
|
||
p._raw(b'\x1b\x45\x01' if bold else b'\x1b\x45\x00')
|
||
|
||
|
||
def _reset_font(p: Network):
|
||
p._raw(b'\x1b\x21\x00')
|
||
p._raw(b'\x1b\x45\x00')
|
||
|
||
|
||
def _print_line(p: Network, text: str, size: int, bold: bool, caps: bool,
|
||
align: bytes = b'\x1b\x61\x00'):
|
||
"""Apply font, optionally capitalize, print text + newline, reset font."""
|
||
p._raw(align)
|
||
_apply_font(p, size, bold)
|
||
out = text.upper() if caps else text
|
||
_raw_text(p, out + "\n")
|
||
_reset_font(p)
|
||
|
||
|
||
def _greek_date(dt: datetime.datetime) -> str:
|
||
"""Return date/time string in Greek format: HH:MM DD-MM-YYYY, converted to venue local time."""
|
||
return to_local(dt).strftime("%H:%M %d-%m-%Y")
|
||
|
||
|
||
def check_printer(ip: str, port: int) -> bool:
|
||
"""Quick TCP connect check — no data sent."""
|
||
try:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
s.settimeout(2)
|
||
s.connect((ip, port))
|
||
s.close()
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def is_spoof_mode() -> bool:
|
||
"""Stateless check — opens its own DB session. For use outside route_and_print."""
|
||
db = SessionLocal()
|
||
try:
|
||
return _is_spoof_mode(db)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def send_test_beep(ip: str, port: int, n1: int, n2: int, n3: int) -> Tuple[bool, str]:
|
||
"""Send a standalone beep-only job. n1=on-time×100ms, n2=off-time×100ms, n3=count."""
|
||
if is_spoof_mode():
|
||
logger.info("Spoof printing ON — dropping test beep")
|
||
return True, ""
|
||
try:
|
||
p = _get_printer(ip, port)
|
||
p._raw(bytes([0x1b, 0x07, n1, n2, n3]))
|
||
p.close()
|
||
return True, ""
|
||
except Exception as e:
|
||
logger.error("Test beep failed for %s:%s — %s", ip, port, e)
|
||
return False, str(e)
|
||
|
||
|
||
def send_test_print(ip: str, port: int, name: str) -> Tuple[bool, str]:
|
||
if is_spoof_mode():
|
||
logger.info("Spoof printing ON — dropping test print for %s", name)
|
||
return True, ""
|
||
try:
|
||
p = _get_printer(ip, port)
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, f"TEST — {name}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
now = to_local(datetime.datetime.now(datetime.timezone.utc)).strftime("%Y-%m-%d %H:%M:%S")
|
||
_raw_text(p, f"{now}\n")
|
||
p._raw(b'\n\n\n')
|
||
p.cut()
|
||
p.close()
|
||
return True, ""
|
||
except Exception as e:
|
||
logger.error("Test print failed for %s:%s — %s", ip, port, e)
|
||
return False, str(e)
|
||
|
||
|
||
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")
|
||
return True, ""
|
||
|
||
# ── Fake data structures (no DB writes) ──────────────────────────────────
|
||
class _Table:
|
||
label = "O2"
|
||
number = 2
|
||
|
||
class _User:
|
||
nickname = "bonamin"
|
||
username = "bonamin"
|
||
|
||
class _Order:
|
||
id = 99
|
||
table = _Table()
|
||
opener = _User()
|
||
table_id = 2
|
||
opened_by = 1
|
||
notes = "Χωρις καψαλισμα παρακαλω"
|
||
|
||
class _Item:
|
||
def __init__(self, product_id, quantity, selected_options, removed_ingredients, notes):
|
||
self.product_id = product_id
|
||
self.quantity = quantity
|
||
self.selected_options = selected_options
|
||
self.removed_ingredients = removed_ingredients
|
||
self.notes = notes
|
||
|
||
import json as _json
|
||
|
||
items = [
|
||
# Item 1: Freddo Espresso — quick options + preference + note
|
||
_Item(
|
||
product_id=1001,
|
||
quantity=2,
|
||
selected_options=_json.dumps([
|
||
{"name": "Διπλος", "price_delta": 0.5, "type": "quick"},
|
||
{"name": "Εξτρα ζαχαρη", "price_delta": 0.0, "type": "quick"},
|
||
{"name": "Παγωμενος", "price_delta": 0.0, "type": "quick"},
|
||
{"name": "Γαλα", "price_delta": 0.0, "type": "pref"},
|
||
{"name": "Βρωμης", "price_delta": 0.3, "type": "pref_sub"},
|
||
]),
|
||
removed_ingredients=None,
|
||
notes="Πολυ κρυο παρακαλω",
|
||
),
|
||
# Item 2: Club Sandwich — extra with sub + removed ingredients
|
||
_Item(
|
||
product_id=1002,
|
||
quantity=1,
|
||
selected_options=_json.dumps([
|
||
{"name": "Extra Bacon", "price_delta": 1.5, "type": "extra"},
|
||
{"name": "Τραγανο", "price_delta": 0.0, "type": "extra_sub"},
|
||
{"name": "Extra Bacon", "price_delta": 1.5, "type": "extra"},
|
||
{"name": "Τραγανο", "price_delta": 0.0, "type": "extra_sub"},
|
||
{"name": "Ψωμι", "price_delta": 0.0, "type": "pref"},
|
||
{"name": "Σικαλεως", "price_delta": 0.0, "type": "pref_sub"},
|
||
]),
|
||
removed_ingredients=_json.dumps(["Ντοματα", "Μουσταρδα"]),
|
||
notes=None,
|
||
),
|
||
# Item 3: Margherita — quick + extra + removed
|
||
_Item(
|
||
product_id=1003,
|
||
quantity=3,
|
||
selected_options=_json.dumps([
|
||
{"name": "Well Done", "price_delta": 0.0, "type": "quick"},
|
||
{"name": "Extra Τυρι", "price_delta": 1.0, "type": "extra"},
|
||
{"name": "Extra Τυρι", "price_delta": 1.0, "type": "extra"},
|
||
{"name": "Extra Τυρι", "price_delta": 1.0, "type": "extra"},
|
||
]),
|
||
removed_ingredients=_json.dumps(["Ελιες", "Κρεμμυδι"]),
|
||
notes=None,
|
||
),
|
||
]
|
||
|
||
# Patch product lookup so _print_kitchen_ticket gets real names
|
||
_FAKE_NAMES = {1001: "Freddo Espresso", 1002: "Club Sandwich", 1003: "Margherita Pizza"}
|
||
|
||
# Monkey-patch db.query for Product only inside this call
|
||
_orig_query = db.query
|
||
|
||
class _FakeQuery:
|
||
def __init__(self, model):
|
||
self._model = model
|
||
self._filter_id = None
|
||
def filter(self, *args):
|
||
# extract id from the filter expression value
|
||
for arg in args:
|
||
try:
|
||
self._filter_id = arg.right.value
|
||
except Exception:
|
||
pass
|
||
return self
|
||
def first(self):
|
||
if self._model.__name__ == "Product" and self._filter_id in _FAKE_NAMES:
|
||
class _P:
|
||
name = _FAKE_NAMES[self._filter_id]
|
||
return _P()
|
||
return _orig_query(self._model).filter(self._model.id == self._filter_id).first()
|
||
|
||
class _PatchedDB:
|
||
def query(self, model):
|
||
from models.product import Product as _Product
|
||
if model is _Product:
|
||
return _FakeQuery(model)
|
||
return _orig_query(model)
|
||
# delegate everything else to real db
|
||
def __getattr__(self, name):
|
||
return getattr(db, name)
|
||
|
||
try:
|
||
p = _get_printer(ip, port)
|
||
_print_kitchen_ticket(p, _Order(), items, _PatchedDB(), line_width)
|
||
p.close()
|
||
return True, ""
|
||
except Exception as e:
|
||
logger.error("Test order print failed for %s:%s — %s", ip, port, e)
|
||
return False, str(e)
|
||
|
||
|
||
# ── Receipt formatting ───────────────────────────────────────────────────────
|
||
|
||
def _parse_options(item: OrderItem) -> dict:
|
||
"""
|
||
Parse selected_options JSON into grouped dict:
|
||
{ 'quick': [(name, qty)], 'pref': [(name, sub|None)],
|
||
'extra': [(name, sub|None, qty)], 'unknown': [name] }
|
||
Falls back gracefully when type tags are absent (old data).
|
||
"""
|
||
result = {"quick": [], "pref": [], "extra": [], "unknown": []}
|
||
if not item.selected_options:
|
||
return result
|
||
|
||
try:
|
||
raw = json.loads(item.selected_options)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return result
|
||
|
||
if not isinstance(raw, list):
|
||
return result
|
||
|
||
i = 0
|
||
while i < len(raw):
|
||
entry = raw[i]
|
||
if not isinstance(entry, dict):
|
||
i += 1
|
||
continue
|
||
name = entry.get("name") or ""
|
||
etype = entry.get("type")
|
||
|
||
# Peek at next entry to collect sub-choice
|
||
sub = None
|
||
if i + 1 < len(raw):
|
||
nxt = raw[i + 1]
|
||
if isinstance(nxt, dict) and nxt.get("type") in ("pref_sub", "extra_sub"):
|
||
sub = nxt.get("name") or ""
|
||
i += 1 # consume sub
|
||
|
||
if etype == "quick":
|
||
# Collapse repeated quick entries into a single (name, qty) tuple
|
||
existing = next((q for q in result["quick"] if q[0] == name), None)
|
||
if existing:
|
||
result["quick"][result["quick"].index(existing)] = (name, existing[1] + 1)
|
||
else:
|
||
result["quick"].append((name, 1))
|
||
elif etype == "pref":
|
||
result["pref"].append((name, sub))
|
||
elif etype == "extra":
|
||
# Collapse repeated extra entries (same name+sub) → (name, sub, qty)
|
||
existing = next((e for e in result["extra"] if e[0] == name and e[1] == sub), None)
|
||
if existing:
|
||
result["extra"][result["extra"].index(existing)] = (name, sub, existing[2] + 1)
|
||
else:
|
||
result["extra"].append((name, sub, 1))
|
||
else:
|
||
# Legacy data without type tag — treat as unknown, display plainly
|
||
if name:
|
||
result["unknown"].append(name + (f" · {sub}" if sub else ""))
|
||
|
||
i += 1
|
||
|
||
return result
|
||
|
||
|
||
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")
|
||
compact = (mode == "compact")
|
||
|
||
sz_ord, b_ord, c_ord = _decode_font(cfg["print.font_order_number"])
|
||
sz_meta, b_meta, c_meta = _decode_font(cfg["print.font_meta"])
|
||
sz_item, b_item, c_item = _decode_font(cfg["print.font_item_name"])
|
||
sz_qk, b_qk, c_qk = _decode_font(cfg["print.font_quick"])
|
||
sz_pr, b_pr, c_pr = _decode_font(cfg["print.font_pref"])
|
||
sz_ex, b_ex, c_ex = _decode_font(cfg["print.font_extra"])
|
||
sz_ing, b_ing, c_ing = _decode_font(cfg["print.font_ingredient"])
|
||
sz_note, b_note, c_note = _decode_font(cfg["print.font_item_note"])
|
||
sz_onote,b_onote,c_onote= _decode_font(cfg["print.font_order_note"])
|
||
|
||
# Resolve display names
|
||
table_name = order.table.label or str(order.table.number) if order.table else str(order.table_id)
|
||
waiter_nick = (order.opener.nickname or order.opener.username) if order.opener else str(order.opened_by)
|
||
now_str = _greek_date(datetime.datetime.now(datetime.timezone.utc))
|
||
|
||
# ── COMPACT header — single line ────────────────────────────────────────
|
||
if compact:
|
||
p._raw(b'\x1b\x61\x00')
|
||
_apply_font(p, sz_ord, b_ord)
|
||
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, 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, 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, line_width)
|
||
|
||
# ── Items ────────────────────────────────────────────────────────────────
|
||
# 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()
|
||
raw_name = product.name if product else f"Product #{item.product_id}"
|
||
item_name = raw_name.upper() if c_item else raw_name
|
||
|
||
p._raw(b'\x1b\x61\x00')
|
||
_print_item_line(
|
||
p, item_name, item.quantity,
|
||
sz_item, b_item,
|
||
dot_fill, dot_sz, dot_bold,
|
||
line_width,
|
||
)
|
||
|
||
opts = _parse_options(item)
|
||
|
||
# Quick options (* marker)
|
||
if opts["quick"]:
|
||
if compact:
|
||
parts = []
|
||
for name, qty in opts["quick"]:
|
||
n = name.upper() if c_qk else name
|
||
parts.append(f"{n} x{qty}" if qty > 1 else n)
|
||
_apply_font(p, sz_qk, b_qk)
|
||
_raw_text(p, "* " + " | ".join(parts) + "\n")
|
||
_reset_font(p)
|
||
else:
|
||
for name, qty in opts["quick"]:
|
||
n = name.upper() if c_qk else name
|
||
line = f"* {n} x{qty}" if qty > 1 else f"* {n}"
|
||
_apply_font(p, sz_qk, b_qk)
|
||
_raw_text(p, line + "\n")
|
||
_reset_font(p)
|
||
|
||
# Preferences (> marker)
|
||
if opts["pref"]:
|
||
if compact:
|
||
parts = []
|
||
for name, sub in opts["pref"]:
|
||
n = name.upper() if c_pr else name
|
||
s = (sub.upper() if c_pr else sub) if sub else None
|
||
parts.append(f"{n} · {s}" if s else n)
|
||
_apply_font(p, sz_pr, b_pr)
|
||
_raw_text(p, "> " + " | ".join(parts) + "\n")
|
||
_reset_font(p)
|
||
else:
|
||
for name, sub in opts["pref"]:
|
||
n = name.upper() if c_pr else name
|
||
s = (sub.upper() if c_pr else sub) if sub else None
|
||
line = f"> {n} · {s}" if s else f"> {n}"
|
||
_apply_font(p, sz_pr, b_pr)
|
||
_raw_text(p, line + "\n")
|
||
_reset_font(p)
|
||
|
||
# Extras (+ marker)
|
||
if opts["extra"]:
|
||
if compact:
|
||
parts = []
|
||
for name, sub, qty in opts["extra"]:
|
||
n = name.upper() if c_ex else name
|
||
s = (sub.upper() if c_ex else sub) if sub else None
|
||
part = f"{n} · {s}" if s else n
|
||
if qty > 1:
|
||
part += f" · x{qty}"
|
||
parts.append(part)
|
||
_apply_font(p, sz_ex, b_ex)
|
||
_raw_text(p, "+ " + " | ".join(parts) + "\n")
|
||
_reset_font(p)
|
||
else:
|
||
for name, sub, qty in opts["extra"]:
|
||
n = name.upper() if c_ex else name
|
||
s = (sub.upper() if c_ex else sub) if sub else None
|
||
line = f"+ {n}"
|
||
if s:
|
||
line += f" · {s}"
|
||
if qty > 1:
|
||
line += f" · x{qty}"
|
||
_apply_font(p, sz_ex, b_ex)
|
||
_raw_text(p, line + "\n")
|
||
_reset_font(p)
|
||
|
||
# Legacy untagged options
|
||
for entry in opts["unknown"]:
|
||
_apply_font(p, sz_ex, b_ex)
|
||
_raw_text(p, f"+ {entry}\n")
|
||
_reset_font(p)
|
||
|
||
# Removed ingredients (- marker)
|
||
if item.removed_ingredients:
|
||
try:
|
||
removed = json.loads(item.removed_ingredients)
|
||
if removed:
|
||
names = [n.upper() if c_ing else n for n in removed]
|
||
joined = " · ".join(names)
|
||
_apply_font(p, sz_ing, b_ing)
|
||
_raw_text(p, f"- ΧΩΡΙΣ: {joined}\n")
|
||
_reset_font(p)
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
# Per-item note
|
||
if item.notes:
|
||
note_text = item.notes.upper() if c_note else item.notes
|
||
_apply_font(p, sz_note, b_note)
|
||
if compact:
|
||
_raw_text(p, f"! {note_text}\n")
|
||
else:
|
||
_raw_text(p, f"\n(!) {note_text}\n\n")
|
||
_reset_font(p)
|
||
|
||
# Blank line between items in detailed mode
|
||
if not compact:
|
||
p._raw(b'\n')
|
||
|
||
_divider(p, div, line_width)
|
||
|
||
# Order-level notes
|
||
if order.notes:
|
||
note_text = order.notes.upper() if c_onote else order.notes
|
||
_apply_font(p, sz_onote, b_onote)
|
||
_raw_text(p, f"Σημ: {note_text}\n")
|
||
_reset_font(p)
|
||
if not compact:
|
||
_divider(p, div, line_width)
|
||
|
||
# Footer (detailed only)
|
||
if not compact:
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, "Τελος Παραγγελιας\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
|
||
p._raw(b'\n\n\n')
|
||
|
||
# Beep before cut so the buzzer fires as the paper advances
|
||
beep_enabled = cfg.get("print.beep_on_ticket", "true") == "true"
|
||
if beep_enabled:
|
||
pattern = cfg.get("print.beep_pattern", "double")
|
||
if pattern.startswith("custom:"):
|
||
# custom:n1:n2:n3
|
||
try:
|
||
_, n1s, n2s, n3s = pattern.split(":")
|
||
beep_bytes = (int(n1s), int(n2s), int(n3s))
|
||
except (ValueError, TypeError):
|
||
beep_bytes = _BEEP_PATTERNS["double"]
|
||
else:
|
||
beep_bytes = _BEEP_PATTERNS.get(pattern, _BEEP_PATTERNS["double"])
|
||
p._raw(bytes([0x1b, 0x07, beep_bytes[0], beep_bytes[1], beep_bytes[2]]))
|
||
|
||
p.cut()
|
||
|
||
|
||
# ── On-demand report / receipt prints ────────────────────────────────────────
|
||
|
||
def print_waiter_report(ip: str, port: int, report: dict, mode: str):
|
||
"""Print a waiter shift/period report. mode='simple'|'extensive'."""
|
||
if is_spoof_mode():
|
||
logger.info("Spoof printing ON — dropping waiter report print")
|
||
return
|
||
try:
|
||
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['waiter_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['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")
|
||
p._raw(b'\x1b\x21\x00')
|
||
|
||
if mode == "extensive" and report.get("order_data"):
|
||
_divider(p)
|
||
p._raw(b'\x1b\x61\x00')
|
||
_raw_text(p, "ΑΝΑΛΥΤΙΚΑ\n")
|
||
_divider(p)
|
||
for od in report["order_data"]:
|
||
# Build right-aligned total: "HH:MM - HH:MM - TABLE . . . 9.99e"
|
||
time_open = od.get("time_open", "")
|
||
time_close = od.get("time_close", "")
|
||
table = od["table"]
|
||
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) # reports always use default width
|
||
if gap < 3:
|
||
line = f"{prefix} {value}"
|
||
else:
|
||
dots = (". " * ((gap // 2) + 1))[:gap]
|
||
line = f"{prefix}{dots}{value}"
|
||
_raw_text(p, line + "\n")
|
||
|
||
p._raw(b'\n\n\n')
|
||
p.cut()
|
||
p.close()
|
||
except Exception as e:
|
||
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 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')
|
||
|
||
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')
|
||
|
||
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')
|
||
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()
|
||
p.close()
|
||
except Exception as e:
|
||
logger.error("print_printer_report failed for %s:%s — %s", ip, port, e)
|
||
|
||
|
||
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")
|
||
return
|
||
try:
|
||
p = _get_printer(ip, port)
|
||
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, f"ΠΑΡΑΓΓΕΛΙΑ #{receipt['order_id']}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p, line_width=line_width)
|
||
|
||
p._raw(b'\x1b\x61\x00')
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Τραπεζι: {receipt['table_name']}\n")
|
||
_raw_text(p, f"Σερβιτορος: {receipt['waiter_name']}\n")
|
||
_raw_text(p, f"Ανοιχτηκε: {receipt['opened_at']}\n")
|
||
if receipt.get("closed_at"):
|
||
_raw_text(p, f"Εκλεισε: {receipt['closed_at']}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_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"], 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, 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, line_width=line_width)
|
||
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, f"ΣΥΝΟΛΟ: {receipt['total']:.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_order_receipt failed for %s:%s — %s", ip, port, e)
|
||
|
||
|
||
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")
|
||
return
|
||
try:
|
||
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, line_width=line_width)
|
||
|
||
p._raw(b'\x1b\x61\x00')
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Τραπεζι: {synopsis['table_name']}\n")
|
||
_raw_text(p, f"Σερβιτορος: {synopsis['waiter_name']}\n")
|
||
_raw_text(p, f"Ωρα: {synopsis['opened_at']}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_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"]
|
||
|
||
if active_items:
|
||
p._raw(b'\x1b\x21\x10')
|
||
_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'], 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, 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'], 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, line_width=line_width)
|
||
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, f"ΣΥΝΟΛΟ: {synopsis['total']:.2f}e\n")
|
||
if synopsis.get('paid_total', 0) > 0:
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Πληρωμενο: {synopsis['paid_total']:.2f}e\n")
|
||
_raw_text(p, f"Εκκρεμει: {synopsis['remaining']:.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_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]):
|
||
"""
|
||
Background task: group items by printer zone, send to each printer.
|
||
Printer failures are logged but never raise — order is already saved.
|
||
"""
|
||
db: Session = SessionLocal()
|
||
try:
|
||
_do_route_and_print(order_id, item_ids, db)
|
||
except Exception as e:
|
||
logger.exception("Unexpected error in route_and_print for order %s: %s", order_id, e)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def route_and_print_sync(order_id: int, item_ids: List[int], db: Session) -> List[dict]:
|
||
"""
|
||
Synchronous variant used when the caller needs print results.
|
||
Returns a list of per-printer result dicts:
|
||
{ printer_name, success, error }
|
||
"""
|
||
return _do_route_and_print(order_id, item_ids, db)
|
||
|
||
|
||
def _is_spoof_mode(db: Session) -> bool:
|
||
row = db.query(PosSettings).filter(PosSettings.key == "dev.spoof_printing").first()
|
||
return row is not None and row.value == "true"
|
||
|
||
|
||
def _do_route_and_print(order_id: int, item_ids: List[int], db: Session) -> List[dict]:
|
||
if _is_spoof_mode(db):
|
||
logger.info("Spoof printing ON — dropping print job for order %s", order_id)
|
||
for item_id in item_ids:
|
||
item = db.query(OrderItem).filter(OrderItem.id == item_id).first()
|
||
if item:
|
||
item.printed = True
|
||
db.commit()
|
||
return [{"printer_name": "spoof", "success": True, "error": None}]
|
||
|
||
results = []
|
||
|
||
order = db.query(Order).filter(Order.id == order_id).first()
|
||
if not order:
|
||
logger.error("route_and_print: order %s not found", order_id)
|
||
return results
|
||
|
||
items = db.query(OrderItem).filter(OrderItem.id.in_(item_ids)).all()
|
||
|
||
# Group items by printer zone
|
||
zone_map: dict[int, List[OrderItem]] = {}
|
||
unzoned: List[OrderItem] = []
|
||
for item in items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
if product and product.printer_zone_id:
|
||
zone_map.setdefault(product.printer_zone_id, []).append(item)
|
||
else:
|
||
unzoned.append(item)
|
||
|
||
if unzoned:
|
||
logger.warning("order %s has %d item(s) with no printer zone — skipped", order_id, len(unzoned))
|
||
|
||
for printer_id, zone_items in zone_map.items():
|
||
printer = db.query(Printer).filter(Printer.id == printer_id, Printer.is_active == True).first()
|
||
if not printer:
|
||
logger.warning("Printer %s not found or inactive", printer_id)
|
||
results.append({"printer_name": f"#{printer_id}", "success": False, "error": "Printer not found or inactive"})
|
||
continue
|
||
|
||
success = False
|
||
error_msg = None
|
||
try:
|
||
p = _get_printer(printer.ip_address, printer.port)
|
||
_print_kitchen_ticket(p, order, zone_items, db, printer.line_width)
|
||
p.close()
|
||
success = True
|
||
for item in zone_items:
|
||
item.printed = True
|
||
db.commit()
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.error("Print failed for printer %s (%s:%s): %s", printer.name, printer.ip_address, printer.port, e)
|
||
|
||
log = PrintLog(
|
||
order_id=order_id,
|
||
printer_id=printer_id,
|
||
item_ids=json.dumps([i.id for i in zone_items]),
|
||
success=success,
|
||
error_message=error_msg,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
|
||
results.append({"printer_name": printer.name, "success": success, "error": error_msg})
|
||
|
||
return results
|