#!/usr/bin/env python3 """RAS IRIS Edge - agente local de impressao Gastronomia (stdlib only).""" from __future__ import annotations import json import os import socket import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime from pathlib import Path VERSION = "printer-agent-1.1.0" DEFAULT_API = "https://api.rasiris.com.br" def load_config() -> dict: candidates = [ Path(os.environ.get("RAS_IRIS_EDGE_CONFIG", "")), Path.cwd() / "ras-iris-edge.json", Path(__file__).resolve().with_name("ras-iris-edge.json"), Path("/etc/ras-iris-edge/ras-iris-edge.json"), ] config = {} for path in candidates: if str(path) and path.is_file(): with path.open("r", encoding="utf-8") as source: config = json.load(source) break aliases = { "api_url": "RAS_IRIS_EDGE_API", "node_id": "RAS_IRIS_EDGE_NODE_ID", "edge_token": "RAS_IRIS_EDGE_TOKEN", "poll_seconds": "RAS_IRIS_EDGE_POLL_SECONDS", } for key, env_name in aliases.items(): if os.environ.get(env_name): config[key] = os.environ[env_name] config["api_url"] = str(config.get("api_url") or DEFAULT_API).rstrip("/") config["poll_seconds"] = max(2, min(int(config.get("poll_seconds") or 4), 60)) if not config.get("node_id") or not config.get("edge_token"): raise RuntimeError("Salve ras-iris-edge.json gerado na administracao antes de iniciar o agente.") return config def request_json(config: dict, method: str, path: str, payload: dict | None = None) -> dict: body = json.dumps(payload).encode("utf-8") if payload is not None else None request = urllib.request.Request( config["api_url"] + path, data=body, method=method, headers={ "Accept": "application/json", "Content-Type": "application/json", "User-Agent": f"RAS-IRIS-Edge/{VERSION}", "X-RAS-IRIS-Edge-Node": str(config["node_id"]), "X-RAS-IRIS-Edge-Token": str(config["edge_token"]), }, ) with urllib.request.urlopen(request, timeout=20) as response: data = json.loads(response.read().decode("utf-8")) if data.get("ok") is False: raise RuntimeError(data.get("mensagem") or "A API recusou a operacao.") return data def fit(text: object, width: int) -> str: value = " ".join(str(text or "").split()) return value[:width] def ticket_bytes(job: dict) -> bytes: payload = job.get("payload") or {} printer = payload.get("printer") or {} order = payload.get("order") or {} width = 32 if int(printer.get("paper_width_mm") or 80) == 58 else 48 lines = [ "RAS IRIS GASTRONOMIA".center(width), ("TESTE DE IMPRESSAO" if payload.get("kind") == "test" else f"PEDIDO {order.get('number') or '-'}").center(width), "=" * width, f"Destino: {printer.get('name') or '-'}", f"Origem: {order.get('origin') or '-'}", f"Cliente/Mesa: {order.get('customer') or '-'}", f"Horario: {order.get('created_at') or datetime.now().isoformat(timespec='seconds')}", "-" * width, ] for item in payload.get("items") or []: quantity = item.get("quantity") or 1 lines.append(f"{quantity}x {fit(item.get('name'), max(8, width - 5))}") if item.get("notes"): lines.append(f" OBS: {fit(item.get('notes'), max(8, width - 7))}") if order.get("notes"): lines.extend(["-" * width, f"OBS. PEDIDO: {fit(order.get('notes'), max(8, width - 13))}"]) lines.extend(["=" * width, "Impresso automaticamente pelo RAS IRIS", "", ""]) text = "\n".join(lines).encode(str(printer.get("encoding") or "cp850"), errors="replace") output = b"\x1b\x40" + text if printer.get("beep"): output += b"\x1b\x42\x02\x02" if printer.get("cut_paper", True): output += b"\x1d\x56\x00" return output def send_windows_raw(printer_name: str, raw: bytes) -> None: """Envia ESC/POS bruto à fila USB instalada no Windows sem dependências extras.""" import ctypes from ctypes import wintypes class DocInfo(ctypes.Structure): _fields_ = [ ("pDocName", wintypes.LPWSTR), ("pOutputFile", wintypes.LPWSTR), ("pDatatype", wintypes.LPWSTR), ] spooler = ctypes.WinDLL("winspool.drv", use_last_error=True) spooler.OpenPrinterW.argtypes = [wintypes.LPWSTR, ctypes.POINTER(wintypes.HANDLE), wintypes.LPVOID] spooler.OpenPrinterW.restype = wintypes.BOOL spooler.StartDocPrinterW.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(DocInfo)] spooler.StartDocPrinterW.restype = wintypes.DWORD spooler.StartPagePrinter.argtypes = [wintypes.HANDLE] spooler.StartPagePrinter.restype = wintypes.BOOL spooler.WritePrinter.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)] spooler.WritePrinter.restype = wintypes.BOOL spooler.EndPagePrinter.argtypes = [wintypes.HANDLE] spooler.EndDocPrinter.argtypes = [wintypes.HANDLE] spooler.ClosePrinter.argtypes = [wintypes.HANDLE] handle = wintypes.HANDLE() if not spooler.OpenPrinterW(printer_name, ctypes.byref(handle), None): raise ctypes.WinError(ctypes.get_last_error()) document_started = False page_started = False try: document = DocInfo("RAS IRIS Gastronomia", None, "RAW") if not spooler.StartDocPrinterW(handle, 1, ctypes.byref(document)): raise ctypes.WinError(ctypes.get_last_error()) document_started = True if not spooler.StartPagePrinter(handle): raise ctypes.WinError(ctypes.get_last_error()) page_started = True buffer = ctypes.create_string_buffer(raw) written = wintypes.DWORD() if not spooler.WritePrinter(handle, buffer, len(raw), ctypes.byref(written)): raise ctypes.WinError(ctypes.get_last_error()) if written.value != len(raw): raise OSError(f"A fila do Windows recebeu {written.value} de {len(raw)} bytes.") finally: if page_started: spooler.EndPagePrinter(handle) if document_started: spooler.EndDocPrinter(handle) spooler.ClosePrinter(handle) def send_system_raw(printer_name: str, raw: bytes) -> None: if os.name == "nt": send_windows_raw(printer_name, raw) return subprocess.run( ["lp", "-d", printer_name, "-o", "raw"], input=raw, check=True, timeout=30, ) def send_to_printer(job: dict) -> None: printer = (job.get("payload") or {}).get("printer") or {} raw = ticket_bytes(job) copies = max(1, min(int(job.get("copies") or 1), 5)) connection = printer.get("connection_type") for _ in range(copies): if connection == "network": with socket.create_connection((str(printer["host"]), int(printer.get("port") or 9100)), timeout=12) as stream: stream.sendall(raw) elif connection == "system": send_system_raw(str(printer["system_name"]), raw) else: raise RuntimeError("Tipo de conexao da impressora nao reconhecido.") def report(config: dict, job_id: str, status: str, error: str | None = None) -> None: payload = {"status": status} if error: payload["error"] = error[:1000] request_json(config, "POST", f"/api/gastronomia/edge/print-jobs/{job_id}/result", payload) def run() -> None: config = load_config() print(f"RAS IRIS Edge {VERSION} iniciado. API: {config['api_url']}", flush=True) failures = 0 while True: try: response = request_json(config, "GET", "/api/gastronomia/edge/print-jobs") failures = 0 for job in response.get("print_jobs") or []: try: send_to_printer(job) report(config, str(job["id"]), "printed") print(f"Impresso: {job['id']}", flush=True) except Exception as exc: # hardware/driver errors must return to the audit queue message = f"{type(exc).__name__}: {exc}" try: report(config, str(job["id"]), "failed", message) finally: print(f"Falha de impressao {job.get('id')}: {message}", file=sys.stderr, flush=True) except (urllib.error.URLError, urllib.error.HTTPError, RuntimeError, TimeoutError, OSError) as exc: failures = min(failures + 1, 6) print(f"Conexao indisponivel: {type(exc).__name__}: {exc}", file=sys.stderr, flush=True) time.sleep(min(config["poll_seconds"] * (2 ** failures), 60)) if __name__ == "__main__": try: run() except (KeyboardInterrupt, SystemExit): print("RAS IRIS Edge encerrado.") except Exception as exc: print(f"Nao foi possivel iniciar: {exc}", file=sys.stderr) raise SystemExit(1)