bonamin
34ae328b0d
Snapshot of in-progress work across local_backend, manager_dashboard, and waiter_pwa (pricing, chat, fiscal, prep zones, recovery codes, CRM, inventory, permissions), plus the nginx/docker-compose deploy fixes for the Unraid + NPM reverse-proxy setup. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2205 lines
87 KiB
Python
2205 lines
87 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, Optional
|
||
|
||
from escpos.printer import Network
|
||
from sqlalchemy.orm import Session
|
||
|
||
from database import SessionLocal
|
||
from models.order import Order, OrderItem, PrintLog, PrintJob
|
||
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, codepage_n: int = 29) -> Network:
|
||
p = Network(ip, port, timeout=PRINTER_TIMEOUT)
|
||
p._raw(b'\x1b\x40') # ESC @ — reset printer
|
||
p._raw(bytes([0x1b, 0x74, codepage_n])) # ESC t n — select Greek code page
|
||
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')
|
||
|
||
|
||
_UNIT_SUFFIXES = {
|
||
"kg": lambda q: f" {_fmt_decimal(q)}kg",
|
||
"liter": lambda q: f" {_fmt_decimal(q)}L",
|
||
"gram": lambda q: f" {int(q)}g",
|
||
"ml": lambda q: f" {int(q)}mL",
|
||
}
|
||
|
||
|
||
def _fmt_decimal(q: float) -> str:
|
||
"""Format a decimal quantity: drop trailing zero if it's a whole number."""
|
||
return f"{q:.1f}" if q != int(q) else f"{int(q)}"
|
||
|
||
|
||
def _qty_suffix(qty: float, unit_type: str) -> str:
|
||
"""Return the quantity suffix string for a given unit, e.g. ' x2', ' 1.5kg'."""
|
||
fn = _UNIT_SUFFIXES.get(unit_type)
|
||
if fn:
|
||
return fn(qty)
|
||
return f" x{int(qty)}"
|
||
|
||
|
||
def _item_line(name: str, qty: float, line_width: int = LINE_WIDTH, dot_style: str = "dot_space",
|
||
unit_type: str = "piece") -> str:
|
||
"""Simple single-font dot-leader string — used by receipt and synopsis printers."""
|
||
suffix = _qty_suffix(qty, unit_type)
|
||
available = line_width - len(name) - 1 - len(suffix)
|
||
if available < 1:
|
||
return f"{name}{suffix}"
|
||
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: float,
|
||
sz_name: int, b_name: bool,
|
||
dot_fill: str, dot_sz: int, dot_bold: bool,
|
||
line_width: int = LINE_WIDTH,
|
||
unit_type: str = "piece",
|
||
):
|
||
"""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 = _qty_suffix(qty, unit_type)
|
||
|
||
# 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}{suffix}\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, codepage_n: int = 29) -> 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, codepage_n)
|
||
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, codepage_n: int = 29) -> 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, codepage_n)
|
||
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, codepage_n: int = 29) -> 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]
|
||
unit_type = "piece"
|
||
category_id = None
|
||
printer_zone_id = None
|
||
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, codepage_n)
|
||
_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 _sort_and_group_items(items: List[OrderItem], zone, db: Session) -> List[OrderItem]:
|
||
"""Apply zone sort_items_by and group_by_category ordering to the item list."""
|
||
if zone is None:
|
||
return items
|
||
|
||
sort_by = getattr(zone, 'sort_items_by', 'order_time') or 'order_time'
|
||
group_cat = bool(getattr(zone, 'group_by_category', False))
|
||
|
||
# First apply item sort within (or without) category groups
|
||
def _sort_key(item: OrderItem):
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
if sort_by == 'item_count':
|
||
return -item.quantity # descending
|
||
if sort_by == 'alpha':
|
||
return (product.name or '').lower() if product else ''
|
||
# 'order_time': keep original ordering (stable sort)
|
||
return item.added_at or datetime.datetime.min
|
||
|
||
if sort_by != 'order_time':
|
||
items = sorted(items, key=_sort_key)
|
||
|
||
if not group_cat:
|
||
return items
|
||
|
||
# Group by category, ordered by zone.category_order list
|
||
try:
|
||
cat_order: List[int] = json.loads(zone.category_order) if zone.category_order else []
|
||
except (json.JSONDecodeError, TypeError):
|
||
cat_order = []
|
||
|
||
cat_rank = {cat_id: i for i, cat_id in enumerate(cat_order)}
|
||
|
||
def _cat_for_item(item: OrderItem) -> int:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
if product and product.category_id:
|
||
return product.category_id
|
||
return 0
|
||
|
||
cat_id_for = {item.id: _cat_for_item(item) for item in items}
|
||
|
||
def _group_key(item: OrderItem) -> tuple:
|
||
cat_id = cat_id_for[item.id]
|
||
rank = cat_rank.get(cat_id, len(cat_order)) # unlisted cats go last
|
||
return (rank, cat_id)
|
||
|
||
return sorted(items, key=_group_key)
|
||
|
||
|
||
def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db: Session, line_width: int = LINE_WIDTH, zone=None, ephemeral_svc_entries: Optional[List[dict]] = None):
|
||
cfg = _load_print_settings(db)
|
||
mode = cfg.get("print.ticket_mode", "detailed")
|
||
div = cfg.get("print.divider_style", "dash")
|
||
compact = (mode == "compact")
|
||
|
||
# Apply zone-level sorting / category grouping
|
||
items = _sort_and_group_items(items, zone, db)
|
||
|
||
print_checkboxes = bool(getattr(zone, 'print_checkboxes', False)) if zone else False
|
||
|
||
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))
|
||
|
||
customer_count = getattr(order, 'customer_count', None)
|
||
|
||
# ── 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}"
|
||
if customer_count:
|
||
header += f" | {customer_count}ατ."
|
||
_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")
|
||
if customer_count:
|
||
_raw_text(p, ("ΑΤΟΜΑ:" if c_meta else "Ατομα:") + f" {customer_count}\n")
|
||
_reset_font(p)
|
||
_divider(p, div, line_width)
|
||
|
||
# ── Dot-leader style ────────────────────────────────────────────────────
|
||
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
|
||
|
||
# Split items: priced service items (saved as OrderItems) vs regular items
|
||
def _item_is_svc(item):
|
||
prod = db.query(Product).filter(Product.id == item.product_id).first()
|
||
return bool(prod and getattr(prod, 'is_service_item', False))
|
||
|
||
svc_order_items = [i for i in items if _item_is_svc(i)]
|
||
regular_items = [i for i in items if not _item_is_svc(i)]
|
||
|
||
# Ephemeral service entries (priceless, not saved as OrderItems)
|
||
eph = ephemeral_svc_entries or []
|
||
|
||
has_svc_section = bool(svc_order_items) or bool(eph)
|
||
has_regular = bool(regular_items)
|
||
has_both = has_svc_section and has_regular
|
||
|
||
# ── SERVICE section (printed FIRST, after header) ────────────────────────
|
||
if has_svc_section:
|
||
if has_both:
|
||
_divider(p, div, line_width)
|
||
p._raw(b'\x1b\x61\x01') # center
|
||
_apply_font(p, sz_item, b_item)
|
||
_raw_text(p, ("SERVICE\n" if c_item else "Service\n"))
|
||
_reset_font(p)
|
||
p._raw(b'\x1b\x61\x00')
|
||
_divider(p, div, line_width)
|
||
|
||
# Priced saved service order items
|
||
for item in svc_order_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
|
||
unit_type = (product.unit_type or "piece") if product else "piece"
|
||
_print_item_line(p, item_name, item.quantity, sz_item, b_item, dot_fill, dot_sz, dot_bold, line_width, unit_type=unit_type)
|
||
if item.notes:
|
||
note_text = item.notes.upper() if c_note else item.notes
|
||
_apply_font(p, sz_note, b_note)
|
||
_raw_text(p, f"! {note_text}\n")
|
||
_reset_font(p)
|
||
if not compact:
|
||
p._raw(b'\n')
|
||
|
||
# Priceless ephemeral service entries
|
||
for entry in eph:
|
||
raw_name = entry['product'].name
|
||
item_name = raw_name.upper() if c_item else raw_name
|
||
_print_item_line(p, item_name, entry['quantity'], sz_item, b_item, dot_fill, dot_sz, dot_bold, line_width, unit_type='piece')
|
||
if not compact:
|
||
p._raw(b'\n')
|
||
|
||
# ── ITEMS section ────────────────────────────────────────────────────────
|
||
if has_both:
|
||
_divider(p, div, line_width)
|
||
p._raw(b'\x1b\x61\x01') # center
|
||
_apply_font(p, sz_item, b_item)
|
||
_raw_text(p, ("ITEMS\n" if c_item else "Items\n"))
|
||
_reset_font(p)
|
||
p._raw(b'\x1b\x61\x00')
|
||
|
||
# Load courses if enabled
|
||
courses_enabled = cfg.get("orders.courses_enabled", "false") == "true"
|
||
courses_raw = cfg.get("orders.courses", "[]")
|
||
try:
|
||
courses_list = json.loads(courses_raw) if courses_enabled else []
|
||
except Exception:
|
||
courses_list = []
|
||
|
||
_divider(p, div, line_width)
|
||
|
||
if courses_list:
|
||
# Group regular_items by course_id
|
||
course_groups = {}
|
||
for item in regular_items:
|
||
key = item.course_id if item.course_id is not None else None
|
||
if key not in course_groups:
|
||
course_groups[key] = []
|
||
course_groups[key].append(item)
|
||
|
||
# Build ordered list: courses in defined order, then None last
|
||
ordered_groups = []
|
||
for course in courses_list:
|
||
cid = course.get("id")
|
||
if cid in course_groups:
|
||
ordered_groups.append((course.get("name", f"Course {cid}"), course_groups[cid]))
|
||
if None in course_groups:
|
||
ordered_groups.append(("No course", course_groups[None]))
|
||
|
||
for group_name, group_items in ordered_groups:
|
||
# Print course section header
|
||
p._raw(b'\x1b\x61\x01') # center
|
||
_apply_font(p, sz_item, b_item)
|
||
header = f"-- {group_name.upper() if c_item else group_name} --"
|
||
_raw_text(p, header + "\n")
|
||
_reset_font(p)
|
||
p._raw(b'\x1b\x61\x00')
|
||
|
||
for item in group_items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
raw_name = product.name if product else f"Product #{item.product_id}"
|
||
if print_checkboxes:
|
||
raw_name = "[ ] " + raw_name
|
||
item_name = raw_name.upper() if c_item else raw_name
|
||
unit_type = (product.unit_type or "piece") if product else "piece"
|
||
|
||
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,
|
||
unit_type=unit_type,
|
||
)
|
||
|
||
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]
|
||
_apply_font(p, sz_ing, b_ing)
|
||
_raw_text(p, f"- ΧΩΡΙΣ: {' · '.join(names)}\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')
|
||
else:
|
||
for item in regular_items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
raw_name = product.name if product else f"Product #{item.product_id}"
|
||
if print_checkboxes:
|
||
raw_name = "[ ] " + raw_name
|
||
item_name = raw_name.upper() if c_item else raw_name
|
||
unit_type = (product.unit_type or "piece") if product else "piece"
|
||
|
||
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,
|
||
unit_type=unit_type,
|
||
)
|
||
|
||
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, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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, codepage_n: int = 29):
|
||
"""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, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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", []):
|
||
ut = item.get("unit_type", "piece")
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, _item_line(item["name"], item["quantity"], line_width=line_width, unit_type=ut) + "\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
qty_str = _qty_suffix(item["quantity"], ut).strip()
|
||
_raw_text(p, f" {item['unit_price']:.2f}e {qty_str} = {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, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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:
|
||
ut = item.get("unit_type", "piece")
|
||
qty_str = _qty_suffix(item["quantity"], ut).strip()
|
||
_raw_text(p, f" {_item_line(item['name'], item['quantity'], line_width=line_width - 2, unit_type=ut)}\n")
|
||
_raw_text(p, f" {item['unit_price']:.2f}e {qty_str} = {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:
|
||
ut = item.get("unit_type", "piece")
|
||
qty_str = _qty_suffix(item["quantity"], ut).strip()
|
||
_raw_text(p, f" {_item_line(item['name'], item['quantity'], line_width=line_width - 2, unit_type=ut)}\n")
|
||
_raw_text(p, f" {item['unit_price']:.2f}e {qty_str} = {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)
|
||
|
||
|
||
def print_cancellation_ticket(order_id: int, item_ids: List[int]):
|
||
"""
|
||
Background task: print a ΑΚΥΡΩΣΗ ticket to the same printer zones
|
||
as the cancelled items. Connects to its own DB session.
|
||
"""
|
||
db: Session = SessionLocal()
|
||
try:
|
||
_do_print_cancellation(order_id, item_ids, db)
|
||
except Exception as e:
|
||
logger.exception("Unexpected error in print_cancellation_ticket for order %s: %s", order_id, e)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _do_print_cancellation(order_id: int, item_ids: List[int], db: Session):
|
||
if _is_spoof_mode(db):
|
||
logger.info("Spoof printing ON — dropping cancellation ticket for order %s", order_id)
|
||
return
|
||
|
||
order = db.query(Order).filter(Order.id == order_id).first()
|
||
if not order:
|
||
return
|
||
|
||
items = db.query(OrderItem).filter(OrderItem.id.in_(item_ids)).all()
|
||
|
||
# Route by prep zones (same logic as normal ticket), keyed by (printer_id, zone_id).
|
||
# Cancellation always prints regardless of auto_print mode — if it would print normally,
|
||
# we need a cancellation ticket.
|
||
printer_job: dict[tuple[int, int], tuple[object, List[OrderItem]]] = {}
|
||
seen: set[tuple[int, int, int]] = set()
|
||
|
||
for item in items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
if not product:
|
||
continue
|
||
|
||
item_routed = False
|
||
for zone in product.prep_zones:
|
||
mode = zone.auto_print or 'none'
|
||
if mode == 'none':
|
||
continue
|
||
|
||
printers_to_use: List[object] = []
|
||
master_id = zone.master_printer_id
|
||
|
||
if mode in ('master', 'all'):
|
||
if master_id:
|
||
master_printer = db.query(Printer).filter(Printer.id == master_id, Printer.is_active == True).first()
|
||
if master_printer:
|
||
printers_to_use.append(master_printer)
|
||
elif zone.printers:
|
||
printers_to_use.append(zone.printers[0])
|
||
|
||
if mode == 'all':
|
||
for zp in zone.printers:
|
||
if zp.id != master_id and zp.is_active:
|
||
printers_to_use.append(zp)
|
||
|
||
for printer in printers_to_use:
|
||
key = (printer.id, zone.id, item.id)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
job_key = (printer.id, zone.id)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = (printer, [])
|
||
printer_job[job_key][1].append(item)
|
||
item_routed = True
|
||
|
||
# Fallback to legacy printer_zone_id
|
||
if not item_routed and product.printer_zone_id:
|
||
printer = db.query(Printer).filter(Printer.id == product.printer_zone_id, Printer.is_active == True).first()
|
||
if printer:
|
||
key = (printer.id, 0, item.id)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
job_key = (printer.id, 0)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = (printer, [])
|
||
printer_job[job_key][1].append(item)
|
||
|
||
div = load_divider_style(db)
|
||
|
||
for (printer, zone_items) in printer_job.values():
|
||
try:
|
||
p = _get_printer(printer.ip_address, printer.port, printer.codepage_n)
|
||
_print_cancel_ticket(p, order, zone_items, db, printer.line_width, div)
|
||
p.close()
|
||
except Exception as e:
|
||
logger.error("Cancellation print failed for printer %s: %s", printer.name, e)
|
||
|
||
|
||
def _print_cancel_ticket(p: Network, order: Order, items: List[OrderItem], db: Session, line_width: int, div: str):
|
||
cfg = _load_print_settings(db)
|
||
|
||
table_name = order.table.label or str(order.table.number) if order.table else str(order.table_id)
|
||
now_str = _greek_date(datetime.datetime.now(datetime.timezone.utc))
|
||
|
||
def _cancel_banner():
|
||
p._raw(b'\x1b\x61\x01') # center
|
||
p._raw(b'\x1b\x21\x30') # double height+width
|
||
p._raw(b'\x1b\x45\x01') # bold on
|
||
_raw_text(p, "*** AKYPΩΣH ***\n")
|
||
p._raw(b'\x1b\x45\x00')
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p, div, line_width)
|
||
|
||
_cancel_banner()
|
||
|
||
p._raw(b'\x1b\x61\x00')
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Παραγγελια: #{order.id}\n")
|
||
_raw_text(p, f"Τραπεζι: {table_name}\n")
|
||
_raw_text(p, f"Ωρα: {now_str}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p, div, line_width)
|
||
|
||
for item in items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
name = product.name if product else f"#{item.product_id}"
|
||
ut = (product.unit_type or "piece") if product else "piece"
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, _item_line(name, item.quantity, line_width, unit_type=ut) + "\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
|
||
_divider(p, div, line_width)
|
||
_cancel_banner()
|
||
|
||
p._raw(b'\n\n\n')
|
||
p.cut()
|
||
|
||
|
||
# ── Analytical report prints (products / categories / tables) ─────────────────
|
||
|
||
def print_products_report(ip: str, port: int, report: dict, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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_prep_zone_summary(ip: str, port: int, report: dict, codepage_n: int = 29):
|
||
"""Print a shopping-list summary for a prep zone."""
|
||
if is_spoof_mode():
|
||
logger.info("Spoof printing ON — dropping prep zone summary print")
|
||
return
|
||
try:
|
||
lw = report.get("line_width", LINE_WIDTH)
|
||
div = report.get("div_style", "dash")
|
||
zone_name = report.get("zone_name", "")
|
||
period_label = report.get("period_label", "")
|
||
items = report.get("items", []) # [{name, count, value}]
|
||
p = _get_printer(ip, port, codepage_n)
|
||
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, "ΖΩΝΗ ΠΡΟΕΤΟΙΜΑΣΙΑΣ\n")
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, zone_name + "\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p, div, lw)
|
||
|
||
p._raw(b'\x1b\x61\x00')
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Περιοδος: {period_label}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p, div, lw)
|
||
|
||
total_items = sum(x["count"] for x in items)
|
||
for item in items:
|
||
left = item["name"]
|
||
right = str(item["count"])
|
||
_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_items}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
|
||
p._raw(b'\n\n\n')
|
||
p.cut()
|
||
p.close()
|
||
except Exception as e:
|
||
logger.error("print_prep_zone_summary failed for %s:%s — %s", ip, port, e)
|
||
|
||
|
||
def print_categories_report(ip: str, port: int, report: dict, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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, codepage_n: int = 29):
|
||
"""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, codepage_n)
|
||
|
||
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, ephemeral_svc_entries: Optional[List[dict]] = None) -> 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, ephemeral_svc_entries=ephemeral_svc_entries)
|
||
|
||
|
||
def print_to_printer(order_id: int, item_ids: Optional[List[int]], printer_id: int, copies: int, db: Session) -> dict:
|
||
"""
|
||
Print specific items from an order directly to a named printer.
|
||
Used by KDS manual-print and auto-print features.
|
||
Returns { printer_name, success, error }.
|
||
"""
|
||
if _is_spoof_mode(db):
|
||
logger.info("Spoof printing ON — dropping manual print job for order %s printer %s", order_id, printer_id)
|
||
return {"printer_name": "spoof", "success": True, "error": None}
|
||
|
||
order = db.query(Order).filter(Order.id == order_id).first()
|
||
if not order:
|
||
return {"printer_name": f"#{printer_id}", "success": False, "error": "Order not found"}
|
||
|
||
printer = db.query(Printer).filter(Printer.id == printer_id, Printer.is_active == True).first()
|
||
if not printer:
|
||
return {"printer_name": f"#{printer_id}", "success": False, "error": "Printer not found or inactive"}
|
||
|
||
if item_ids is None:
|
||
items = db.query(OrderItem).filter(OrderItem.order_id == order_id, OrderItem.status != "cancelled").all()
|
||
else:
|
||
items = db.query(OrderItem).filter(OrderItem.id.in_(item_ids)).all()
|
||
copies = max(1, copies)
|
||
|
||
success = False
|
||
error_msg = None
|
||
try:
|
||
for _ in range(copies):
|
||
p = _get_printer(printer.ip_address, printer.port, printer.codepage_n)
|
||
_print_kitchen_ticket(p, order, items, db, printer.line_width)
|
||
p.close()
|
||
success = True
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.error("Manual 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(item_ids),
|
||
success=success,
|
||
error_message=error_msg,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
|
||
return {"printer_name": printer.name, "success": success, "error": error_msg}
|
||
|
||
|
||
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, ephemeral_svc_entries: Optional[List[dict]] = None) -> 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()
|
||
|
||
# Build (printer_id, copies, zone_id) → (items, zone) map respecting zone auto_print settings.
|
||
# zone_id is included in the key so that items from different zones that happen to share
|
||
# the same printer still get separate tickets with the correct zone-level formatting.
|
||
# Deduplication: (printer_id, copies, zone_id, item_id) prevents duplicate sends.
|
||
printer_job: dict[tuple[int, int, int], tuple[list, object]] = {}
|
||
seen: set[tuple[int, int, int, int]] = set()
|
||
unzoned: List[OrderItem] = []
|
||
printed_item_ids_no_print: set[int] = set() # intentional no-print zones
|
||
|
||
for item in items:
|
||
product = db.query(Product).filter(Product.id == item.product_id).first()
|
||
if not product:
|
||
unzoned.append(item)
|
||
continue
|
||
|
||
item_routed = False
|
||
for zone in product.prep_zones:
|
||
mode = zone.auto_print or 'none'
|
||
if mode == 'none':
|
||
continue
|
||
|
||
master_id = zone.master_printer_id
|
||
master_copies = max(1, zone.master_copies or 1)
|
||
secondary_copies = max(1, zone.secondary_copies or 1)
|
||
|
||
# Determine which printers to use
|
||
printers_to_use: List[tuple] = [] # (printer, copies)
|
||
|
||
if mode in ('master', 'all'):
|
||
if master_id:
|
||
master_printer = db.query(Printer).filter(Printer.id == master_id, Printer.is_active == True).first()
|
||
if master_printer:
|
||
printers_to_use.append((master_printer, master_copies))
|
||
elif zone.printers:
|
||
# Fallback: no master set, treat first printer as master
|
||
printers_to_use.append((zone.printers[0], master_copies))
|
||
|
||
if mode == 'all':
|
||
for zp in zone.printers:
|
||
if zp.id != master_id and zp.is_active:
|
||
printers_to_use.append((zp, secondary_copies))
|
||
|
||
for (printer, copies) in printers_to_use:
|
||
key = (printer.id, copies, zone.id, item.id)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
job_key = (printer.id, copies, zone.id)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = ([], zone)
|
||
printer_job[job_key][0].append(item)
|
||
item_routed = True
|
||
|
||
if not item_routed:
|
||
# If the product has prep zones but all are set to 'none', it's an intentional
|
||
# no-print choice — mark it printed immediately so the PWA doesn't show a warning.
|
||
if product.prep_zones and all((z.auto_print or 'none') == 'none' for z in product.prep_zones):
|
||
printed_item_ids_no_print.add(item.id)
|
||
continue
|
||
# Fall back to legacy printer_zone_id (single printer, 1 copy, no zone formatting)
|
||
if product.printer_zone_id:
|
||
printer = db.query(Printer).filter(Printer.id == product.printer_zone_id, Printer.is_active == True).first()
|
||
if printer:
|
||
key = (printer.id, 1, 0, item.id)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
job_key = (printer.id, 1, 0)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = ([], None)
|
||
printer_job[job_key][0].append(item)
|
||
else:
|
||
unzoned.append(item)
|
||
else:
|
||
unzoned.append(item)
|
||
|
||
if unzoned:
|
||
logger.warning("order %s has %d item(s) with no prep zone/printer — skipped", order_id, len(unzoned))
|
||
|
||
# Route ephemeral service entries into the same printer jobs.
|
||
# Each entry goes to every printer job that already exists (broad broadcast),
|
||
# OR to its own prep-zone printers if no regular jobs exist for that printer.
|
||
# We track which entries land in which job_key so each ticket gets the right subset.
|
||
job_svc_entries: dict[tuple, list] = {} # job_key → list of {product, quantity}
|
||
|
||
for entry in (ephemeral_svc_entries or []):
|
||
product = entry['product']
|
||
entry_routed = False
|
||
for zone in product.prep_zones:
|
||
mode_z = zone.auto_print or 'none'
|
||
if mode_z == 'none':
|
||
continue
|
||
master_id = zone.master_printer_id
|
||
master_copies = max(1, zone.master_copies or 1)
|
||
secondary_copies = max(1, zone.secondary_copies or 1)
|
||
printers_to_use = []
|
||
if mode_z in ('master', 'all'):
|
||
if master_id:
|
||
pr = db.query(Printer).filter(Printer.id == master_id, Printer.is_active == True).first()
|
||
if pr:
|
||
printers_to_use.append((pr, master_copies, zone))
|
||
elif zone.printers:
|
||
printers_to_use.append((zone.printers[0], master_copies, zone))
|
||
if mode_z == 'all':
|
||
for zp in zone.printers:
|
||
if zp.id != master_id and zp.is_active:
|
||
printers_to_use.append((zp, secondary_copies, zone))
|
||
for (pr, copies, z) in printers_to_use:
|
||
job_key = (pr.id, copies, z.id)
|
||
# Ensure job exists even if no regular items route here
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = ([], z)
|
||
if job_key not in job_svc_entries:
|
||
job_svc_entries[job_key] = []
|
||
job_svc_entries[job_key].append(entry)
|
||
entry_routed = True
|
||
# Fallback: legacy printer_zone_id
|
||
if not entry_routed and getattr(product, 'printer_zone_id', None):
|
||
pr = db.query(Printer).filter(Printer.id == product.printer_zone_id, Printer.is_active == True).first()
|
||
if pr:
|
||
job_key = (pr.id, 1, 0)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = ([], None)
|
||
if job_key not in job_svc_entries:
|
||
job_svc_entries[job_key] = []
|
||
job_svc_entries[job_key].append(entry)
|
||
|
||
printed_item_ids: set[int] = set()
|
||
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
|
||
for (printer_id, copies, zone_id), (job_items, job_zone) in printer_job.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
|
||
|
||
job_key = (printer_id, copies, zone_id)
|
||
job_eph = job_svc_entries.get(job_key, [])
|
||
item_ids_json = json.dumps([i.id for i in job_items])
|
||
|
||
# Find or create the PrintJob for this (order, printer, zone) combination
|
||
pj = db.query(PrintJob).filter(
|
||
PrintJob.order_id == order_id,
|
||
PrintJob.printer_id == printer_id,
|
||
PrintJob.zone_id == (zone_id if zone_id != 0 else None),
|
||
PrintJob.status == "pending",
|
||
).first()
|
||
if pj is None:
|
||
pj = PrintJob(
|
||
order_id=order_id,
|
||
printer_id=printer_id,
|
||
zone_id=zone_id if zone_id != 0 else None,
|
||
item_ids=item_ids_json,
|
||
copies=copies,
|
||
status="pending",
|
||
first_attempted_at=now,
|
||
)
|
||
db.add(pj)
|
||
db.flush()
|
||
|
||
pj.last_attempted_at = now
|
||
pj.retry_count = (pj.retry_count or 0) + 1
|
||
|
||
success = False
|
||
error_msg = None
|
||
try:
|
||
for _ in range(copies):
|
||
p = _get_printer(printer.ip_address, printer.port, printer.codepage_n)
|
||
_print_kitchen_ticket(p, order, job_items, db, printer.line_width, zone=job_zone, ephemeral_svc_entries=job_eph)
|
||
p.close()
|
||
success = True
|
||
pj.status = "success"
|
||
pj.succeeded_at = now
|
||
for item in job_items:
|
||
printed_item_ids.add(item.id)
|
||
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=item_ids_json,
|
||
success=success,
|
||
error_message=error_msg,
|
||
)
|
||
db.add(log)
|
||
results.append({"printer_name": printer.name, "success": success, "error": error_msg})
|
||
|
||
# Mark all successfully printed items (and intentional no-prints) in one commit
|
||
all_done_ids = printed_item_ids | printed_item_ids_no_print
|
||
if all_done_ids:
|
||
for item in items:
|
||
if item.id in all_done_ids:
|
||
item.printed = True
|
||
db.commit()
|
||
|
||
return results
|
||
|
||
|
||
def print_service_items_sync(order: Order, service_entries: List[dict], db: Session) -> List[dict]:
|
||
"""
|
||
Print service items (cutlery, glasses, etc.) that are NOT persisted as order items.
|
||
service_entries: list of { product, quantity } dicts.
|
||
Routes to prep zones defined on the product; groups by printer/zone.
|
||
Returns per-printer result dicts like route_and_print_sync.
|
||
"""
|
||
if _is_spoof_mode(db):
|
||
logger.info("Spoof printing ON — dropping service item print for order %s", order.id)
|
||
return [{"printer_name": "spoof", "success": True, "error": None}]
|
||
|
||
if not service_entries:
|
||
return []
|
||
|
||
results = []
|
||
cfg = _load_print_settings(db)
|
||
mode = cfg.get("print.ticket_mode", "detailed")
|
||
compact = (mode == "compact")
|
||
div = cfg.get("print.divider_style", "dash")
|
||
|
||
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))
|
||
customer_count = getattr(order, 'customer_count', None)
|
||
|
||
# Group entries by (printer_id, copies, zone_id) → list of (product, quantity)
|
||
printer_job: dict = {}
|
||
seen: set = set()
|
||
|
||
for entry in service_entries:
|
||
product = entry['product']
|
||
quantity = entry['quantity']
|
||
|
||
for zone in product.prep_zones:
|
||
mode_z = zone.auto_print or 'none'
|
||
if mode_z == 'none':
|
||
continue
|
||
|
||
master_id = zone.master_printer_id
|
||
master_copies = max(1, zone.master_copies or 1)
|
||
secondary_copies = max(1, zone.secondary_copies or 1)
|
||
|
||
printers_to_use = []
|
||
if mode_z in ('master', 'all'):
|
||
if master_id:
|
||
pr = db.query(Printer).filter(Printer.id == master_id, Printer.is_active == True).first()
|
||
if pr:
|
||
printers_to_use.append((pr, master_copies, zone))
|
||
elif zone.printers:
|
||
printers_to_use.append((zone.printers[0], master_copies, zone))
|
||
if mode_z == 'all':
|
||
for zp in zone.printers:
|
||
if zp.id != master_id and zp.is_active:
|
||
printers_to_use.append((zp, secondary_copies, zone))
|
||
|
||
for (pr, copies, z) in printers_to_use:
|
||
job_key = (pr.id, copies, z.id)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = {'printer': pr, 'copies': copies, 'zone': z, 'entries': []}
|
||
printer_job[job_key]['entries'].append({'product': product, 'quantity': quantity})
|
||
|
||
# Fallback: legacy printer_zone_id
|
||
if not product.prep_zones and product.printer_zone_id:
|
||
pr = db.query(Printer).filter(Printer.id == product.printer_zone_id, Printer.is_active == True).first()
|
||
if pr:
|
||
job_key = (pr.id, 1, 0)
|
||
if job_key not in printer_job:
|
||
printer_job[job_key] = {'printer': pr, 'copies': 1, 'zone': None, 'entries': []}
|
||
printer_job[job_key]['entries'].append({'product': product, 'quantity': quantity})
|
||
|
||
for job_key, job in printer_job.items():
|
||
pr = job['printer']
|
||
copies = job['copies']
|
||
entries = job['entries']
|
||
|
||
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"])
|
||
|
||
success = False
|
||
error_msg = None
|
||
try:
|
||
for _ in range(copies):
|
||
p = _get_printer(pr.ip_address, pr.port, pr.codepage_n)
|
||
|
||
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}"
|
||
if customer_count:
|
||
header += f" | {customer_count}ατ."
|
||
_raw_text(p, (header.upper() if c_ord else header) + "\n")
|
||
_reset_font(p)
|
||
_divider(p, div, pr.line_width or LINE_WIDTH)
|
||
else:
|
||
_print_line(p, f"Παραγγελια #{order.id}", sz_ord, b_ord, c_ord, align=b'\x1b\x61\x01')
|
||
_divider(p, div, pr.line_width or 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")
|
||
if customer_count:
|
||
_raw_text(p, ("ΑΤΟΜΑ:" if c_meta else "Ατομα:") + f" {customer_count}\n")
|
||
_reset_font(p)
|
||
_divider(p, div, pr.line_width or LINE_WIDTH)
|
||
|
||
for entry in entries:
|
||
name = entry['product'].name
|
||
qty = entry['quantity']
|
||
qty_str = f"{int(qty)}x" if qty == int(qty) else f"{qty}x"
|
||
_apply_font(p, sz_item, b_item)
|
||
line = f"{qty_str} {name}"
|
||
_raw_text(p, (line.upper() if c_item else line) + "\n")
|
||
_reset_font(p)
|
||
|
||
p._raw(b'\n\n\n')
|
||
p._raw(b'\x1d\x56\x41\x03') # cut
|
||
p.close()
|
||
success = True
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.error("Service print failed for printer %s (%s:%s): %s", pr.name, pr.ip_address, pr.port, e)
|
||
|
||
results.append({"printer_name": pr.name, "success": success, "error": error_msg})
|
||
|
||
return results
|
||
|
||
|
||
# ── Auto-retry background thread ──────────────────────────────────────────────
|
||
|
||
import threading as _threading
|
||
import time as _time
|
||
|
||
_RETRY_INTERVAL = 8 # seconds between retry sweeps
|
||
|
||
def _retry_sweep():
|
||
"""
|
||
Find all pending PrintJob rows whose order items are still active and unprinted,
|
||
and retry each one. Each job gets its own fresh DB session to avoid SQLAlchemy
|
||
identity-map caching stale printed/status values across iterations.
|
||
"""
|
||
# Collect pending job IDs first, then process each in isolation
|
||
db: Session = SessionLocal()
|
||
try:
|
||
if _is_spoof_mode(db):
|
||
return
|
||
pending_job_ids = [
|
||
pj.id for pj in db.query(PrintJob.id).filter(PrintJob.status == "pending").all()
|
||
]
|
||
except Exception as e:
|
||
logger.exception("Error fetching pending print jobs: %s", e)
|
||
return
|
||
finally:
|
||
db.close()
|
||
|
||
for job_id in pending_job_ids:
|
||
# Fresh session per job — no stale identity-map entries
|
||
db: Session = SessionLocal()
|
||
try:
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
|
||
pj = db.query(PrintJob).filter(PrintJob.id == job_id).first()
|
||
if not pj or pj.status != "pending":
|
||
continue
|
||
|
||
order = db.query(Order).filter(Order.id == pj.order_id).first()
|
||
if not order or order.status in ("closed", "cancelled"):
|
||
pj.status = "cancelled"
|
||
pj.cancelled_at = now
|
||
pj.cancel_reason = "order_closed"
|
||
db.commit()
|
||
continue
|
||
|
||
item_ids = json.loads(pj.item_ids or "[]")
|
||
active_unprinted = db.query(OrderItem).filter(
|
||
OrderItem.id.in_(item_ids),
|
||
OrderItem.status == "active",
|
||
OrderItem.printed == False, # noqa: E712
|
||
).all()
|
||
|
||
if not active_unprinted:
|
||
# All items already printed or removed — job is done
|
||
pj.status = "success"
|
||
pj.succeeded_at = now
|
||
db.commit()
|
||
continue
|
||
|
||
printer = db.query(Printer).filter(Printer.id == pj.printer_id, Printer.is_active == True).first() # noqa: E712
|
||
if not printer:
|
||
continue
|
||
|
||
pj.last_attempted_at = now
|
||
pj.retry_count = (pj.retry_count or 0) + 1
|
||
|
||
success = False
|
||
error_msg = None
|
||
try:
|
||
for _ in range(pj.copies or 1):
|
||
p = _get_printer(printer.ip_address, printer.port, printer.codepage_n)
|
||
_print_kitchen_ticket(p, order, active_unprinted, db, printer.line_width)
|
||
p.close()
|
||
success = True
|
||
pj.status = "success"
|
||
pj.succeeded_at = now
|
||
for item in active_unprinted:
|
||
item.printed = True
|
||
logger.info("PrintJob %s succeeded on retry %d", pj.id, pj.retry_count)
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.warning("PrintJob %s retry %d failed: %s", pj.id, pj.retry_count, e)
|
||
|
||
db.add(PrintLog(
|
||
order_id=pj.order_id,
|
||
printer_id=pj.printer_id,
|
||
item_ids=pj.item_ids,
|
||
success=success,
|
||
error_message=error_msg,
|
||
))
|
||
db.commit()
|
||
|
||
except Exception as e:
|
||
logger.exception("Unexpected error processing PrintJob %s: %s", job_id, e)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _retry_loop():
|
||
while True:
|
||
_time.sleep(_RETRY_INTERVAL)
|
||
_retry_sweep()
|
||
|
||
|
||
def start_print_retry_thread():
|
||
t = _threading.Thread(target=_retry_loop, daemon=True, name="print-retry")
|
||
t.start()
|
||
logger.info("Print auto-retry thread started (interval=%ds, no retry cap)", _RETRY_INTERVAL)
|