# Part of Odoo. See LICENSE file for full copyright and licensing details. import logging from base64 import b64decode from datetime import datetime, timezone import subprocess import win32print import pywintypes from odoo.addons.iot_drivers.iot_handlers.drivers.printer_driver_base import PrinterDriverBase from odoo.addons.iot_drivers.tools import system from odoo.tools.mimetypes import guess_mimetype from odoo.addons.iot_drivers.iot_handlers.interfaces.printer_interface_W import win32print_lock _logger = logging.getLogger(__name__) class PrinterDriver(PrinterDriverBase): def __init__(self, identifier, device): super().__init__(identifier, device) self.device_connection = self._compute_device_connection(device) self.device_name = device.get('identifier') self.printer_handle = device.get('printer_handle') self.receipt_protocol = 'escpos' if any(cmd in device['identifier'] for cmd in ['STAR', 'Receipt']): self.device_subtype = "receipt_printer" elif "ZPL" in device['identifier']: self.device_subtype = "label_printer" else: self.device_subtype = "office_printer" @classmethod def supported(cls, device): return True @staticmethod def _compute_device_connection(device): return 'direct' if device['port'].startswith(('USB', 'TMUSB', 'COM', 'LPT')) else 'network' def disconnect(self): self.send_status('disconnected', 'Printer was disconnected') super().disconnect() def print_raw(self, data, action_unique_id=None): job_id = False page_started = False try: with win32print_lock: job_id = win32print.StartDocPrinter(self.printer_handle, 1, ('', None, "RAW")) win32print.StartPagePrinter(self.printer_handle) page_started = True win32print.WritePrinter(self.printer_handle, data) win32print.EndPagePrinter(self.printer_handle) win32print.EndDocPrinter(self.printer_handle) self.job_ids.append(job_id) if action_unique_id: self.job_action_ids[job_id] = action_unique_id except pywintypes.error as error: _logger.error("Error while printing raw data to printer %s: %s", self.device_name, error) if job_id or page_started: try: with win32print_lock: if page_started: win32print.EndPagePrinter(self.printer_handle) if job_id: win32print.EndDocPrinter(self.printer_handle) self.job_ids.append(job_id) if action_unique_id: self.job_action_ids[job_id] = action_unique_id except pywintypes.error as err: _logger.error("Error while finalizing print job to printer %s after failure: %s", self.device_name, err) self.send_status(status='error', message='ERROR_FAILED') raise def print_report(self, data): with win32print_lock: file_name = system.path_file('document.pdf') file_name.write_bytes(data) sumatra_pdf_path = system.path_file("SumatraPDF") / "SumatraPDF.exe" args = [ str(sumatra_pdf_path), "-print-to", self.device_name, str(file_name), "-silent", "-print-settings", "duplex" ] _logger.debug("Printing report with SumatraPDF using %s", args) try: subprocess.run(args, check=True) self.send_status(status='success') except subprocess.CalledProcessError as error: _logger.exception("Error while printing report, SumatraPDF args: %s, exit code: %s", args, error.returncode) self.send_status(status='error', message='ERROR_FAILED') def _action_default(self, data): _logger.debug("_action_default called for printer %s", self.device_name) document = b64decode(data['document']) mimetype = guess_mimetype(document) action_unique_id = data.get('action_unique_id') if mimetype == 'application/pdf': self.print_report(document) else: self.print_raw(document, action_unique_id=action_unique_id) _logger.debug("_action_default finished with mimetype %s for printer %s", mimetype, self.device_name) return {'print_id': data['print_id']} if 'print_id' in data else {} def status(self, _data=None): """Prints the status ticket of the IoT Box on the current printer. :param _data: dict provided by the action route """ if self.device_subtype == "receipt_printer": commands = self.RECEIPT_PRINTER_COMMANDS[self.receipt_protocol] self.print_raw(commands['center'] + (commands['title'] % b'IoT Box Test Receipt') + commands['cut']) elif self.device_type == "label_printer": self.print_raw("^XA^CI28 ^FT35,40 ^A0N,30 ^FDIoT Box Test Label^FS^XZ".encode()) # noqa: UP012 else: self.print_raw("IoT Box Test Page".encode()) # noqa: UP012 def _cancel_job_with_error(self, job_id, error_message): self.job_ids.remove(job_id) win32print.SetJob(self.printer_handle, job_id, 0, None, win32print.JOB_CONTROL_DELETE) self.send_status( status='error', message=error_message, action_unique_id=self.job_action_ids.pop(job_id, None) ) def _check_job_status(self, job_id): try: job = win32print.GetJob(self.printer_handle, job_id, win32print.JOB_INFO_1) elapsed_time = datetime.now(timezone.utc) - job['Submitted'] _logger.debug('job details for job id #%d: %s', job_id, job) if job['Status'] & win32print.JOB_STATUS_PRINTED: self.job_ids.remove(job_id) self.job_action_ids.pop(job_id, None) self.send_status(status='success') # Print timeout, e.g. network printer is disconnected if elapsed_time.seconds > self.job_timeout_seconds: self._cancel_job_with_error(job_id, 'ERROR_TIMEOUT') # Generic error, e.g. USB printer is not connected elif job['Status'] & win32print.JOB_STATUS_ERROR: self._cancel_job_with_error(job_id, 'ERROR_UNKNOWN') except pywintypes.error as error: # GetJob returns error 87 (incorrect parameter) if the print job doesn't exist. # Windows deletes print jobs on completion, so this actually means the print # was succcessful. if error.winerror == 87: self.send_status(status='success') else: _logger.exception('Win32 error occurred while querying print job') self.job_ids.remove(job_id) self._recent_action_ids.pop(self.job_action_ids.pop(job_id, None), None)