#!/usr/bin/env python3 """ SORP Print Agent — lokálny tlačový agent pre softvérovú on-line registračnú pokladnicu sorp.verteco.sk (FR SR 485018/2026, nález N1: § 5 ods. 4 písm. d). Účel: originál pokladničného dokladu sa musí vytlačiť AUTOMATICKY, bez zásahu obsluhy a bez tlačového dialógu operačného systému, aj na tlačiarňach, ktoré prehliadač nevie ovládať priamo (napr. EPSON TM-T20III cez USB s ovládačom Windows). Agent beží na klientskom zariadení predávajúceho, počúva VÝHRADNE na 127.0.0.1 a od webovej aplikácie SORP (kasa.verteco.sk) preberá hotový tlačový výstup, ktorý je už uložený v CHDÚ (ESC/POS raster, resp. PDF), a odovzdá ho tlačovému subsystému OS ako RAW úlohu — bez zobrazenia dialógu. Agent NEMENÍ obsah dokladu, nevytvára žiadne fiškálne údaje a nemá prístup k účtu predávajúceho — je to výstupný kanál, rovnako ako Bluetooth/USB. Protokol (záväzný kontrakt s webovou aplikáciou, v1.0.0; polia označené + sú aditívne, webová aplikácia bez ich znalosti ich ignoruje): GET /status[?printer=] -> {"ok":true,"agent":"sorp-print-agent","version":..., "platform":"windows|darwin|linux","default_printer":..., + "printer":,"printer_status":"ready|offline| paper_out|error|paused|not_found|unknown"} GET /printers -> {"ok":true,"printers":[{"name":...,"default":bool,+"status":...}]} POST /print <- {"format":"escpos"|"pdf","data":"","printer":...|null, "job_id":"","width_dots":384|576|null} -> {"ok":true,"job_id":...,"printer":...,"bytes":N, "spooler_job_id":...|null} (+ "duplicate":true pri opakovaní) -> 503 {"ok":false,"error":"printer_offline",...} ak je tlačiareň vypnutá/odpojená/pozastavená alebo úloha neopustila tlačový rad (úloha sa z radu odstráni, +"job_cancelled":bool) OPTIONS * -> 204 (CORS preflight pre povolené originy) Každý ne-OPTIONS request musí niesť hlavičku X-Print-Agent-Token (inak 401). Odpoveď 200 na /print znamená, že tlačový subsystém OS úlohu ODOVZDAL tlačiarni (úloha opustila tlačový rad alebo má stav vytlačené), nie iba, že ju zaradil do radu. Ak tlačiareň úlohu neprevezme (vypnutá, odpojený kábel, pozastavený rad, chyba), agent úlohu z radu odstráni a vráti 503 printer_offline, takže webová aplikácia potvrdenie tlače nepošle a doklad zostane v stave „čaká na tlač". Fyzický koniec papiera pri jednosmernej RAW úlohe agent nezistí, ak ho ovládač nepremietne do stavu tlačiarne alebo úlohy. Výhradne štandardná knižnica Pythonu 3.10+ (odporúčané 3.12). Windows RAW tlač cez ctypes + winspool.drv (bez pywin32); macOS/Linux cez CUPS (`lp`, `lpstat`). Použitie: python sorp_print_agent.py # spustí agenta (server) python sorp_print_agent.py --list-printers # vypíše tlačiarne OS python sorp_print_agent.py --print-test "EPSON TM-T20III Receipt" python sorp_print_agent.py --show-token python sorp_print_agent.py --install-autostart # spúšťať po prihlásení Konfigurácia: ~/.sorp-print-agent/config.json (vytvorí sa pri prvom spustení). """ from __future__ import annotations import argparse import base64 import binascii import hmac import json import logging import os import re import secrets import shutil import subprocess import sys import tempfile import threading import time import uuid from collections.abc import Callable from dataclasses import asdict, dataclass, field from datetime import datetime from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any AGENT_NAME = "sorp-print-agent" AGENT_VERSION = "1.0.0" PROTOCOL_VERSION = "1.0.0" DEFAULT_PORT = 47831 BIND_HOST = "127.0.0.1" # nikdy nič iné — agent je dostupný len z toho istého zariadenia MAX_BODY_BYTES = 8 * 1024 * 1024 # 8 MB IDEMPOTENCY_TTL_S = 60.0 TOKEN_HEADER = "X-Print-Agent-Token" # Originy nasadenia webovej aplikácie (.do/app.yaml CORS_ALLOWED_ORIGINS + testovacia # inštancia). Vývojové lokálne originy sa pridávajú len pri behu zo skriptu, nie # v skompilovanej binárke (PyInstaller, sys.frozen) u predávajúceho. DEFAULT_ALLOWED_ORIGINS = [ "https://kasa.verteco.sk", "https://test.kasa.verteco.sk", "https://app.superkasa.sk", ] if not getattr(sys, "frozen", False): DEFAULT_ALLOWED_ORIGINS += ["http://localhost:5173", "http://127.0.0.1:5173"] ALLOWED_FORMATS = ("escpos", "pdf") ALLOWED_WIDTHS = (384, 576) CONFIG_DIR_NAME = ".sorp-print-agent" # Čakanie na odovzdanie úlohy tlačiarni (musí byť kratšie než PRINT_TIMEOUT_MS = 30 s # webovej aplikácie): úloha, ktorá do JOB_WAIT_S neopustí tlačový rad ani nezačne # tlačiť, sa zruší; úloha v stave „tlačí sa" (port monitor odovzdáva dáta pomalej # tlačiarni) smie bežať až JOB_PRINTING_WAIT_S. JOB_WAIT_S = 20.0 JOB_PRINTING_WAIT_S = 90.0 JOB_POLL_S = 0.25 JOB_CANCEL_WAIT_S = 3.0 # Stavy tlačiarne v /status a /printers (spoločné pre Windows aj CUPS) PRINTER_STATE_READY = "ready" PRINTER_STATE_OFFLINE = "offline" PRINTER_STATE_PAPER_OUT = "paper_out" PRINTER_STATE_ERROR = "error" PRINTER_STATE_PAUSED = "paused" PRINTER_STATE_NOT_FOUND = "not_found" PRINTER_STATE_UNKNOWN = "unknown" PRINTER_STATES_OK = (PRINTER_STATE_READY, PRINTER_STATE_UNKNOWN) # Stavy úlohy v tlačovom rade pre wait_for_spool_job (nezávislé od OS) JOB_STATE_DONE = "done" # úloha opustila rad alebo má stav vytlačené JOB_STATE_PENDING = "pending" # čaká v rade / spooluje JOB_STATE_PRINTING = "printing" # port monitor odovzdáva dáta tlačiarni JOB_ERROR_STATES = ("error", "offline", "paper_out", "paused", "blocked", "user_intervention", "cancelled") # Známe zástupné tokeny (config.example.json, dokumentácia) sa nikdy neprijmú ako # platné — agent si vygeneruje nový a vypíše ho obsluhe. PLACEHOLDER_TOKENS = frozenset({ "0123456789abcdef0123456789abcdef", "00000000000000000000000000000000", "ffffffffffffffffffffffffffffffff", "deadbeefdeadbeefdeadbeefdeadbeef", }) ESC = b"\x1b" GS = b"\x1d" log = logging.getLogger(AGENT_NAME) # --------------------------------------------------------------------------- # Chyby # --------------------------------------------------------------------------- class PrintAgentError(Exception): """Základ chýb agenta; `status` je HTTP kód, `code` krátky strojový kód, `details` sú doplňujúce polia do JSON odpovede (napr. spooler_job_id).""" status = HTTPStatus.INTERNAL_SERVER_ERROR code = "internal_error" def __init__( self, message: str, *, code: str | None = None, status: int | None = None, details: dict[str, Any] | None = None, ) -> None: super().__init__(message) if code: self.code = code if status: self.status = HTTPStatus(status) self.details = dict(details or {}) class BadRequestError(PrintAgentError): status = HTTPStatus.BAD_REQUEST code = "bad_request" class PrinterNotFoundError(PrintAgentError): status = HTTPStatus.NOT_FOUND code = "printer_not_found" class NotSupportedError(PrintAgentError): status = HTTPStatus.NOT_IMPLEMENTED code = "not_supported" class SpoolerError(PrintAgentError): status = HTTPStatus.INTERNAL_SERVER_ERROR code = "spooler_error" class PrinterOfflineError(PrintAgentError): """Tlačiareň je vypnutá, odpojená, pozastavená alebo úlohu neprevzala — nič sa nevytlačilo (úloha bola z radu odstránená), doklad ostáva „čaká na tlač".""" status = HTTPStatus.SERVICE_UNAVAILABLE code = "printer_offline" # --------------------------------------------------------------------------- # Konfigurácia # --------------------------------------------------------------------------- @dataclass class AgentConfig: port: int = DEFAULT_PORT token: str = "" allowed_origins: list[str] = field(default_factory=lambda: list(DEFAULT_ALLOWED_ORIGINS)) default_printer: str | None = None log_file: str | None = None def to_json(self) -> str: return json.dumps(asdict(self), ensure_ascii=False, indent=2) + "\n" def default_config_path() -> Path: return Path.home() / CONFIG_DIR_NAME / "config.json" def generate_token() -> str: """Token = 32 hex znakov (128 bitov náhodnosti).""" return secrets.token_hex(16) def load_or_create_config(path: Path | None = None) -> tuple[AgentConfig, bool]: """Načíta config.json; ak neexistuje, vytvorí ho s novým tokenom. Vracia (config, created). Chýbajúce kľúče v existujúcom súbore sa doplnia predvolenými hodnotami (a súbor sa dopíše), neznáme kľúče sa ignorujú. """ path = path or default_config_path() path.parent.mkdir(parents=True, exist_ok=True) created = False raw: dict[str, Any] = {} if path.exists(): try: raw = json.loads(path.read_text(encoding="utf-8")) if not isinstance(raw, dict): raise ValueError("config.json nie je JSON objekt") except (OSError, ValueError) as exc: raise PrintAgentError(f"Konfiguračný súbor {path} sa nedá načítať: {exc}") from exc else: created = True cfg = AgentConfig() port = raw.get("port", cfg.port) if not isinstance(port, int) or not (1 <= port <= 65535): raise PrintAgentError(f"Neplatný port v {path}: {port!r}") cfg.port = port token = raw.get("token") or "" if not isinstance(token, str) or not is_acceptable_token(token): token = generate_token() created = True # nový token = treba ho znova zobraziť obsluhe cfg.token = token.lower() origins = raw.get("allowed_origins", cfg.allowed_origins) if not isinstance(origins, list) or not all(isinstance(o, str) for o in origins): raise PrintAgentError(f"allowed_origins v {path} musí byť zoznam reťazcov") cfg.allowed_origins = [normalize_origin(o) for o in origins if o.strip()] dp = raw.get("default_printer") cfg.default_printer = dp if isinstance(dp, str) and dp.strip() else None lf = raw.get("log_file", str(path.parent / "agent.log")) cfg.log_file = lf if isinstance(lf, str) and lf.strip() else None # Zapíš späť (doplní chýbajúce kľúče / nový token). Súbor s tokenom má byť # čitateľný len pre používateľa. if created or not set(asdict(cfg)) <= set(raw): path.write_text(cfg.to_json(), encoding="utf-8") try: os.chmod(path, 0o600) except OSError: pass return cfg, created def normalize_origin(origin: str) -> str: return origin.strip().rstrip("/").lower() def is_safe_printer_name(name: str) -> bool: """Názov tlačiarne z JSON tela: bez riadiacich znakov, nie UNC/URL cesta ani prepínač (`\\\\server\\share`, `//host/x`, `-o ...`) — allowlist zoznamu OS platí navyše.""" if not name or len(name) > 255: return False if name.startswith(("\\\\", "//", "-")): return False return not any(ord(ch) < 32 or ch == "\x7f" for ch in name) def is_acceptable_token(token: str) -> bool: """32 až 128 hex znakov, nie zástupná hodnota z ukážkovej konfigurácie a nie triviálny reťazec (jeden opakovaný znak, vzostupná/zostupná sekvencia hex číslic).""" if not re.fullmatch(r"[0-9a-fA-F]{32,128}", token): return False t = token.lower() if t in PLACEHOLDER_TOKENS or len(set(t)) <= 2: return False seq = "0123456789abcdef" * 8 if t in seq or t in seq[::-1]: return False return True # --------------------------------------------------------------------------- # Čakanie na odovzdanie úlohy tlačiarni (spoločné pre Windows aj CUPS) # --------------------------------------------------------------------------- def wait_for_spool_job( get_state: Callable[[], str], cancel: Callable[[], None], *, timeout_s: float = JOB_WAIT_S, printing_timeout_s: float = JOB_PRINTING_WAIT_S, poll_s: float = JOB_POLL_S, cancel_wait_s: float = JOB_CANCEL_WAIT_S, sleep: Callable[[float], None] = time.sleep, clock: Callable[[], float] = time.monotonic, printer: str = "", ) -> None: """Počká, kým tlačový subsystém úlohu odovzdá tlačiarni. `get_state()` vracia JOB_STATE_DONE (úloha opustila rad / vytlačená), JOB_STATE_PENDING (v rade), JOB_STATE_PRINTING (dáta sa odovzdávajú) alebo jeden z JOB_ERROR_STATES (tlačiareň offline, bez papiera, rad pozastavený, úloha zrušená obsluhou v OS...). Pri chybe alebo po uplynutí lehoty sa úloha zruší (`cancel()`), počká sa na jej zmiznutie z radu a vyhodí sa PrinterOfflineError (HTTP 503) s poľom job_cancelled. Návrat bez výnimky = úloha bola odovzdaná tlačiarni. """ start = clock() while True: state = get_state() if state == JOB_STATE_DONE: return elapsed = clock() - start reason: str | None = None if state in JOB_ERROR_STATES: reason = state elif state == JOB_STATE_PRINTING and elapsed > printing_timeout_s: reason = f"timeout_printing_{int(printing_timeout_s)}s" elif state != JOB_STATE_PRINTING and elapsed > timeout_s: reason = f"timeout_{int(timeout_s)}s" if reason is None: sleep(poll_s) continue cancelled = _cancel_spool_job(get_state, cancel, cancel_wait_s, sleep, clock) where = f" '{printer}'" if printer else "" if reason.startswith("timeout"): what = f"neprevzala tlačovú úlohu do {int(timeout_s)} s" elif reason == "cancelled": what = "úlohu zrušila obsluha v tlačovom rade operačného systému" else: what = f"je vypnutá, odpojená alebo hlási chybu ({reason})" tail = ( "úloha bola z tlačového radu odstránená, doklad zostáva v stave čaká na tlač." if cancelled else "úlohu sa NEPODARILO z tlačového radu odstrániť; pred opakovaním tlače ju odstráňte " "v tlačovom rade operačného systému, inak hrozí dvojitý výtlačok. Doklad zostáva v stave čaká na tlač." ) raise PrinterOfflineError( f"Tlačiareň{where} {what}; {tail}", details={"job_cancelled": cancelled, "job_state": reason}, ) def _cancel_spool_job(get_state, cancel, cancel_wait_s: float, sleep, clock) -> bool: try: cancel() except Exception as exc: # noqa: BLE001 — zrušenie je najlepšia snaha log.warning("zrušenie úlohy v tlačovom rade zlyhalo: %s", exc) return False deadline = clock() + cancel_wait_s while clock() < deadline: try: if get_state() == JOB_STATE_DONE: return True except Exception as exc: # noqa: BLE001 log.warning("stav úlohy po zrušení sa nedá prečítať: %s", exc) return False sleep(0.1) return False # --------------------------------------------------------------------------- # ESC/POS skúšobný blok # --------------------------------------------------------------------------- def build_test_print_escpos(printer_name: str, now: datetime | None = None) -> bytes: """Skúšobný výtlačok agenta (textový ESC/POS, len ASCII — textový režim závisí od kódovej stránky tlačiarne, preto bez diakritiky). Nie je to pokladničný doklad a nikdy sa neukladá do CHDÚ.""" now = now or datetime.now() lines = [ "SORP PRINT AGENT " + AGENT_VERSION, "SKUSOBNA TLAC", "NIE JE POKLADNICNY DOKLAD", "", "Tlaciaren: " + _ascii(printer_name)[:40], "Cas: " + now.strftime("%d.%m.%Y %H:%M:%S"), "System: " + platform_id(), "", "Ak vidite tento text, tlaciaren", "prijima RAW ulohy bez dialogu OS.", "", ] out = bytearray() out += ESC + b"@" # ESC @ inicializácia out += ESC + b"a\x01" # ESC a 1 centrovanie out += ESC + b"E\x01" # ESC E 1 tučné out += lines[0].encode("ascii") + b"\n" out += ESC + b"E\x00" for line in lines[1:]: out += line.encode("ascii", "replace") + b"\n" out += b"\n\n\n" out += GS + b"V\x42\x00" # GS V 66 0 čiastočné odrezanie s posunom return bytes(out) def _ascii(s: str) -> str: return s.encode("ascii", "replace").decode("ascii") def platform_id() -> str: if sys.platform.startswith("win"): return "windows" if sys.platform == "darwin": return "darwin" return "linux" # --------------------------------------------------------------------------- # Tlačové backendy # --------------------------------------------------------------------------- @dataclass(frozen=True) class PrinterInfo: name: str default: bool = False class PrinterBackend: """Rozhranie backendu. `print_raw`/`print_pdf` vracajú identifikátor úlohy v spooleri (alebo None) až keď tlačový subsystém úlohu ODOVZDAL tlačiarni (úloha opustila rad / má stav vytlačené); pri zlyhaní vyhodia PrinterOfflineError (tlačiareň vypnutá/odpojená/pozastavená, úloha z radu odstránená) / SpoolerError / PrinterNotFoundError / NotSupportedError. `printer_status` vracia PRINTER_STATE_* (backend bez informácie: unknown).""" platform = "unknown" def list_printers(self) -> list[PrinterInfo]: raise NotImplementedError def printer_status(self, printer: str) -> str: return PRINTER_STATE_UNKNOWN def ensure_printer_ready(self, printer: str) -> None: """Pred zaradením úlohy: tlačiareň v stave offline/paper_out/error/paused úlohu nedostane (nič sa nezaradí do radu, žiadne riziko duplikátu).""" state = self.printer_status(printer) if state == PRINTER_STATE_NOT_FOUND: raise PrinterNotFoundError(f"Tlačiareň '{printer}' neexistuje") if state not in PRINTER_STATES_OK: raise PrinterOfflineError( f"Tlačiareň '{printer}' nie je pripravená (stav: {state}); úloha sa nezaradila do " "tlačového radu, doklad zostáva v stave čaká na tlač.", details={"job_cancelled": True, "job_state": state}, ) def default_printer(self) -> str | None: for p in self.list_printers(): if p.default: return p.name return None def print_raw(self, printer: str, data: bytes, title: str) -> str | None: raise NotImplementedError def print_pdf(self, printer: str, data: bytes, title: str) -> str | None: raise NotImplementedError # ----- CUPS (macOS / Linux) -------------------------------------------------- _LPSTAT_PRINTER_RE = re.compile(r"^printer\s+(\S+)\s", re.MULTILINE) _LPSTAT_DEFAULT_RE = re.compile(r"^system default destination:\s*(\S+)", re.MULTILINE) _LP_REQUEST_ID_RE = re.compile(r"request id is\s+(\S+)") def parse_lpstat_printers(text: str) -> list[str]: return _LPSTAT_PRINTER_RE.findall(text) def parse_lpstat_default(text: str) -> str | None: m = _LPSTAT_DEFAULT_RE.search(text) return m.group(1) if m else None def parse_lp_request_id(text: str) -> str | None: m = _LP_REQUEST_ID_RE.search(text) return m.group(1) if m else None def parse_lpstat_printer_state(text: str, printer: str) -> str: """Stav tlačiarne z `lpstat -p ` (+ dôvody na ďalších riadkoch).""" lines = text.splitlines() block: list[str] = [] for line in lines: if line.startswith("printer "): if block: break if line.split()[1:2] == [printer]: block.append(line) elif block: block.append(line) if not block: return PRINTER_STATE_UNKNOWN # neexistujúcu tlačiareň hlási lpstat návratovým kódom (printer_status) joined = "\n".join(block).lower() if "disabled" in block[0].lower() or "paused" in joined: return PRINTER_STATE_PAUSED if "not connected" in joined or "offline" in joined or "unable to connect" in joined: return PRINTER_STATE_OFFLINE if "media-empty" in joined or "out of paper" in joined or "media-needed" in joined: return PRINTER_STATE_PAPER_OUT if "door-open" in joined or "cover-open" in joined or "media-jam" in joined: return PRINTER_STATE_ERROR return PRINTER_STATE_READY class CupsBackend(PrinterBackend): """macOS a Linux: `lpstat` na zoznam a stav, `lp -o raw` na RAW (ESC/POS), `lp` na PDF; po odovzdaní sa čaká (`lpstat -W not-completed -o`), kým úloha opustí rad, inak sa zruší (`cancel`) a vráti 503 printer_offline.""" def __init__(self) -> None: self.platform = platform_id() self._lp = shutil.which("lp") self._lpstat = shutil.which("lpstat") def _run(self, argv: list[str], data: bytes | None = None, timeout: float = 60.0) -> subprocess.CompletedProcess: try: return subprocess.run(argv, input=data, capture_output=True, timeout=timeout, check=False) except FileNotFoundError as exc: raise NotSupportedError(f"Príkaz {argv[0]} nie je dostupný (CUPS nie je nainštalované)") from exc except subprocess.TimeoutExpired as exc: raise SpoolerError(f"Príkaz {argv[0]} neodpovedal do {timeout:.0f} s") from exc def list_printers(self) -> list[PrinterInfo]: if not self._lpstat: return [] names = parse_lpstat_printers(self._run([self._lpstat, "-p"]).stdout.decode("utf-8", "replace")) default = parse_lpstat_default(self._run([self._lpstat, "-d"]).stdout.decode("utf-8", "replace")) return [PrinterInfo(name=n, default=(n == default)) for n in names] def _lp_submit(self, argv_tail: list[str], data: bytes | None) -> str | None: if not self._lp: raise NotSupportedError("Príkaz lp nie je dostupný (CUPS nie je nainštalované)") proc = self._run([self._lp, *argv_tail], data=data) out = proc.stdout.decode("utf-8", "replace") + proc.stderr.decode("utf-8", "replace") if proc.returncode != 0: raise SpoolerError(f"lp zlyhalo (kód {proc.returncode}): {out.strip()[:300]}") return parse_lp_request_id(out) def printer_status(self, printer: str) -> str: if not self._lpstat: return PRINTER_STATE_UNKNOWN try: proc = self._run([self._lpstat, "-p", printer], timeout=10) except PrintAgentError: return PRINTER_STATE_UNKNOWN out = proc.stdout.decode("utf-8", "replace") + proc.stderr.decode("utf-8", "replace") if proc.returncode != 0 and ("does not exist" in out or "Invalid destination" in out): return PRINTER_STATE_NOT_FOUND return parse_lpstat_printer_state(out, printer) def _job_state(self, printer: str, request_id: str) -> str: proc = self._run([self._lpstat, "-W", "not-completed", "-o", printer], timeout=10) out = proc.stdout.decode("utf-8", "replace") if not any(line.split()[:1] == [request_id] for line in out.splitlines()): return JOB_STATE_DONE state = self.printer_status(printer) if state in (PRINTER_STATE_OFFLINE, PRINTER_STATE_PAPER_OUT, PRINTER_STATE_ERROR, PRINTER_STATE_PAUSED): return state line = self._printer_line(printer).lower() return JOB_STATE_PRINTING if ("now printing" in line or "processing" in line) else JOB_STATE_PENDING def _printer_line(self, printer: str) -> str: proc = self._run([self._lpstat, "-p", printer], timeout=10) return proc.stdout.decode("utf-8", "replace") def _wait_delivered(self, printer: str, request_id: str | None) -> None: if not request_id or not self._lpstat: return cancel_cmd = shutil.which("cancel") def cancel() -> None: if cancel_cmd: self._run([cancel_cmd, request_id], timeout=10) wait_for_spool_job(lambda: self._job_state(printer, request_id), cancel, printer=printer) def print_raw(self, printer: str, data: bytes, title: str) -> str | None: self.ensure_printer_ready(printer) request_id = self._lp_submit(["-d", printer, "-o", "raw", "-t", title, "--"], data) self._wait_delivered(printer, request_id) return request_id def print_pdf(self, printer: str, data: bytes, title: str) -> str | None: self.ensure_printer_ready(printer) fd, tmp = tempfile.mkstemp(prefix="sorp-", suffix=".pdf") try: with os.fdopen(fd, "wb") as fh: fh.write(data) # lp súbor pri odovzdaní skopíruje do spoolu — po návrate ho možno zmazať. request_id = self._lp_submit(["-d", printer, "-t", title, "--", tmp], None) finally: try: os.unlink(tmp) except OSError: pass self._wait_delivered(printer, request_id) return request_id # ----- Windows (ctypes + winspool.drv) -------------------------------------- _WINDOWS_PDF_TOOLS = ( ("SumatraPDF", ["-print-to", "{printer}", "-silent", "{file}"]), ("SumatraPDF.exe", ["-print-to", "{printer}", "-silent", "{file}"]), ("PDFtoPrinter", ["{file}", "{printer}"]), ("PDFtoPrinter.exe", ["{file}", "{printer}"]), ) def find_windows_pdf_tool() -> tuple[str, list[str]] | None: """Nájde v PATH nástroj na tichú tlač PDF na Windows (SumatraPDF alebo PDFtoPrinter).""" for exe, args in _WINDOWS_PDF_TOOLS: path = shutil.which(exe) if path: return path, args return None # Win32 konštanty (wingdi.h / winspool.h) PRINTER_ATTRIBUTE_WORK_OFFLINE = 0x00000400 PRINTER_STATUS_PAUSED = 0x00000001 PRINTER_STATUS_ERROR = 0x00000002 PRINTER_STATUS_PAPER_JAM = 0x00000008 PRINTER_STATUS_PAPER_OUT = 0x00000010 PRINTER_STATUS_PAPER_PROBLEM = 0x00000040 PRINTER_STATUS_OFFLINE = 0x00000080 PRINTER_STATUS_NOT_AVAILABLE = 0x00001000 PRINTER_STATUS_USER_INTERVENTION = 0x00100000 PRINTER_STATUS_DOOR_OPEN = 0x00400000 JOB_STATUS_PAUSED = 0x00000001 JOB_STATUS_ERROR = 0x00000002 JOB_STATUS_DELETING = 0x00000004 JOB_STATUS_SPOOLING = 0x00000008 JOB_STATUS_PRINTING = 0x00000010 JOB_STATUS_OFFLINE = 0x00000020 JOB_STATUS_PAPEROUT = 0x00000040 JOB_STATUS_PRINTED = 0x00000080 JOB_STATUS_DELETED = 0x00000100 JOB_STATUS_BLOCKED_DEVQ = 0x00000200 JOB_STATUS_USER_INTERVENTION = 0x00000400 JOB_STATUS_COMPLETE = 0x00001000 JOB_CONTROL_DELETE = 5 ERROR_INVALID_PARAMETER = 87 ERROR_INSUFFICIENT_BUFFER = 122 def printer_state_from_win32(attributes: int, status: int) -> str: """PRINTER_INFO_4.Attributes + PRINTER_INFO_6.dwStatus -> PRINTER_STATE_*.""" if attributes & PRINTER_ATTRIBUTE_WORK_OFFLINE or status & (PRINTER_STATUS_OFFLINE | PRINTER_STATUS_NOT_AVAILABLE): return PRINTER_STATE_OFFLINE if status & (PRINTER_STATUS_PAPER_OUT | PRINTER_STATUS_PAPER_JAM | PRINTER_STATUS_PAPER_PROBLEM): return PRINTER_STATE_PAPER_OUT if status & PRINTER_STATUS_PAUSED: return PRINTER_STATE_PAUSED if status & (PRINTER_STATUS_ERROR | PRINTER_STATUS_USER_INTERVENTION | PRINTER_STATUS_DOOR_OPEN): return PRINTER_STATE_ERROR return PRINTER_STATE_READY def job_state_from_win32(job_status: int, printer_state: str = PRINTER_STATE_READY) -> str: """JOB_INFO_1.Status (+ stav tlačiarne) -> JOB_STATE_* / chybový stav.""" if job_status & (JOB_STATUS_PRINTED | JOB_STATUS_COMPLETE): return JOB_STATE_DONE if job_status & (JOB_STATUS_DELETING | JOB_STATUS_DELETED): return "cancelled" if job_status & JOB_STATUS_PAPEROUT: return "paper_out" if job_status & JOB_STATUS_OFFLINE: return "offline" if job_status & JOB_STATUS_PAUSED: return "paused" if job_status & JOB_STATUS_BLOCKED_DEVQ: return "blocked" if job_status & JOB_STATUS_USER_INTERVENTION: return "user_intervention" if job_status & JOB_STATUS_ERROR: return "error" if printer_state in (PRINTER_STATE_OFFLINE, PRINTER_STATE_PAPER_OUT, PRINTER_STATE_PAUSED, PRINTER_STATE_ERROR): return printer_state if job_status & JOB_STATUS_PRINTING: return JOB_STATE_PRINTING return JOB_STATE_PENDING class WindowsBackend(PrinterBackend): """Windows: RAW tlač cez winspool.drv (OpenPrinterW / StartDocPrinterW s pDatatype="RAW" / WritePrinter / EndDocPrinter) — bez pywin32, bez dialógu. Pred StartDocPrinter sa cez GetPrinterW overí, že tlačiareň nie je offline, bez papiera, pozastavená ani v chybe; po EndDocPrinter sa cez GetJobW čaká, kým úloha opustí tlačový rad (port monitor ju odovzdal tlačiarni), inak sa úloha cez SetJobW zruší a vráti sa 503 printer_offline. PDF sa tlačí len ak je v PATH SumatraPDF alebo PDFtoPrinter (inak 501).""" platform = "windows" PRINTER_ENUM_LOCAL = 0x00000002 PRINTER_ENUM_CONNECTIONS = 0x00000004 ERROR_INSUFFICIENT_BUFFER = ERROR_INSUFFICIENT_BUFFER def __init__(self) -> None: import ctypes from ctypes import wintypes self._ctypes = ctypes self._wt = wintypes class DOC_INFO_1W(ctypes.Structure): # noqa: N801 - názov podľa Win32 API _fields_ = [ ("pDocName", wintypes.LPWSTR), ("pOutputFile", wintypes.LPWSTR), ("pDatatype", wintypes.LPWSTR), ] class PRINTER_INFO_4W(ctypes.Structure): # noqa: N801 - názov podľa Win32 API _fields_ = [ ("pPrinterName", wintypes.LPWSTR), ("pServerName", wintypes.LPWSTR), ("Attributes", wintypes.DWORD), ] class PRINTER_INFO_6(ctypes.Structure): # noqa: N801 - názov podľa Win32 API _fields_ = [("dwStatus", wintypes.DWORD)] class SYSTEMTIME(ctypes.Structure): # noqa: N801 - názov podľa Win32 API _fields_ = [(n, wintypes.WORD) for n in ( "wYear", "wMonth", "wDayOfWeek", "wDay", "wHour", "wMinute", "wSecond", "wMilliseconds", )] class JOB_INFO_1W(ctypes.Structure): # noqa: N801 - názov podľa Win32 API _fields_ = [ ("JobId", wintypes.DWORD), ("pPrinterName", wintypes.LPWSTR), ("pMachineName", wintypes.LPWSTR), ("pUserName", wintypes.LPWSTR), ("pDocument", wintypes.LPWSTR), ("pDatatype", wintypes.LPWSTR), ("pStatus", wintypes.LPWSTR), ("Status", wintypes.DWORD), ("Priority", wintypes.DWORD), ("Position", wintypes.DWORD), ("TotalPages", wintypes.DWORD), ("PagesPrinted", wintypes.DWORD), ("Submitted", SYSTEMTIME), ] self.DOC_INFO_1W = DOC_INFO_1W self.PRINTER_INFO_4W = PRINTER_INFO_4W self.PRINTER_INFO_6 = PRINTER_INFO_6 self.JOB_INFO_1W = JOB_INFO_1W ws = ctypes.WinDLL("winspool.drv", use_last_error=True) ws.OpenPrinterW.argtypes = [wintypes.LPWSTR, ctypes.POINTER(wintypes.HANDLE), ctypes.c_void_p] ws.OpenPrinterW.restype = wintypes.BOOL ws.ClosePrinter.argtypes = [wintypes.HANDLE] ws.ClosePrinter.restype = wintypes.BOOL ws.StartDocPrinterW.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.c_void_p] ws.StartDocPrinterW.restype = wintypes.DWORD ws.StartPagePrinter.argtypes = [wintypes.HANDLE] ws.StartPagePrinter.restype = wintypes.BOOL ws.WritePrinter.argtypes = [wintypes.HANDLE, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)] ws.WritePrinter.restype = wintypes.BOOL ws.EndPagePrinter.argtypes = [wintypes.HANDLE] ws.EndPagePrinter.restype = wintypes.BOOL ws.EndDocPrinter.argtypes = [wintypes.HANDLE] ws.EndDocPrinter.restype = wintypes.BOOL ws.EnumPrintersW.argtypes = [ wintypes.DWORD, wintypes.LPWSTR, wintypes.DWORD, wintypes.LPBYTE, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ctypes.POINTER(wintypes.DWORD), ] ws.EnumPrintersW.restype = wintypes.BOOL ws.GetDefaultPrinterW.argtypes = [wintypes.LPWSTR, ctypes.POINTER(wintypes.DWORD)] ws.GetDefaultPrinterW.restype = wintypes.BOOL ws.GetPrinterW.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.LPBYTE, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ] ws.GetPrinterW.restype = wintypes.BOOL ws.GetJobW.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.LPBYTE, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ] ws.GetJobW.restype = wintypes.BOOL ws.SetJobW.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.LPBYTE, wintypes.DWORD] ws.SetJobW.restype = wintypes.BOOL self._ws = ws def _last_error(self, what: str) -> SpoolerError: code = self._ctypes.get_last_error() try: text = self._ctypes.FormatError(code).strip() except Exception: text = "" return SpoolerError(f"{what} zlyhalo (Win32 chyba {code}: {text})") def _enum_printers(self) -> list[str]: ctypes, wt, ws = self._ctypes, self._wt, self._ws flags = self.PRINTER_ENUM_LOCAL | self.PRINTER_ENUM_CONNECTIONS needed = wt.DWORD(0) returned = wt.DWORD(0) ws.EnumPrintersW(flags, None, 4, None, 0, ctypes.byref(needed), ctypes.byref(returned)) if needed.value == 0: return [] buf = (ctypes.c_byte * needed.value)() ok = ws.EnumPrintersW( flags, None, 4, ctypes.cast(buf, wt.LPBYTE), needed.value, ctypes.byref(needed), ctypes.byref(returned) ) if not ok: raise self._last_error("EnumPrintersW") arr = ctypes.cast(buf, ctypes.POINTER(self.PRINTER_INFO_4W)) return [arr[i].pPrinterName for i in range(returned.value) if arr[i].pPrinterName] def default_printer(self) -> str | None: ctypes, wt, ws = self._ctypes, self._wt, self._ws size = wt.DWORD(0) ws.GetDefaultPrinterW(None, ctypes.byref(size)) if size.value == 0: return None buf = ctypes.create_unicode_buffer(size.value) if not ws.GetDefaultPrinterW(buf, ctypes.byref(size)): return None return buf.value or None def list_printers(self) -> list[PrinterInfo]: default = self.default_printer() return [PrinterInfo(name=n, default=(n == default)) for n in self._enum_printers()] # --- stav tlačiarne a úlohy (GetPrinterW / GetJobW / SetJobW) --- def _open(self, printer: str): ctypes, wt, ws = self._ctypes, self._wt, self._ws handle = wt.HANDLE() if not ws.OpenPrinterW(printer, ctypes.byref(handle), None): code = ctypes.get_last_error() if code in (1801, 2): # ERROR_INVALID_PRINTER_NAME, ERROR_FILE_NOT_FOUND raise PrinterNotFoundError(f"Tlačiareň '{printer}' neexistuje") raise self._last_error("OpenPrinterW") return handle def _get_printer_level(self, handle, level: int): """GetPrinterW pre úroveň 4 (Attributes) alebo 6 (dwStatus); None pri chybe.""" ctypes, wt, ws = self._ctypes, self._wt, self._ws needed = wt.DWORD(0) ws.GetPrinterW(handle, level, None, 0, ctypes.byref(needed)) if needed.value == 0: return None buf = (ctypes.c_byte * needed.value)() if not ws.GetPrinterW(handle, level, ctypes.cast(buf, wt.LPBYTE), needed.value, ctypes.byref(needed)): log.warning("GetPrinterW(level %s) zlyhalo (Win32 chyba %s)", level, ctypes.get_last_error()) return None struct = self.PRINTER_INFO_4W if level == 4 else self.PRINTER_INFO_6 return ctypes.cast(buf, ctypes.POINTER(struct)).contents def _printer_state(self, handle) -> str: info4 = self._get_printer_level(handle, 4) info6 = self._get_printer_level(handle, 6) if info4 is None and info6 is None: return PRINTER_STATE_UNKNOWN attributes = int(info4.Attributes) if info4 is not None else 0 status = int(info6.dwStatus) if info6 is not None else 0 return printer_state_from_win32(attributes, status) def printer_status(self, printer: str) -> str: try: handle = self._open(printer) except PrinterNotFoundError: return PRINTER_STATE_NOT_FOUND except PrintAgentError as exc: log.warning("printer_status(%s): %s", printer, exc) return PRINTER_STATE_UNKNOWN try: return self._printer_state(handle) finally: self._ws.ClosePrinter(handle) def _job_state(self, handle, job_id: int) -> str: ctypes, wt, ws = self._ctypes, self._wt, self._ws needed = wt.DWORD(0) ws.GetJobW(handle, job_id, 1, None, 0, ctypes.byref(needed)) code = ctypes.get_last_error() if code == ERROR_INVALID_PARAMETER or needed.value == 0: return JOB_STATE_DONE # úloha opustila tlačový rad (port ju prevzal) if code not in (0, ERROR_INSUFFICIENT_BUFFER): log.warning("GetJobW(%s) zlyhalo (Win32 chyba %s) — stav úlohy sa nedá určiť", job_id, code) return JOB_STATE_DONE # optimisticky: bez informácie sa úloha nezruší (riziko dvojitej tlače) buf = (ctypes.c_byte * needed.value)() if not ws.GetJobW(handle, job_id, 1, ctypes.cast(buf, wt.LPBYTE), needed.value, ctypes.byref(needed)): code = ctypes.get_last_error() if code == ERROR_INVALID_PARAMETER: return JOB_STATE_DONE log.warning("GetJobW(%s) zlyhalo (Win32 chyba %s)", job_id, code) return JOB_STATE_DONE info = ctypes.cast(buf, ctypes.POINTER(self.JOB_INFO_1W)).contents return job_state_from_win32(int(info.Status), self._printer_state(handle)) def _cancel_job(self, handle, job_id: int) -> None: if not self._ws.SetJobW(handle, job_id, 0, None, JOB_CONTROL_DELETE): raise self._last_error(f"SetJobW(DELETE, {job_id})") def print_raw(self, printer: str, data: bytes, title: str) -> str | None: ctypes, wt, ws = self._ctypes, self._wt, self._ws handle = self._open(printer) try: # Pred zaradením do radu: vypnutá/odpojená (WORK_OFFLINE), bez papiera, # pozastavená alebo chybová tlačiareň úlohu nedostane — nič sa nezaradí. state = self._printer_state(handle) if state not in PRINTER_STATES_OK: raise PrinterOfflineError( f"Tlačiareň '{printer}' nie je pripravená (stav: {state}); úloha sa nezaradila do " "tlačového radu, doklad zostáva v stave čaká na tlač.", details={"job_cancelled": True, "job_state": state}, ) doc = self.DOC_INFO_1W(title, None, "RAW") job_id = ws.StartDocPrinterW(handle, 1, ctypes.byref(doc)) if job_id == 0: raise self._last_error("StartDocPrinterW") try: if not ws.StartPagePrinter(handle): raise self._last_error("StartPagePrinter") buf = ctypes.create_string_buffer(data, len(data)) base = ctypes.addressof(buf) offset = 0 total = len(data) while offset < total: written = wt.DWORD(0) ok = ws.WritePrinter(handle, ctypes.c_void_p(base + offset), total - offset, ctypes.byref(written)) if not ok: raise self._last_error("WritePrinter") if written.value == 0: raise SpoolerError("WritePrinter nezapísal žiadne bajty (tlačiareň neprijíma dáta)") offset += written.value if not ws.EndPagePrinter(handle): raise self._last_error("EndPagePrinter") except Exception: # Neúplná úloha nesmie zostať v rade (po zapnutí tlačiarne by vyšiel # torzový výtlačok); pôvodná chyba sa vracia volajúcemu. ws.EndDocPrinter(handle) try: self._cancel_job(handle, job_id) except PrintAgentError as exc: log.warning("neúplnú úlohu %s sa nepodarilo zrušiť: %s", job_id, exc) raise if not ws.EndDocPrinter(handle): err = self._last_error("EndDocPrinter") try: self._cancel_job(handle, job_id) except PrintAgentError as exc: log.warning("úlohu %s sa po zlyhaní EndDocPrinter nepodarilo zrušiť: %s", job_id, exc) raise err # Spooler úlohu prijal — teraz sa čaká, kým ju port monitor odovzdá # tlačiarni (úloha opustí rad); pri vypnutej tlačiarni sa zruší (503). wait_for_spool_job( lambda: self._job_state(handle, job_id), lambda: self._cancel_job(handle, job_id), printer=printer, ) return str(job_id) finally: ws.ClosePrinter(handle) def print_pdf(self, printer: str, data: bytes, title: str) -> str | None: tool = find_windows_pdf_tool() if tool is None: raise NotSupportedError( "PDF tlač na Windows vyžaduje nástroj SumatraPDF alebo PDFtoPrinter v PATH; " "pre pokladničné doklady použite formát escpos (RAW), ktorý agent tlačí priamo." ) self.ensure_printer_ready(printer) exe, arg_tpl = tool fd, tmp = tempfile.mkstemp(prefix="sorp-", suffix=".pdf") try: with os.fdopen(fd, "wb") as fh: fh.write(data) argv = [exe] + [a.format(printer=printer, file=tmp) for a in arg_tpl] try: proc = subprocess.run(argv, capture_output=True, timeout=120, check=False) except subprocess.TimeoutExpired as exc: raise SpoolerError(f"{os.path.basename(exe)} neodpovedal do 120 s") from exc if proc.returncode != 0: err = (proc.stderr or proc.stdout).decode("utf-8", "replace").strip()[:300] raise SpoolerError(f"{os.path.basename(exe)} zlyhal (kód {proc.returncode}): {err}") return None finally: try: os.unlink(tmp) except OSError: pass def make_backend() -> PrinterBackend: if sys.platform.startswith("win"): return WindowsBackend() return CupsBackend() # --------------------------------------------------------------------------- # Stav agenta: idempotencia úloh + spracovanie tlače # --------------------------------------------------------------------------- @dataclass class JobRecord: at: float result: dict[str, Any] class JobStore: """Pamäť úspešne odovzdaných úloh (job_id -> výsledok) s TTL 60 s. Zlyhané úlohy sa NEPAMÄTAJÚ — opakovanie po chybe má tlač skúsiť znova.""" def __init__(self, ttl_s: float = IDEMPOTENCY_TTL_S, clock=time.monotonic) -> None: self._ttl = ttl_s self._clock = clock self._items: dict[str, JobRecord] = {} self._lock = threading.Lock() def _prune(self, now: float) -> None: stale = [k for k, rec in self._items.items() if now - rec.at > self._ttl] for k in stale: del self._items[k] def get(self, job_id: str) -> dict[str, Any] | None: with self._lock: now = self._clock() self._prune(now) rec = self._items.get(job_id) return dict(rec.result) if rec else None def put(self, job_id: str, result: dict[str, Any]) -> None: with self._lock: now = self._clock() self._prune(now) self._items[job_id] = JobRecord(at=now, result=dict(result)) class AgentState: def __init__(self, config: AgentConfig, backend: PrinterBackend, jobs: JobStore | None = None) -> None: self.config = config self.backend = backend self.jobs = jobs or JobStore() self._print_lock = threading.Lock() # tlač sa serializuje; duplicitný job_id čaká a dostane výsledok self.allowed_origins = {normalize_origin(o) for o in config.allowed_origins} # --- autorizácia / CORS --- def token_ok(self, presented: str | None) -> bool: if not presented or not self.config.token: return False return hmac.compare_digest(presented.strip().lower(), self.config.token) def origin_allowed(self, origin: str | None) -> bool: return origin is not None and normalize_origin(origin) in self.allowed_origins # --- tlačiarne --- def resolve_printer(self, requested: str | None) -> str: printers = self.backend.list_printers() names = [p.name for p in printers] name = (requested or "").strip() or self.config.default_printer or self.backend.default_printer() if not name: raise PrinterNotFoundError("Nie je určená ani predvolená žiadna tlačiareň") if not is_safe_printer_name(name): raise PrinterNotFoundError(f"Názov tlačiarne '{name[:60]}' nie je povolený") # Allowlist zoznamu tlačiarní OS platí VŽDY: bez enumerácie (žiadna # tlačiareň / chyba spoolera) sa cudzí názov (UNC cesta, iný cieľ) # neodovzdá OpenPrinterW ani `lp -d`. if not names: raise PrinterNotFoundError("V operačnom systéme nie je nainštalovaná žiadna tlačiareň") if name not in names: raise PrinterNotFoundError(f"Tlačiareň '{name}' neexistuje (dostupné: {', '.join(names)})") return name def _safe_printer_status(self, name: str | None) -> str: if not name: return PRINTER_STATE_NOT_FOUND try: return self.backend.printer_status(name) except Exception as exc: # noqa: BLE001 — stav je informatívny, nesmie zhodiť /status log.warning("printer_status(%s) zlyhalo: %s", name, exc) return PRINTER_STATE_UNKNOWN def status(self, printer: str | None = None) -> dict[str, Any]: try: default = self.config.default_printer or self.backend.default_printer() except Exception as exc: log.warning("default_printer zlyhalo: %s", exc) default = None used = (printer or "").strip() or default try: names = [p.name for p in self.backend.list_printers()] except Exception as exc: # noqa: BLE001 log.warning("list_printers zlyhalo: %s", exc) names = [] if used and names and used not in names: printer_status = PRINTER_STATE_NOT_FOUND else: printer_status = self._safe_printer_status(used) return { "ok": True, "agent": AGENT_NAME, "version": AGENT_VERSION, "protocol": PROTOCOL_VERSION, "platform": self.backend.platform, "default_printer": default, # Aditívne: tlačiareň, ktorú by /print použil (požadovaná alebo predvolená), # a jej stav — webová aplikácia pri inom stave než ready/unknown kanál # nepovažuje za pripravený (blokujúci stav „čaká na tlač“). "printer": used, "printer_status": printer_status, } def printers(self) -> dict[str, Any]: items = self.backend.list_printers() cfg_default = self.config.default_printer out = [] for p in items: is_default = (p.name == cfg_default) if cfg_default else p.default out.append({"name": p.name, "default": bool(is_default), "status": self._safe_printer_status(p.name)}) return {"ok": True, "printers": out} # --- tlač --- @staticmethod def parse_print_request(body: bytes) -> dict[str, Any]: try: req = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, ValueError) as exc: raise BadRequestError(f"Telo požiadavky nie je platný JSON: {exc}", code="invalid_json") from exc if not isinstance(req, dict): raise BadRequestError("Telo požiadavky musí byť JSON objekt", code="invalid_json") fmt = req.get("format") if fmt not in ALLOWED_FORMATS: raise BadRequestError(f"format musí byť jeden z {list(ALLOWED_FORMATS)}", code="invalid_format") data_b64 = req.get("data") if not isinstance(data_b64, str) or not data_b64.strip(): raise BadRequestError("data (base64) chýba", code="invalid_data") try: data = base64.b64decode(data_b64, validate=True) except (binascii.Error, ValueError) as exc: raise BadRequestError(f"data nie je platný base64: {exc}", code="invalid_base64") from exc if not data: raise BadRequestError("data je prázdne", code="invalid_data") if fmt == "pdf" and not data.startswith(b"%PDF-"): raise BadRequestError("data pre format=pdf nie je PDF súbor", code="invalid_pdf") printer = req.get("printer") if printer is not None and not isinstance(printer, str): raise BadRequestError("printer musí byť reťazec alebo null", code="invalid_printer") job_id = req.get("job_id") if job_id is None: job_id = uuid.uuid4().hex if not isinstance(job_id, str) or not job_id.strip() or len(job_id) > 200: raise BadRequestError("job_id musí byť neprázdny reťazec (max. 200 znakov)", code="invalid_job_id") width = req.get("width_dots") if width is not None and (isinstance(width, bool) or width not in ALLOWED_WIDTHS): raise BadRequestError( f"width_dots musí byť {ALLOWED_WIDTHS[0]}, {ALLOWED_WIDTHS[1]} alebo null", code="invalid_width" ) return {"format": fmt, "data": data, "printer": printer, "job_id": job_id.strip(), "width_dots": width} def handle_print(self, body: bytes) -> dict[str, Any]: req = self.parse_print_request(body) job_id = req["job_id"] started = time.time() with self._print_lock: cached = self.jobs.get(job_id) if cached is not None: cached["duplicate"] = True log.info("job=%s DUPLIKÁT (do %d s) -> pôvodný výsledok, netlačí sa", job_id, int(IDEMPOTENCY_TTL_S)) return cached try: printer = self.resolve_printer(req["printer"]) title = f"SORP {job_id}" if req["format"] == "escpos": spool_id = self.backend.print_raw(printer, req["data"], title) else: spool_id = self.backend.print_pdf(printer, req["data"], title) except PrintAgentError as exc: log.error( "job=%s format=%s printer=%s bytes=%d CHYBA %s: %s %s", job_id, req["format"], req["printer"], len(req["data"]), exc.code, exc, json.dumps(exc.details, ensure_ascii=False) if exc.details else "", ) raise except Exception as exc: log.exception("job=%s neočakávaná chyba: %s", job_id, exc) raise SpoolerError(f"Neočakávaná chyba tlače: {exc}") from exc result = { "ok": True, "job_id": job_id, "printer": printer, "format": req["format"], "bytes": len(req["data"]), "width_dots": req["width_dots"], "spooler_job_id": spool_id, } self.jobs.put(job_id, result) log.info( "job=%s format=%s printer=%s bytes=%d spooler=%s OK (%.0f ms)", job_id, req["format"], printer, len(req["data"]), spool_id, (time.time() - started) * 1000, ) return result # --------------------------------------------------------------------------- # HTTP server # --------------------------------------------------------------------------- class PrintAgentServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True def __init__(self, state: AgentState, port: int) -> None: self.state = state # Bind výhradne na loopback — bez ohľadu na konfiguráciu. super().__init__((BIND_HOST, port), PrintAgentHandler) class PrintAgentHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = f"{AGENT_NAME}/{AGENT_VERSION}" sys_version = "" @property def state(self) -> AgentState: return self.server.state # type: ignore[attr-defined] # --- pomocné --- def log_message(self, fmt: str, *args: Any) -> None: # noqa: D401 log.debug("http %s - %s", self.address_string(), fmt % args) def _cors_headers(self, origin: str | None) -> dict[str, str]: if not self.state.origin_allowed(origin): return {} return { "Access-Control-Allow-Origin": origin or "", "Access-Control-Allow-Methods": "GET, POST, OPTIONS", "Access-Control-Allow-Headers": f"Content-Type, {TOKEN_HEADER}", "Access-Control-Max-Age": "600", "Vary": "Origin", } def _send_json(self, status: int, payload: dict[str, Any], *, close: bool = False) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") for k, v in self._cors_headers(self.headers.get("Origin")).items(): self.send_header(k, v) if close: self.send_header("Connection", "close") self.close_connection = True self.end_headers() self.wfile.write(body) def _error(self, exc: PrintAgentError, *, close: bool = False) -> None: payload: dict[str, Any] = {"ok": False, "error": exc.code, "message": str(exc)} payload.update(getattr(exc, "details", None) or {}) self._send_json(int(exc.status), payload, close=close) @staticmethod def _origin_error(origin: str | None) -> dict[str, Any]: return {"ok": False, "error": "origin_not_allowed", "message": f"Origin {origin} nie je povolený"} def _authorized(self) -> bool: # Pri odmietnutí POSTu sa telo nečíta — spojenie treba zavrieť, inak by # neprečítané bajty pri keep-alive poškodili nasledujúcu požiadavku. close = self.command == "POST" origin = self.headers.get("Origin") if origin is not None and not self.state.origin_allowed(origin): self._send_json(403, self._origin_error(origin), close=close) return False if not self.state.token_ok(self.headers.get(TOKEN_HEADER)): self._send_json(401, {"ok": False, "error": "unauthorized"}, close=close) return False return True def _read_body(self) -> bytes: length_raw = self.headers.get("Content-Length") if length_raw is None: raise BadRequestError("Chýba hlavička Content-Length", code="length_required", status=411) try: length = int(length_raw) except ValueError as exc: raise BadRequestError("Neplatná hlavička Content-Length", code="invalid_length") from exc if length < 0: raise BadRequestError("Neplatná hlavička Content-Length", code="invalid_length") if length > MAX_BODY_BYTES: raise BadRequestError( f"Telo požiadavky presahuje limit {MAX_BODY_BYTES // (1024 * 1024)} MB", code="payload_too_large", status=413, ) return self.rfile.read(length) # --- metódy --- def do_OPTIONS(self) -> None: # noqa: N802 origin = self.headers.get("Origin") if not self.state.origin_allowed(origin): self._send_json(403, self._origin_error(origin)) return self.send_response(204) for k, v in self._cors_headers(origin).items(): self.send_header(k, v) self.send_header("Content-Length", "0") self.end_headers() def do_GET(self) -> None: # noqa: N802 if not self._authorized(): return path, _, query = self.path.partition("?") try: if path == "/status": from urllib.parse import parse_qs requested = parse_qs(query).get("printer", [None])[0] if query else None self._send_json(200, self.state.status(requested)) elif path == "/printers": self._send_json(200, self.state.printers()) else: self._send_json(404, {"ok": False, "error": "not_found"}) except PrintAgentError as exc: self._error(exc) except Exception as exc: log.exception("GET %s zlyhalo", path) self._send_json(500, {"ok": False, "error": "internal_error", "message": str(exc)}) def do_POST(self) -> None: # noqa: N802 if not self._authorized(): return path = self.path.split("?", 1)[0] if path != "/print": self._send_json(404, {"ok": False, "error": "not_found"}, close=True) return try: body = self._read_body() except BadRequestError as exc: # Telo sa neprečítalo — spojenie treba zavrieť, inak by zostali bajty v prúde. self._error(exc, close=True) return try: self._send_json(200, self.state.handle_print(body)) except PrintAgentError as exc: self._error(exc) except Exception as exc: log.exception("POST /print zlyhalo") self._send_json(500, {"ok": False, "error": "internal_error", "message": str(exc)}) def make_server(config: AgentConfig, backend: PrinterBackend, port: int | None = None) -> PrintAgentServer: state = AgentState(config, backend) return PrintAgentServer(state, config.port if port is None else port) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def setup_logging(log_file: str | None, verbose: bool) -> None: fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%Y-%m-%d %H:%M:%S") log.setLevel(logging.DEBUG if verbose else logging.INFO) log.handlers.clear() sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) log.addHandler(sh) if log_file: try: Path(log_file).parent.mkdir(parents=True, exist_ok=True) fh = logging.FileHandler(log_file, encoding="utf-8") fh.setFormatter(fmt) log.addHandler(fh) except OSError as exc: log.warning("Log súbor %s sa nedá otvoriť: %s", log_file, exc) def print_banner(cfg: AgentConfig, config_path: Path, created: bool) -> None: line = "=" * 64 print(line) print(f" SORP PRINT AGENT {AGENT_VERSION} ({platform_id()})") print(line) if created: print(" Prvé spustenie: vygenerovaný nový prístupový TOKEN.") print() print(f" URL: http://{BIND_HOST}:{cfg.port}") print(f" TOKEN: {cfg.token}") print() print(" V pokladnici SORP: Nastavenia -> Tlačiareň -> Tlačový agent") print(" zadajte URL a TOKEN uvedené vyššie a zvoľte tlačiareň.") print() print(f" Konfigurácia: {config_path}") if cfg.log_file: print(f" Log: {cfg.log_file}") print(f" Povolené originy: {', '.join(cfg.allowed_origins)}") print(f" Predvolená tlačiareň: {cfg.default_printer or '(predvolená tlačiareň systému)'}") print(line) print(" Agent beží. Ukončenie: Ctrl+C") sys.stdout.flush() def cmd_list_printers(backend: PrinterBackend) -> int: printers = backend.list_printers() if not printers: print("Nenašla sa žiadna tlačiareň.") return 1 for p in printers: print(f"{'*' if p.default else ' '} {p.name}") print("(* = predvolená tlačiareň systému)") return 0 def cmd_print_test(backend: PrinterBackend, printer: str | None, cfg: AgentConfig | None = None) -> int: name = (printer or "").strip() or (cfg.default_printer if cfg else None) or backend.default_printer() if not name: raise PrinterNotFoundError("Nie je určená ani predvolená žiadna tlačiareň (--print-test NAZOV)") data = build_test_print_escpos(name) spool = backend.print_raw(name, data, "SORP skusobna tlac") print(f"Skúšobná tlač odovzdaná spooleru: tlačiareň={name} bajtov={len(data)} úloha={spool}") return 0 # ----- automatické spustenie po prihlásení -------------------------------- AUTOSTART_LABEL = "sk.verteco.sorp-print-agent" def agent_command() -> list[str]: """Príkaz, ktorým sa spustí táto inštancia agenta: skompilovaná binárka (PyInstaller, sys.frozen) alebo interpret Pythonu + tento skript.""" if getattr(sys, "frozen", False): return [sys.executable] return [sys.executable, str(Path(__file__).resolve())] def autostart_path(plat: str | None = None) -> Path: plat = plat or platform_id() if plat == "windows": appdata = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming") return Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup" / "sorp-print-agent.cmd" if plat == "darwin": return Path.home() / "Library" / "LaunchAgents" / f"{AUTOSTART_LABEL}.plist" return Path.home() / ".config" / "autostart" / "sorp-print-agent.desktop" def render_autostart(plat: str, command: list[str], config_path: Path | None) -> str: """Obsah súboru automatického spustenia pre daný OS (bez externých závislostí).""" args = list(command) if config_path is not None: args += ["--config", str(config_path)] if plat == "windows": quoted = " ".join(f'"{a}"' for a in args) return ( "@echo off\r\n" "rem SORP Print Agent - automaticke spustenie po prihlaseni\r\n" f'start "SORP Print Agent" /min {quoted}\r\n' ) if plat == "darwin": from xml.sax.saxutils import escape items = "".join(f"\n {escape(a)}" for a in args) return ( '\n' '\n' '\n\n' f" Label\n {AUTOSTART_LABEL}\n" f" ProgramArguments\n {items}\n \n" " RunAtLoad\n \n" " ProcessType\n Background\n" "\n\n" ) exec_line = " ".join('"' + a.replace("\\", "\\\\").replace('"', '\\"') + '"' for a in args) return ( "[Desktop Entry]\nType=Application\nName=SORP Print Agent\n" "Comment=Lokalny tlacovy agent pokladnice SORP (127.0.0.1)\n" f"Exec={exec_line}\nTerminal=false\nX-GNOME-Autostart-enabled=true\n" ) def _launchctl(args: list[str]) -> None: """launchctl (macOS) — najlepšia snaha, chyby sa len zalogujú.""" try: subprocess.run(["launchctl", *args], capture_output=True, timeout=15, check=False) except (OSError, subprocess.TimeoutExpired) as exc: log.warning("launchctl %s zlyhalo: %s", " ".join(args), exc) def _gui_domain() -> str: uid = os.getuid() if hasattr(os, "getuid") else 0 return f"gui/{uid}" def install_autostart(config_path: Path | None) -> Path: plat = platform_id() path = autostart_path(plat) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(render_autostart(plat, agent_command(), config_path), encoding="utf-8") if plat == "darwin": _launchctl(["bootstrap", _gui_domain(), str(path)]) return path def uninstall_autostart() -> Path | None: plat = platform_id() path = autostart_path(plat) if not path.exists(): return None if plat == "darwin": _launchctl(["bootout", _gui_domain(), str(path)]) path.unlink() return path def cmd_install_autostart(config_path: Path | None) -> int: path = install_autostart(config_path) print(f"Automatické spustenie nastavené: {path}") print("Agent sa spustí pri každom prihlásení používateľa (minimalizované okno).") if platform_id() == "windows": print("Zrušenie: sorp-print-agent --uninstall-autostart (alebo súbor zmažte).") return 0 def cmd_uninstall_autostart() -> int: path = uninstall_autostart() if path is None: print("Automatické spustenie nebolo nastavené.") return 0 print(f"Automatické spustenie zrušené: {path}") return 0 def serve(cfg: AgentConfig, backend: PrinterBackend, config_path: Path, created: bool) -> int: try: server = make_server(cfg, backend) except OSError as exc: log.error("Port %s na %s sa nedá otvoriť: %s (beží už iná inštancia agenta?)", cfg.port, BIND_HOST, exc) return 2 print_banner(cfg, config_path, created) log.info( "štart %s %s na http://%s:%s (platform=%s)", AGENT_NAME, AGENT_VERSION, BIND_HOST, cfg.port, backend.platform ) try: server.serve_forever(poll_interval=0.5) except KeyboardInterrupt: pass finally: server.server_close() log.info("agent ukončený") return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog=AGENT_NAME, description="SORP Print Agent, lokálny tlačový agent (127.0.0.1)") parser.add_argument( "--config", type=Path, default=None, help="cesta ku config.json (predvolene ~/.sorp-print-agent/config.json)" ) parser.add_argument("--port", type=int, default=None, help="port servera (prepíše hodnotu z konfigurácie)") parser.add_argument("--list-printers", action="store_true", help="vypíše tlačiarne operačného systému a skončí") parser.add_argument( "--print-test", nargs="?", const="", default=None, metavar="TLACIAREN", help="odošle ESC/POS skúšobný výtlačok (bez názvu = predvolená tlačiareň) a skončí", ) parser.add_argument("--show-token", action="store_true", help="vypíše token a URL a skončí") parser.add_argument( "--install-autostart", action="store_true", help="nastaví automatické spustenie po prihlásení (Windows: Po spustení, macOS: LaunchAgent, Linux: autostart)", ) parser.add_argument("--uninstall-autostart", action="store_true", help="zruší automatické spustenie agenta") parser.add_argument("--verbose", action="store_true", help="podrobné logovanie (aj HTTP prístupy)") parser.add_argument("--version", action="version", version=f"{AGENT_NAME} {AGENT_VERSION}") args = parser.parse_args(argv) try: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # Windows konzola s inou kódovou stránkou except Exception: pass try: cfg, created = load_or_create_config(args.config) except PrintAgentError as exc: print(f"CHYBA: {exc}", file=sys.stderr) return 2 if args.port is not None: cfg.port = args.port config_path = args.config or default_config_path() setup_logging(cfg.log_file, args.verbose) try: backend = make_backend() except Exception as exc: print(f"CHYBA: tlačový backend sa nedá inicializovať: {exc}", file=sys.stderr) return 2 try: if args.list_printers: return cmd_list_printers(backend) if args.print_test is not None: return cmd_print_test(backend, args.print_test, cfg) if args.show_token: print(f"URL: http://{BIND_HOST}:{cfg.port}") print(f"TOKEN: {cfg.token}") return 0 if args.install_autostart: return cmd_install_autostart(args.config) if args.uninstall_autostart: return cmd_uninstall_autostart() return serve(cfg, backend, config_path, created) except PrintAgentError as exc: print(f"CHYBA ({exc.code}): {exc}", file=sys.stderr) return 1 except OSError as exc: print(f"CHYBA: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())