import argparse import importlib.util import json import os import platform import socket import sys import uuid from datetime import datetime from pypos.core.utils.config_utils import read_app_settings from pypos.core.utils.path_utils import get_app_data_dir, get_app_data_resource_dir, get_db_path from pypos.modules.printer.models.printer_settings_model import PrinterSettingsModel class PrinterDiagnosticService: def __init__(self, timeout_seconds=2.0): self.timeout_seconds = float(timeout_seconds) self.settings_model = PrinterSettingsModel() self.report = { "meta": {}, "paths": {}, "settings": {}, "checks": [], "printers": [], "recommendations": [], "summary": {"PASS": 0, "WARN": 0, "FAIL": 0}, } # edited by glg @staticmethod def _new_trace_id(scope: str = "printer_diag") -> str: return f"printer-{str(scope or 'diag').strip().lower()}-{uuid.uuid4().hex[:10]}" # edited by glg def _build_status_result(self, status: str, message: str, *, error_code: str = "", reason: str = ""): normalized_status = str(status or "").strip().upper() reason_text = str(reason or "").strip() or ( "ok" if normalized_status == "PASS" else "printer_diagnostic_issue" ) return { "status": normalized_status, "message": str(message or "").strip(), "error_code": str(error_code or "").strip(), "reason": reason_text, "trace_id": self._new_trace_id("check"), } def run(self): self._collect_meta() self._collect_paths() self._collect_settings() self._check_dependencies() self._check_printer_configuration() self._check_printer_runtime() self._build_recommendations() self._summarize() return self.report def save_report(self, output_path=None): if not output_path: stamp = datetime.now().strftime("%Y%m%d_%H%M%S") logs_dir = os.path.join(get_app_data_dir(), "logs") os.makedirs(logs_dir, exist_ok=True) output_path = os.path.join(logs_dir, f"printer_diagnostic_{stamp}.json") with open(output_path, "w", encoding="utf-8") as fh: json.dump(self.report, fh, ensure_ascii=False, indent=2) return output_path def _collect_meta(self): self.report["meta"] = { "generated_at": datetime.now().isoformat(timespec="seconds"), "platform": platform.platform(), "python_version": sys.version, } def _collect_paths(self): app_data = get_app_data_dir() app_res = get_app_data_resource_dir() db_path = get_db_path() printers_json = os.path.join(app_res, "printers.json") self.report["paths"] = { "app_data_dir": app_data, "resource_dir": app_res, "db_path": db_path, "printers_json": printers_json, "db_exists": os.path.exists(db_path), "printers_json_exists": os.path.exists(printers_json), } if not os.path.exists(db_path): self._add_check( "paths", "db_exists", "WARN", f"File DB tidak ditemukan di path runtime: {db_path}", ) else: self._add_check("paths", "db_exists", "PASS", f"File DB ditemukan: {db_path}") def _collect_settings(self): app_cfg = read_app_settings() or {} self.report["settings"] = { "db_path": app_cfg.get("db_path"), "print_mode_default": app_cfg.get("print_mode_default"), "local_online_host": app_cfg.get("local_online_host"), "local_online_port": app_cfg.get("local_online_port"), "local_online_timeout": app_cfg.get("local_online_timeout"), } def _check_dependencies(self): uses_usb = False try: for printer in self.settings_model.load_printers(): mode = str(printer.get("koneksi") or "").strip().lower() address = str(printer.get("address") or "").strip() if mode == "usb" and address: uses_usb = True break except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): uses_usb = False escpos_available = importlib.util.find_spec("escpos") is not None usb_available = importlib.util.find_spec("usb") is not None serial_available = importlib.util.find_spec("serial") is not None win32_available = importlib.util.find_spec("win32print") is not None self._add_check( "dependency", "escpos", "PASS" if escpos_available else "FAIL", "Module escpos terdeteksi." if escpos_available else "Module escpos belum tersedia.", ) usb_dep_status = "PASS" if usb_available else ("FAIL" if uses_usb else "WARN") self._add_check( "dependency", "pyusb", usb_dep_status, "Module usb (pyusb) terdeteksi." if usb_available else ("Module usb (pyusb) wajib dipasang karena printer USB terkonfigurasi." if uses_usb else "Module usb (pyusb) belum tersedia."), ) self._add_check( "dependency", "pyserial", "PASS" if serial_available else "WARN", "Module serial (pyserial) terdeteksi." if serial_available else "Module serial (pyserial) belum tersedia.", ) self._add_check( "dependency", "pywin32", "PASS" if win32_available else "WARN", "win32print tersedia untuk cek status spooler." if win32_available else "win32print tidak tersedia (opsional).", ) if usb_available: try: import usb.core # type: ignore # Trigger backend load untuk membedakan pyusb terpasang vs backend USB tidak tersedia. list(usb.core.find(find_all=True)) self._add_check("dependency", "usb_backend", "PASS", "Backend USB siap dipakai.") except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as exc: msg = str(exc) if "No backend available" in msg: backend_status = "FAIL" if uses_usb else "WARN" self._add_check( "dependency", "usb_backend", backend_status, "Backend USB tidak tersedia (libusb belum siap).", ) else: self._add_check( "dependency", "usb_backend", "WARN", f"Gagal inisialisasi backend USB: {msg}", ) def _check_printer_configuration(self): try: printers = self.settings_model.load_printers() except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as exc: self._add_check( "configuration", "load_printers", "FAIL", f"Gagal membaca konfigurasi printer: {exc}", ) return if not printers: self._add_check("configuration", "printers_count", "FAIL", "Tidak ada printer terdaftar.") return self._add_check("configuration", "printers_count", "PASS", f"Printer terdaftar: {len(printers)}") default_count = 0 for idx, printer in enumerate(printers): record = { "index": idx, "name": str(printer.get("name") or printer.get("nama") or "").strip(), "paper_size": str(printer.get("paper_size") or printer.get("lebar_kertas") or "").strip(), "default": bool(printer.get("default") or int(printer.get("is_default") or 0) == 1), "koneksi": str(printer.get("koneksi") or "").strip().lower(), "address": str(printer.get("address") or "").strip(), "checks": [], } if record["default"]: default_count += 1 if not record["name"]: record["checks"].append({"status": "FAIL", "message": "Nama printer kosong."}) if not record["paper_size"]: record["checks"].append({"status": "WARN", "message": "Ukuran kertas kosong."}) if not record["koneksi"] or not record["address"]: record["checks"].append( { "status": "INFO", "message": "Mode spooler OS: field koneksi/address tidak diisi.", } ) self.report["printers"].append(record) if default_count != 1: self._add_check( "configuration", "default_printer", "FAIL", f"Jumlah default printer tidak valid: {default_count} (harus 1).", ) else: self._add_check("configuration", "default_printer", "PASS", "Default printer tunggal valid.") def _check_printer_runtime(self): try: from PySide6.QtPrintSupport import QPrinterInfo # type: ignore available = [p.printerName() for p in QPrinterInfo.availablePrinters()] os_default = QPrinterInfo.defaultPrinter() os_default_name = os_default.printerName() if os_default and not os_default.isNull() else "" self._add_check( "runtime", "os_printers", "PASS" if available else "WARN", f"Printer OS terdeteksi: {len(available)}", ) self._add_check( "runtime", "os_default_printer", "PASS" if os_default_name else "WARN", f"Default printer OS: {os_default_name or '-'}", ) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as exc: available = [] self._add_check("runtime", "os_printers", "WARN", f"Gagal baca printer OS: {exc}") for record in self.report["printers"]: name = record["name"] koneksi = record["koneksi"] address = record["address"] if name: if name in available: record["checks"].append({"status": "PASS", "message": "Nama printer ditemukan di OS."}) else: record["checks"].append({"status": "WARN", "message": "Nama printer tidak ditemukan di OS."}) if koneksi == "lan" and address: record["checks"].append(self._check_lan_address(address)) elif koneksi == "usb" and address: record["checks"].append(self._check_usb_address(address)) def _check_lan_address(self, address): try: host, port_raw = address.split(":", 1) port = int(port_raw) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return self._build_status_result( "FAIL", f"Format address LAN tidak valid: {address}", error_code="PRINTER_DIAG_LAN_ADDRESS_INVALID", reason="invalid_lan_address", ) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(self.timeout_seconds) try: sock.connect((host, port)) return self._build_status_result( "PASS", f"Koneksi LAN berhasil: {host}:{port}", ) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as exc: return self._build_status_result( "WARN", f"Koneksi LAN gagal {host}:{port} ({exc})", error_code="PRINTER_DIAG_LAN_CONNECT_FAIL", reason="lan_connect_failed", ) finally: try: sock.close() except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): pass def _check_usb_address(self, address): if importlib.util.find_spec("usb") is None: return self._build_status_result( "WARN", "Module usb (pyusb) belum tersedia.", error_code="PRINTER_DIAG_USB_MODULE_MISSING", reason="usb_module_missing", ) try: vid_raw, pid_raw = address.split(":", 1) vid = int(vid_raw, 16) pid = int(pid_raw, 16) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return self._build_status_result( "FAIL", f"Format address USB tidak valid: {address}", error_code="PRINTER_DIAG_USB_ADDRESS_INVALID", reason="invalid_usb_address", ) try: import usb.core # type: ignore device = usb.core.find(idVendor=vid, idProduct=pid) if device is None: return self._build_status_result( "WARN", f"Device USB tidak ditemukan (VID:PID {address}).", error_code="PRINTER_DIAG_USB_DEVICE_NOT_FOUND", reason="usb_device_not_found", ) return self._build_status_result( "PASS", f"Device USB terdeteksi (VID:PID {address}).", ) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as exc: text = str(exc) if "No backend available" in text: return self._build_status_result( "FAIL", "Backend USB tidak tersedia (libusb).", error_code="PRINTER_DIAG_USB_BACKEND_MISSING", reason="usb_backend_missing", ) return self._build_status_result( "WARN", f"Gagal cek USB {address} ({text})", error_code="PRINTER_DIAG_USB_CHECK_FAILED", reason="usb_check_failed", ) def _build_recommendations(self): recs = [] failed = self._collect_status("FAIL") warned = self._collect_status("WARN") if any(c.get("name") == "usb_backend" and c.get("status") == "FAIL" for c in self.report["checks"]): recs.append("Instal backend libusb pada Windows outlet agar mode USB dapat dipakai.") elif any(c.get("name") == "usb_backend" and c.get("status") == "WARN" for c in self.report["checks"]): recs.append("Backend USB belum siap. Abaikan jika outlet hanya memakai spooler/LAN, atau instal libusb jika akan memakai USB.") if any("Format address LAN tidak valid" in c.get("message", "") for c in warned + failed): recs.append("Perbaiki format printer LAN menjadi IP:PORT (contoh 192.168.1.10:9100).") if any("Format address USB tidak valid" in c.get("message", "") for c in warned + failed): recs.append("Perbaiki format printer USB menjadi VID:PID heksadesimal (contoh 0x04b8:0x0202).") if any(c.get("name") == "default_printer" and c.get("status") == "FAIL" for c in self.report["checks"]): recs.append("Pastikan hanya satu printer yang ditandai default di pengaturan printer.") if any("Nama printer tidak ditemukan di OS." in c.get("message", "") for c in warned): recs.append("Sinkronkan nama printer di aplikasi dengan nama printer Windows yang terpasang.") if any(c.get("name") == "printers_count" and c.get("status") == "FAIL" for c in self.report["checks"]): recs.append("Tambahkan minimal satu printer di menu Pengaturan Printer sebelum operasional.") if not recs: recs.append("Konfigurasi printer terlihat sehat. Lanjutkan uji cetak transaksi dan settlement di mesin outlet.") self.report["recommendations"] = recs def _summarize(self): for check in self.report["checks"]: status = check.get("status") if status in self.report["summary"]: self.report["summary"][status] += 1 for printer in self.report["printers"]: for check in printer.get("checks", []): status = check.get("status") if status in self.report["summary"]: self.report["summary"][status] += 1 def _collect_status(self, status): rows = [x for x in self.report["checks"] if x.get("status") == status] for printer in self.report["printers"]: for check in printer.get("checks", []): if check.get("status") == status: merged = dict(check) merged["printer"] = printer.get("name") rows.append(merged) return rows def _add_check(self, category, name, status, message): self.report["checks"].append( { "category": category, "name": name, "status": status, "message": message, } ) def _print_console_report(report, output_path): summary = report.get("summary", {}) print("[Printer Diagnostic]") print(f"Report file : {output_path}") print(f"PASS={summary.get('PASS',0)} WARN={summary.get('WARN',0)} FAIL={summary.get('FAIL',0)}") print("") for check in report.get("checks", []): print(f"- [{check.get('status')}] {check.get('category')}.{check.get('name')}: {check.get('message')}") for printer in report.get("printers", []): print(f"- [INFO] printer.{printer.get('name') or '-'}:") for check in printer.get("checks", []): print(f" - [{check.get('status')}] {check.get('message')}") print("") print("Rekomendasi:") for rec in report.get("recommendations", []): print(f"- {rec}") def run_cli(argv=None): parser = argparse.ArgumentParser(description="Diagnosa otomatis printer USB/LAN per outlet.") parser.add_argument("--timeout", type=float, default=2.0, help="Timeout socket LAN dalam detik.") parser.add_argument("--output", type=str, default="", help="Path output report JSON.") parser.add_argument("--json-only", action="store_true", help="Cetak report JSON ke stdout.") args = parser.parse_args(argv) service = PrinterDiagnosticService(timeout_seconds=args.timeout) report = service.run() output_path = service.save_report(output_path=args.output or None) if args.json_only: print(json.dumps(report, ensure_ascii=False, indent=2)) else: _print_console_report(report, output_path) return 1 if report.get("summary", {}).get("FAIL", 0) > 0 else 0 if __name__ == "__main__": raise SystemExit(run_cli())