507 lines
18 KiB
Python
507 lines
18 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
|
||
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))
|
||
|
||
|
||
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 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_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 = 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()
|
||
|
||
|
||
# ── 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)
|
||
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 print_printer_report(ip: str, port: int, report: dict, mode: str):
|
||
"""Print a per-printer totals report. mode='simple'|'extensive'."""
|
||
if is_spoof_mode():
|
||
logger.info("Spoof printing ON — dropping printer 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['printer_name']}\n")
|
||
_raw_text(p, f"Απο: {report['from_dt']}\n")
|
||
_raw_text(p, f"Εως: {report['to_dt']}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p)
|
||
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, f"Εργασιες εκτ.: {report['print_jobs']}\n")
|
||
_raw_text(p, f"Παραγγελιες: {report['orders']}\n")
|
||
_raw_text(p, f"Αντικειμενα: {report['items']}\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
_divider(p)
|
||
|
||
p._raw(b'\x1b\x61\x01')
|
||
p._raw(b'\x1b\x21\x30')
|
||
_raw_text(p, f"ΣΥΝΟΛΟ: {report['total']:.2f}e\n")
|
||
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"]:
|
||
# Header line: "HH:MM - TABLE . . . . 9.99e"
|
||
prefix = f"{od['time']} - {od['table']}"
|
||
value = f"{od['total']:.2f}e"
|
||
gap = LINE_WIDTH - len(prefix) - len(value)
|
||
if gap < 3:
|
||
header_line = f"{prefix} {value}"
|
||
else:
|
||
dots = (". " * ((gap // 2) + 1))[:gap]
|
||
header_line = f"{prefix}{dots}{value}"
|
||
p._raw(b'\x1b\x45\x01')
|
||
_raw_text(p, header_line + "\n")
|
||
p._raw(b'\x1b\x45\x00')
|
||
# Indented items
|
||
for item in od.get("items", []):
|
||
_raw_text(p, f" {item['quantity']} x {item['name']}\n")
|
||
|
||
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):
|
||
"""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)
|
||
|
||
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)
|
||
|
||
for item in receipt.get("items", []):
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, _item_line(item["name"], item["quantity"]) + "\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)
|
||
|
||
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)
|
||
|
||
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):
|
||
"""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)
|
||
|
||
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)
|
||
|
||
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'])}\n")
|
||
_raw_text(p, f" {item['unit_price']:.2f}e x{item['quantity']} = {item['total']:.2f}e\n")
|
||
_divider(p)
|
||
|
||
if paid_items:
|
||
p._raw(b'\x1b\x21\x10')
|
||
_raw_text(p, "ΠΛΗΡΩΜΕΝΑ:\n")
|
||
p._raw(b'\x1b\x21\x00')
|
||
for item in paid_items:
|
||
_raw_text(p, f" {_item_line(item['name'], item['quantity'])}\n")
|
||
_raw_text(p, f" {item['unit_price']:.2f}e x{item['quantity']} = {item['total']:.2f}e\n")
|
||
_divider(p)
|
||
|
||
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)
|
||
|
||
|
||
# ── 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)
|
||
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
|