""" 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 models.printer import Printer from models.product import Product 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)) def _divider(p: Network): p._raw(b'\x1b\x61\x00') p._raw(_gr("-" * LINE_WIDTH + "\n")) def _item_line(name: str, qty: int) -> str: """Build a dot-leader line: 'Club Sandwich . . . . 1' at 48 chars.""" qty_str = str(qty) gap = LINE_WIDTH - len(name) - len(qty_str) if gap < 3: return f"{name} {qty_str}" dots = (". " * ((gap // 2) + 1))[:gap] return f"{name}{dots}{qty_str}" 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 send_test_print(ip: str, port: int, name: str) -> Tuple[bool, str]: 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 = datetime.datetime.now().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) # ── Receipt formatting ─────────────────────────────────────────────────────── def _print_kitchen_ticket(p: Network, order: Order, items: List[OrderItem], db: Session): # Header p._raw(b'\x1b\x61\x01') p._raw(b'\x1b\x21\x38') # bold + double height + double width _raw_text(p, f"Παραγγελια #{order.id}\n") p._raw(b'\x1b\x21\x00') _divider(p) # Meta p._raw(b'\x1b\x61\x00') p._raw(b'\x1b\x21\x10') # double height only — keeps 48-char width now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") _raw_text(p, f"Date: {now}\n") _raw_text(p, f"Table: {order.table_id}\n") _raw_text(p, f"Waiter: {order.opened_by}\n") p._raw(b'\x1b\x21\x00') _divider(p) # Items for item in items: product = db.query(Product).filter(Product.id == item.product_id).first() name = product.name if product else f"Product #{item.product_id}" p._raw(b'\x1b\x21\x10') p._raw(b'\x1b\x45\x01') # bold on _raw_text(p, _item_line(name, item.quantity) + "\n") p._raw(b'\x1b\x45\x00') # bold off if item.removed_ingredients: try: removed_ids = json.loads(item.removed_ingredients) if removed_ids: _raw_text(p, f" - χωρις: {', '.join(str(i) for i in removed_ids)}\n") except (json.JSONDecodeError, TypeError): pass if item.selected_options: try: option_ids = json.loads(item.selected_options) if option_ids: _raw_text(p, f" + επιλογες: {', '.join(str(i) for i in option_ids)}\n") except (json.JSONDecodeError, TypeError): pass if item.notes: _raw_text(p, f" (i) {item.notes}\n") p._raw(b'\x1b\x21\x00') _divider(p) if order.notes: p._raw(b'\x1b\x21\x30') _raw_text(p, "Σημειωσεις:\n") p._raw(b'\x1b\x21\x10') _raw_text(p, f"{order.notes}\n") p._raw(b'\x1b\x21\x00') _divider(p) 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') p.cut() # ── 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: 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 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) continue success = False error_msg = None try: p = _get_printer(printer.ip_address, printer.port) _print_kitchen_ticket(p, order, zone_items, db) p.close() success = True # Mark items as printed 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() except Exception as e: logger.exception("Unexpected error in route_and_print for order %s: %s", order_id, e) finally: db.close()