227 lines
7.7 KiB
Python
227 lines
7.7 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 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()
|