# Python实战：5分钟搞定Windows打印机自动化（含win32print完整示例） 如果你是一名IT管理员，或者正在开发一个需要集成打印功能的办公自动化系统，那么“打印机”这三个字很可能曾让你头疼不已。在Windows环境下，打印机驱动、队列管理、状态监控、批量任务……这些看似简单的操作，背后却隐藏着复杂的系统交互。手动操作不仅效率低下，在需要处理成百上千份文档时，更是几乎不可能完成的任务。 这正是Python和`win32print`模块大显身手的地方。不同于那些只能简单调用系统打印对话框的脚本，`win32print`提供了从底层直接与Windows打印子系统对话的能力。这意味着你可以精确地控制打印作业的生命周期，实时监控打印机状态，处理各种异常，甚至绕过系统对话框直接发送原始打印数据。对于需要将打印功能无缝嵌入到ERP、CRM、文档管理系统中的开发者来说，这无异于打开了一扇新的大门。 本文将带你深入`win32print`的实战应用，从最基础的获取打印机列表开始，逐步构建一个健壮、高效的企业级打印自动化方案。我们不仅会讲解代码怎么写，更会探讨在真实办公场景中可能遇到的坑，以及如何优雅地跨过去。无论你是想批量打印工资条、自动生成并打印报告，还是需要监控网络打印机的状态，这里都有你需要的答案。 ## 1. 环境准备与核心模块初探 在开始编写任何代码之前，我们需要确保开发环境就绪。对于Windows平台下的打印机编程，`pywin32`库是我们的核心武器。它封装了Windows API，让我们能够用Python调用系统底层的功能。 首先，通过pip安装必要的库： ```bash pip install pywin32 ``` 安装过程通常很顺利。如果遇到权限问题，可以尝试以管理员身份运行命令行，或者使用虚拟环境。 `win32print`模块是`pywin32`的一部分，它专门负责与打印机相关的所有操作。在深入代码之前，理解几个核心概念至关重要： * **打印机句柄（Printer Handle）**：一个指向系统中特定打印机对象的引用，几乎所有`win32print`函数都需要它来操作目标打印机。你可以把它想象成操作文件时的文件描述符。 * **打印作业（Print Job）**：用户提交的一次打印请求，包含文档数据、份数、优先级等信息。每个作业都有一个唯一的ID。 * **打印队列（Print Queue）**：打印机上等待处理或正在处理的打印作业列表。监控队列是自动化任务的关键。 * **原始打印数据（RAW Data）**：绕过Windows的图形设备接口（GDI），直接发送给打印机的指令流。这对于驱动标签打印机、POS机等特殊设备非常有用。 一个常见的误区是认为`win32print`只能打印文本或PDF。实际上，通过它，你可以控制打印过程的几乎每一个环节。让我们从一个最简单的例子开始，感受一下它的能力。 ## 2. 打印机发现与基础操作 自动化打印的第一步，是让程序知道系统中有哪些打印机可用，并选择正确的那一台。在大型企业环境中，用户可能连接了网络打印机、本地USB打印机、甚至虚拟PDF打印机。 ### 2.1 枚举与选择打印机 获取打印机列表是基础操作。`EnumPrinters`函数提供了多种枚举方式，你可以根据需求选择。 ```python import win32print def list_all_printers(): """ 列出系统中所有可用的打印机（本地、网络连接、共享）。 """ # PRINTER_ENUM_LOCAL | PRINTER_ENUM_CONNECTIONS 组合可以列出大多数打印机 flags = win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS printers = win32print.EnumPrinters(flags) print("=== 系统打印机列表 ===") for i, printer_info in enumerate(printers): # printer_info 是一个元组，索引2的位置通常是打印机的友好名称 printer_name = printer_info[2] print(f"{i+1}. {printer_name}") return printers if __name__ == "__main__": printers = list_all_printers() # 获取默认打印机，这在很多自动化场景中是首选 default_printer = win32print.GetDefaultPrinter() print(f"\n当前默认打印机是：{default_printer}") ``` 运行这段代码，你会看到一个清晰的列表。但在实际项目中，我们往往需要更智能地选择打印机，而不是依赖默认设置。例如，根据文档类型选择彩色或黑白打印机，或者根据部门选择特定的网络打印机。 > 注意：`EnumPrinters`返回的信息结构在不同版本的Windows上可能略有差异。对于需要高可靠性的生产代码，建议查阅微软官方文档或增加额外的错误处理。 ### 2.2 打印一个简单的文本文件 最直接的打印需求就是发送一个已存在的文件（如.txt, .pdf, .docx）到打印机。虽然可以用`ShellExecute`调用系统关联程序，但`win32print`提供了更可控的方式。 下面是一个更“程序化”的打印文件示例，它允许你在打印前设置一些基本属性： ```python import win32print import win32api import os def print_file_smart(file_path, printer_name=None, copies=1): """ 智能打印文件。如果未指定打印机，则使用默认打印机。 参数: file_path (str): 要打印的文件的完整路径。 printer_name (str, optional): 目标打印机名称。默认为None（使用默认打印机）。 copies (int, optional): 打印份数。默认为1。 """ if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在: {file_path}") # 确定使用的打印机 if printer_name is None: printer_name = win32print.GetDefaultPrinter() print(f"使用默认打印机: {printer_name}") else: # 验证打印机是否存在 all_printers = [info[2] for info in win32print.EnumPrinters(win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS)] if printer_name not in all_printers: raise ValueError(f"未找到打印机: {printer_name}") # 使用ShellExecute的‘print’动词进行打印。 # 这是最简单的方式，系统会使用文件关联的应用程序来处理打印。 # 对于需要更多控制（如份数、双面）的情况，需要更复杂的方法（见后续章节）。 try: # 注意：ShellExecute的‘print’命令可能无法直接设置份数。 # 对于多份打印，一个可行的方法是多次调用，或使用更底层的API。 win32api.ShellExecute( 0, # 父窗口句柄 "print", # 操作 file_path, # 要打印的文件 f'"{printer_name}"', # 参数（指定打印机） ".", # 工作目录 0 # 显示命令（0=不显示） ) print(f"打印任务已提交: {os.path.basename(file_path)} -> {printer_name}") except Exception as e: print(f"打印失败: {e}") # 这里可以添加更详细的错误处理逻辑，比如记录日志、重试等 # 使用示例 if __name__ == "__main__": # 打印当前目录下的一个测试文档到默认打印机 print_file_smart("月度报告.pdf") # 打印到指定的网络打印机，2份 # print_file_smart("工资条.txt", printer_name="\\PrintServer\FinanceHP", copies=2) ``` 这个方法简单快捷，适合大多数“一键打印”场景。但它把打印渲染工作交给了关联的应用程序（如Acrobat Reader打印PDF，Word打印DOCX），你无法控制打印对话框的弹出，也无法设置复杂的打印属性。对于全自动的后台打印，我们需要更强大的工具。 ## 3. 高级打印控制与作业管理 当你的自动化脚本需要处理批量任务，或者在无人值守的服务器上运行时，基础的文件打印就显得力不从心了。你需要能够管理打印队列、监控作业状态、处理错误，甚至直接向打印机发送原始指令。 ### 3.1 直接发送原始数据打印 这是`win32print`最强大的功能之一。通过`StartDocPrinter`、`WritePrinter`等函数，你可以绕过Windows的打印驱动，直接向打印机发送它能够理解的原始数据流。这对于以下场景至关重要： * 驱动特殊打印机（如热敏票据打印机、标签打印机）。 * 打印已生成好的特定格式数据（如PCL、PostScript、ESC/POS指令）。 * 实现最高性能的打印，避免GDI渲染的开销。 下面是一个发送ESC/POS指令到热敏打印机的完整示例： ```python import win32print def print_raw_escpos(printer_name, text_content): """ 向支持ESC/POS指令的打印机发送原始打印命令。 常用于超市小票、POS机、热敏打印机。 参数: printer_name (str): 打印机名称。 text_content (str): 要打印的文本内容。 """ # 打开打印机，获取句柄 hprinter = win32print.OpenPrinter(printer_name) if hprinter is None: raise RuntimeError(f"无法打开打印机: {printer_name}") try: # 1. 开始一个打印文档 job_name = "Python ESC/POS 打印任务" doc_info = (job_name, None, "RAW") # 数据类型指定为RAW job_id = win32print.StartDocPrinter(hprinter, 1, doc_info) try: # 2. 开始一个新页面 win32print.StartPagePrinter(hprinter) # 3. 准备原始数据 # ESC/POS指令示例： # \x1B\x40: 初始化打印机 # \x1B\x21\x00: 设置字符格式（默认） # \x1B\x45\x01: 设置加粗开 # \x0A: 换行 (LF) # \x1B\x69: 部分切纸（如果有切刀） raw_data = b"" raw_data += b"\x1B\x40" # 初始化 raw_data += b"\x1B\x21\x00" # 取消所有字符格式 raw_data += b"\x1B\x45\x01" # 开启加粗 raw_data += b"=== 销售凭据 ===\n".encode('gbk') # 注意编码，中文常用GBK raw_data += b"\x1B\x45\x00" # 关闭加粗 raw_data += b"\n".encode('gbk') raw_data += f"商品: {text_content}\n".encode('gbk') raw_data += b"-" * 32 + b"\n" raw_data += b"感谢惠顾！\n".encode('gbk') raw_data += b"\x1B\x69" # 执行切纸（如果支持） # 4. 将数据写入打印机 bytes_written = win32print.WritePrinter(hprinter, raw_data) print(f"已向打印机发送 {bytes_written} 字节数据。") # 5. 结束当前页 win32print.EndPagePrinter(hprinter) except Exception as e: # 如果页面或写入过程中出错，需要中止作业 win32print.AbortPrinter(hprinter) raise e finally: # 6. 结束文档 win32print.EndDocPrinter(hprinter) finally: # 7. 无论如何，都要关闭打印机句柄 win32print.ClosePrinter(hprinter) print("原始打印任务完成。") # 使用示例 if __name__ == "__main__": # 请将‘POS-80’替换为你的实际票据打印机名称 print_raw_escpos("POS-80", "测试商品A x 2 单价：25.00元") ``` > **关键点**：原始打印时，数据的编码和指令集完全取决于你的打印机型号。务必查阅打印机的编程手册。常见的指令集有ESC/POS（针打、热敏）、PCL（惠普激光）、PostScript（Adobe）。 ### 3.2 全面的打印队列监控与管理 在自动化流程中，仅仅提交打印任务是不够的。你需要知道任务是否成功进入队列、是否卡住、何时完成。`win32print`提供了完整的作业枚举和控制功能。 下面的类封装了一个简单的打印队列监控器： ```python import win32print import time from datetime import datetime import threading class PrinterQueueMonitor: """打印机队列监控与管理类""" def __init__(self, printer_name): self.printer_name = printer_name self.hprinter = None self._monitoring = False def connect(self): """连接到打印机""" self.hprinter = win32print.OpenPrinter(self.printer_name) if self.hprinter is None: raise ConnectionError(f"无法连接到打印机: {self.printer_name}") print(f"已连接到打印机: {self.printer_name}") def disconnect(self): """断开连接""" if self.hprinter: win32print.ClosePrinter(self.hprinter) self.hprinter = None print("打印机连接已关闭。") def get_job_list(self, detail_level=1): """ 获取当前打印队列中的所有作业。 参数: detail_level (int): 信息详细程度（1-基本，2-详细，3-扩展）。 返回: list: 作业字典列表。 """ if not self.hprinter: self.connect() try: # 枚举所有作业（从第0个到最后一个） jobs = win32print.EnumJobs(self.hprinter, 0, -1, detail_level) except win32print.error as e: # 可能队列为空，返回空列表 if ‘参数不正确‘ in str(e): # 队列为空时的常见错误 return [] else: raise e formatted_jobs = [] for job in jobs: job_info = { 'JobId': job['JobId'], 'Document': job['pDocument'], 'Status': self._decode_job_status(job['Status']), 'Submitted': job['Submitted'].strftime('%Y-%m-%d %H:%M:%S') if job['Submitted'] else 'N/A', 'TotalPages': job['TotalPages'], 'Size': job['Size'], 'Priority': job['Priority'], } formatted_jobs.append(job_info) return formatted_jobs def _decode_job_status(self, status_code): """将作业状态码转换为可读文本""" status_map = { 0: "就绪", 1: "暂停", 2: "错误", 3: "正在删除", 4: "正在打印", 5: "脱机", 6: "缺纸", 7: "需要用户干预", } # 状态码可能是多个值的组合（位标志） status_texts = [] for code, text in status_map.items(): if status_code & (1 << code): status_texts.append(text) return " | ".join(status_texts) if status_texts else "未知" def control_job(self, job_id, action): """ 控制特定的打印作业。 参数: job_id (int): 作业ID。 action (str): 操作类型，可选 "PAUSE", "RESUME", "RESTART", "DELETE", "CANCEL"。 """ action_map = { "PAUSE": win32print.JOB_CONTROL_PAUSE, "RESUME": win32print.JOB_CONTROL_RESUME, "RESTART": win32print.JOB_CONTROL_RESTART, "DELETE": win32print.JOB_CONTROL_DELETE, "CANCEL": win32print.JOB_CONTROL_CANCEL, } if action not in action_map: raise ValueError(f"不支持的操作: {action}。请使用 {list(action_map.keys())}") try: win32print.SetJob(self.hprinter, job_id, 0, None, action_map[action]) print(f"作业 {job_id} 已执行 {action} 操作。") return True except win32print.error as e: print(f"操作作业 {job_id} 失败: {e}") return False def start_monitoring(self, interval=5, callback=None): """ 启动后台线程，定期监控打印队列。 参数: interval (int): 检查间隔（秒）。 callback (function): 当队列变化时调用的回调函数。 """ self._monitoring = True def monitor_loop(): last_job_count = -1 while self._monitoring: try: current_jobs = self.get_job_list() current_count = len(current_jobs) if current_count != last_job_count and callback: callback(current_jobs, current_count, last_job_count) last_job_count = current_count # 打印当前状态（简单日志） if current_jobs: print(f"[{datetime.now().strftime('%H:%M:%S')}] 队列中有 {current_count} 个作业。") for job in current_jobs[:2]: # 只显示前两个 print(f" - 作业 {job['JobId']}: {job['Document']} [{job['Status']}]") if current_count > 2: print(f" ... 还有 {current_count - 2} 个作业") # time.sleep(interval) except Exception as e: print(f"监控循环出错: {e}") time.sleep(interval * 2) # 出错后等待更长时间 time.sleep(interval) monitor_thread = threading.Thread(target=monitor_loop, daemon=True) monitor_thread.start() print(f"已启动对打印机 '{self.printer_name}' 的队列监控，间隔 {interval} 秒。") def stop_monitoring(self): """停止监控""" self._monitoring = False print("队列监控已停止。") # 使用示例 if __name__ == "__main__": monitor = PrinterQueueMonitor(win32print.GetDefaultPrinter()) monitor.connect() # 一次性查看队列 print("当前打印队列：") jobs = monitor.get_job_list() for job in jobs: print(f" ID:{job['JobId']:4d} | {job['Document'][:30]:30s} | 状态: {job['Status']}") # 启动后台监控（在实际脚本中，你可能需要保持主线程运行） # def on_queue_change(current_jobs, current_count, last_count): # if current_count > last_count: # print(f"新作业到达！当前共 {current_count} 个作业。") # elif current_count < last_count: # print(f"作业完成或移除。剩余 {current_count} 个作业。") # # monitor.start_monitoring(interval=3, callback=on_queue_change) # # try: # while True: # 保持主线程运行 # time.sleep(1) # except KeyboardInterrupt: # monitor.stop_monitoring() monitor.disconnect() ``` 这个监控器可以轻松集成到你的管理后台中，提供一个实时仪表板来显示所有打印机的状态。你甚至可以扩展它，在作业失败时发送邮件或Teams通知。 ## 4. 企业级实战：构建健壮的批量打印服务 现在，我们将前面学到的知识组合起来，构建一个面向企业环境的、健壮的批量打印服务。这个服务需要处理以下需求： 1. 从指定文件夹监控待打印文件。 2. 支持不同的打印机路由规则（如PDF去彩色打印机，TXT去高速激光打印机）。 3. 自动重试失败的打印任务。 4. 生成详细的打印日志。 5. 友好的配置界面（通过配置文件）。 ### 4.1 服务架构与配置 我们首先设计一个配置文件（`config.yaml`）来定义打印规则和参数： ```yaml # config.yaml printers: color_laser: name: "HP Color LaserJet MFP" description: "部门彩色打印机" supported_types: [".pdf", ".jpg", ".png"] default_copies: 1 monochrome_laser: name: "Xerox B205" description: "高速黑白打印机" supported_types: [".txt", ".log", ".csv", ".pdf"] default_copies: 1 label_printer: name: "Zebra GK420d" description: "标签打印机" supported_types: [".zpl"] # Zebra编程语言文件 default_copies: 1 watch_folders: - path: "C:\PrintQueue\Incoming" rules: - file_pattern: "*.pdf" printer: "color_laser" action: "print" - file_pattern: "report_*.pdf" printer: "monochrome_laser" copies: 2 action: "print" - file_pattern: "label_*.zpl" printer: "label_printer" action: "print_raw" - path: "C:\PrintQueue\Logs" rules: - file_pattern: "*.log" printer: "monochrome_laser" action: "print" delete_after: false # 打印后不删除，用于存档 service: scan_interval: 10 # 扫描文件夹间隔（秒） max_retries: 3 retry_delay: 30 log_file: "C:\PrintService\print_service.log" ``` 接下来，是服务的主干代码。由于篇幅限制，这里展示核心的调度逻辑： ```python # print_service_core.py import os import time import yaml import logging import win32print from pathlib import Path from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import threading import queue class PrintJob: """封装一个打印作业的所有信息""" def __init__(self, file_path, target_printer_config, action="print", copies=1, metadata=None): self.file_path = Path(file_path) self.printer_config = target_printer_config self.action = action # "print", "print_raw" self.copies = copies self.metadata = metadata or {} self.retry_count = 0 self.status = "PENDING" # PENDING, PROCESSING, SUCCESS, FAILED self.submitted_time = time.time() self.completed_time = None def __str__(self): return f"PrintJob({self.file_path.name} -> {self.printer_config.get('name')}, Status: {self.status})" class PrintService: """核心打印服务""" def __init__(self, config_path): self.config = self._load_config(config_path) self.job_queue = queue.Queue() self.printers = self.config['printers'] self.watch_folders = self.config['watch_folders'] self.service_config = self.config['service'] self._setup_logging() self._running = False self.worker_threads = [] def _load_config(self, config_path): with open(config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) def _setup_logging(self): log_file = self.service_config.get('log_file', 'print_service.log') logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, encoding='utf-8'), logging.StreamHandler() ] ) self.logger = logging.getLogger('PrintService') def start(self): """启动打印服务""" self.logger.info("启动打印服务...") self._running = True # 1. 启动文件夹监控（使用watchdog） self._start_folder_watchers() # 2. 启动工作线程池（处理打印队列） num_workers = 2 # 可以根据打印机数量调整 for i in range(num_workers): worker = threading.Thread(target=self._worker_loop, daemon=True, name=f"PrintWorker-{i}") worker.start() self.worker_threads.append(worker) self.logger.info(f"启动工作线程 {worker.name}") # 3. 初始扫描现有文件 self._scan_existing_files() self.logger.info("打印服务已启动并运行。") # 主线程保持运行 try: while self._running: time.sleep(1) # 这里可以添加健康检查、统计信息输出等 if len(self.worker_threads) == 0 or not any(t.is_alive() for t in self.worker_threads): self.logger.error("所有工作线程已停止，服务异常！") self.stop() except KeyboardInterrupt: self.logger.info("收到停止信号。") self.stop() def stop(self): """停止打印服务""" self.logger.info("正在停止打印服务...") self._running = False # 等待工作线程结束（设置超时） for worker in self.worker_threads: worker.join(timeout=5) self.logger.info("打印服务已停止。") def _start_folder_watchers(self): """为每个监控文件夹启动文件系统观察者""" self.observer = Observer() for folder_cfg in self.watch_folders: folder_path = Path(folder_cfg['path']) if not folder_path.exists(): folder_path.mkdir(parents=True, exist_ok=True) self.logger.info(f"创建监控文件夹: {folder_path}") event_handler = PrintFolderHandler(self, folder_cfg) self.observer.schedule(event_handler, str(folder_path), recursive=False) self.observer.start() self.logger.info(f"已启动文件系统监控，监控 {len(self.watch_folders)} 个文件夹。") def _scan_existing_files(self): """服务启动时，扫描监控文件夹中已存在的文件""" self.logger.info("扫描监控文件夹中的现有文件...") for folder_cfg in self.watch_folders: folder_path = Path(folder_cfg['path']) for file_path in folder_path.iterdir(): if file_path.is_file(): # 模拟文件创建事件 self._on_file_created(str(file_path), folder_cfg) def _on_file_created(self, file_path, folder_cfg): """当监控到新文件时调用，根据规则创建打印作业""" file_path = Path(file_path) matched_rule = None # 查找匹配的规则 for rule in folder_cfg.get('rules', []): import fnmatch if fnmatch.fnmatch(file_path.name, rule['file_pattern']): matched_rule = rule break if not matched_rule: self.logger.debug(f"文件 {file_path.name} 未匹配任何规则，忽略。") return # 获取打印机配置 printer_key = matched_rule.get('printer') if printer_key not in self.printers: self.logger.error(f"规则中指定的打印机 '{printer_key}' 在配置中未找到。") return printer_config = self.printers[printer_key] # 创建打印作业 job = PrintJob( file_path=file_path, target_printer_config=printer_config, action=matched_rule.get('action', 'print'), copies=matched_rule.get('copies', printer_config.get('default_copies', 1)), metadata={ 'rule_matched': matched_rule['file_pattern'], 'source_folder': folder_cfg['path'], 'delete_after': matched_rule.get('delete_after', True) } ) # 将作业加入队列 self.job_queue.put(job) self.logger.info(f"创建打印作业: {job}") def _worker_loop(self): """工作线程的主循环，从队列中取出作业并执行""" while self._running: try: job = self.job_queue.get(timeout=1) job.status = "PROCESSING" self._process_job(job) self.job_queue.task_done() except queue.Empty: continue except Exception as e: self.logger.error(f"工作线程发生未知错误: {e}", exc_info=True) def _process_job(self, job): """处理单个打印作业，包含重试逻辑""" self.logger.info(f"开始处理作业: {job}") max_retries = self.service_config.get('max_retries', 3) retry_delay = self.service_config.get('retry_delay', 30) while job.retry_count <= max_retries and self._running: try: if job.action == "print": success = self._print_file(job) elif job.action == "print_raw": success = self._print_raw_file(job) else: self.logger.error(f"未知的打印操作类型: {job.action}") job.status = "FAILED" break if success: job.status = "SUCCESS" job.completed_time = time.time() self.logger.info(f"作业处理成功: {job}") # 打印成功后，根据配置决定是否删除源文件 if job.metadata.get('delete_after', True): try: job.file_path.unlink() self.logger.debug(f"已删除源文件: {job.file_path}") except Exception as e: self.logger.warning(f"删除文件失败 {job.file_path}: {e}") break else: # 打印函数返回False（非异常） job.status = "FAILED" self.logger.error(f"作业处理失败（非异常）: {job}") break except Exception as e: job.retry_count += 1 self.logger.error(f"处理作业时出错 (尝试 {job.retry_count}/{max_retries}): {e}", exc_info=True) if job.retry_count <= max_retries: self.logger.info(f"等待 {retry_delay} 秒后重试...") time.sleep(retry_delay) else: job.status = "FAILED" job.completed_time = time.time() self.logger.error(f"作业达到最大重试次数，最终失败: {job}") # 这里可以将失败作业移动到“失败”文件夹，或发送警报 # 记录最终状态 self._log_job_result(job) def _print_file(self, job): """执行常规文件打印""" printer_name = job.printer_config['name'] file_path = job.file_path # 验证打印机是否存在 try: all_printer_names = [info[2] for info in win32print.EnumPrinters(win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS)] if printer_name not in all_printer_names: self.logger.error(f"打印机 '{printer_name}' 当前不可用。") return False except Exception as e: self.logger.error(f"枚举打印机失败: {e}") return False # 这里可以使用更高级的打印方法，例如通过win32ui创建DC并设置份数 # 为简单起见，这里使用ShellExecute，但注意它可能无法直接设置份数 # 对于多份打印，一个方法是循环调用，或使用更底层的API（如StartDocPrinter） import win32api try: for i in range(job.copies): win32api.ShellExecute( 0, "print", str(file_path), f'"{printer_name}"', ".", 0 ) if job.copies > 1: self.logger.debug(f"已提交第 {i+1}/{job.copies} 份打印。") time.sleep(0.5) # 份数间短暂延迟，避免队列拥堵 return True except Exception as e: self.logger.error(f"调用ShellExecute打印失败: {e}") return False def _print_raw_file(self, job): """执行原始数据打印（用于ZPL、EPL等指令文件）""" printer_name = job.printer_config['name'] file_path = job.file_path if not file_path.exists(): self.logger.error(f"原始打印文件不存在: {file_path}") return False try: # 读取文件内容作为原始数据 with open(file_path, 'rb') as f: raw_data = f.read() # 使用win32print原始打印API hprinter = win32print.OpenPrinter(printer_name) if hprinter is None: self.logger.error(f"无法打开打印机: {printer_name}") return False try: job_name = f"RawPrint_{file_path.name}" doc_info = (job_name, None, "RAW") job_id = win32print.StartDocPrinter(hprinter, 1, doc_info) try: win32print.StartPagePrinter(hprinter) bytes_written = win32print.WritePrinter(hprinter, raw_data) win32print.EndPagePrinter(hprinter) self.logger.debug(f"原始打印成功，发送 {bytes_written} 字节。") except Exception as e: win32print.AbortPrinter(hprinter) raise e finally: win32print.EndDocPrinter(hprinter) finally: win32print.ClosePrinter(hprinter) return True except Exception as e: self.logger.error(f"原始打印失败: {e}", exc_info=True) return False def _log_job_result(self, job): """将作业结果记录到数据库或日志文件（示例为简单日志）""" duration = (job.completed_time or time.time()) - job.submitted_time log_entry = ( f"RESULT | Job: {job.file_path.name} | " f"Printer: {job.printer_config.get('name')} | " f"Status: {job.status} | " f"Retries: {job.retry_count} | " f"Duration: {duration:.2f}s" ) self.logger.info(log_entry) class PrintFolderHandler(FileSystemEventHandler): """处理文件系统事件""" def __init__(self, print_service, folder_cfg): self.print_service = print_service self.folder_cfg = folder_cfg def on_created(self, event): if not event.is_directory: self.print_service._on_file_created(event.src_path, self.folder_cfg) # 服务启动入口 if __name__ == "__main__": # 请确保已安装PyYAML和watchdog: pip install pyyaml watchdog service = PrintService("config.yaml") service.start() ``` 这个服务框架已经具备了企业应用的核心要素：配置化、异步处理、错误重试、日志记录。你可以根据实际需求进一步扩展，例如： * **添加数据库支持**：将作业历史、打印机状态存入SQLite或MySQL。 * **实现Web管理界面**：使用Flask或FastAPI创建一个控制面板，用于查看队列、手动提交作业、暂停打印机。 * **集成通知系统**：当打印机缺纸、卡纸或连续作业失败时，发送邮件、短信或Teams消息。 * **增加作业优先级**：让重要的报告可以插队打印。 ### 4.2 错误处理与性能优化 在最后这部分，我们讨论几个在实际部署中必然会遇到的问题及其解决方案。 **常见错误处理：** 1. **打印机脱机或未就绪**： ```python def check_printer_status(printer_name): try: hprinter = win32print.OpenPrinter(printer_name) status_info = win32print.GetPrinter(hprinter, 2) status = status_info.get('Status', 0) win32print.ClosePrinter(hprinter) # 检查状态位 if status & win32print.PRINTER_STATUS_OFFLINE: return "OFFLINE" elif status & win32print.PRINTER_STATUS_PAPER_OUT: return "PAPER_OUT" elif status & win32print.PRINTER_STATUS_ERROR: return "ERROR" else: return "READY" except Exception as e: return f"ERROR: {str(e)}" # 在提交作业前检查状态 status = check_printer_status("MyPrinter") if status != "READY": print(f"打印机不可用，状态: {status}") # 将作业暂存或路由到备用打印机 ``` 2. **打印作业超时**：有些复杂的文档（如大型PDF）渲染时间很长，可能导致作业卡住。 ```python import threading import ctypes def print_with_timeout(file_path, printer_name, timeout=120): """带超时控制的打印函数""" def _print_task(): win32api.ShellExecute(0, "print", file_path, f'"{printer_name}"', ".", 0) print_thread = threading.Thread(target=_print_task) print_thread.start() print_thread.join(timeout=timeout) if print_thread.is_alive(): # 强制终止线程（注意：这不会取消已提交的打印作业） # 更好的做法是监控作业ID，然后使用SetJob取消 print("打印任务超时，可能需要手动检查打印机队列。") return False return True ``` **性能优化建议：** * **连接池**：频繁打开和关闭打印机句柄有开销。对于高吞吐量服务，可以维护一个小的打印机句柄连接池。 * **批量提交**：当需要打印多份相同文档时，如果打印机驱动支持，应使用作业级的“份数”设置，而不是多次提交相同作业。 * **异步与回调**：对于不需要即时结果的打印任务，使用完全异步的模式。主线程只负责将作业放入队列，由工作线程处理并回调通知结果。 * **内存管理**：打印大型文件时，避免一次性将整个文件读入内存。对于原始打印，可以分块读取和发送数据。 我在一个需要每天自动打印数千份质检报告的项目中使用了类似的架构。最初我们直接调用系统打印，经常遇到程序假死、队列堆积的问题。迁移到这种基于服务的异步模型后，系统的稳定性大幅提升，即使个别打印机故障，也不会影响其他任务的提交。最关键的是，我们终于有了清晰的日志，知道每一份文档“打印了吗？什么时候打的？用哪台打印机打的？”，这在审计时提供了极大的便利。