"""Windows 打印机管理服务
封装 win32print + win32ui 底层调用，提供打印机枚举、状态查询、
GDI 精确打印、PDF/图片/HTML 打印等功能。
64-bit Python 环境兼容，不依赖 Access/32位 ODBC。
"""
from __future__ import annotations
import contextlib
import json
import os
import subprocess
import tempfile
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
try:
import win32print
import win32ui
_HAS_WIN32 = True
except ImportError:
_HAS_WIN32 = False
win32print = None # type: ignore[assignment]
win32ui = None # type: ignore[assignment]
try:
from PySide6.QtCore import QObject, QTimer, Signal
except ImportError: # 分层：services 层在无 Qt（headless / 无 X11）环境仍需可导入
QObject = None # type: ignore[assignment]
QTimer = None # type: ignore[assignment]
Signal = None # type: ignore[assignment]
from certflow.config.paths_override import cfg
from certflow.utils.logger import logger
class _NullSignal:
"""无 Qt 环境下的 Signal 占位实现。
提供与 PySide6 ``Signal`` 兼容的最小接口：类体内 ``x = Signal(...)``
创建实例后，``.emit()`` 静默空转、``.connect()`` 返回自身，
使 ``PrinterManager`` 在非 GUI 环境下仍可导入与实例化。
"""
def __init__(self, *types: object) -> None:
self._types = types
def emit(self, *args: object) -> None:
return None
def connect(self, *args: object, **kwargs: object) -> _NullSignal:
return self
def disconnect(self, *args: object, **kwargs: object) -> None:
return None
def _make_signal(*types: object):
"""根据运行环境创建信号对象。
Args:
*types: 信号载荷类型（仅 Qt 环境下生效，用于类型校验）。
Returns:
真实 Qt Signal 或降级用的 _NullSignal 实例。
"""
if Signal is not None:
return Signal(*types)
return _NullSignal(*types)
[文档]
class PrinterStatus(Enum):
"""打印机状态枚举"""
READY = "就绪"
PAUSED = "暂停"
ERROR = "错误"
PENDING = "待机"
PAPER_OUT = "缺纸"
OFFLINE = "离线"
UNKNOWN = "未知"
[文档]
class PaperSize(Enum):
"""纸张大小枚举"""
CERTIFICATE_60x100 = (60, 100, "合格证标准 (60x100mm)")
CERTIFICATE_130x75 = (130, 75, "合格证 (130x75mm)")
A4 = (210, 297, "A4 (210x297mm)")
CUSTOM = (0, 0, "自定义")
def __init__(self, width: float, height: float, display_name: str) -> None:
self.width = width
self.height = height
self.display_name = display_name
[文档]
@classmethod
def from_name(cls, name: str) -> PaperSize:
"""根据名称查找纸张规格。
Args:
name: 显示名称或枚举名（如 "A4"、"合格证标准 (60x100mm)"）。
Returns:
PaperSize: 匹配到的纸张规格，未匹配时返回 PaperSize.CUSTOM。
"""
for size in cls:
if size.display_name == name or size.name == name:
return size
return cls.CUSTOM
[文档]
@dataclass
class PrinterInfo:
"""打印机信息"""
name: str
port: str = ""
driver: str = ""
status: PrinterStatus = PrinterStatus.UNKNOWN
is_default: bool = False
paper_sizes: list[str] = field(default_factory=list)
[文档]
def to_dict(self) -> dict:
"""序列化为字典。
Returns:
dict: 包含 name/port/driver/status/is_default/paper_sizes 的字典。
"""
return {
"name": self.name,
"port": self.port,
"driver": self.driver,
"status": self.status.value,
"is_default": self.is_default,
"paper_sizes": self.paper_sizes,
}
[文档]
@dataclass
class PrintJob:
"""打印任务信息"""
job_id: int
document_name: str
printer_name: str
pages: int = 0
copies: int = 1
status: str = ""
submitted_time: str = ""
# ============================================================
# GDI 打印坐标常量（60mm × 100mm 预印卡纸）
# ============================================================
MM2PT = 2.83465 # 1mm = 2.83465 points
PAPER_WIDTH_MM = 60
PAPER_HEIGHT_MM = 100
PAPER_SIZE_PT = (PAPER_WIDTH_MM * MM2PT, PAPER_HEIGHT_MM * MM2PT)
# 文字打印位置（从卡纸左上角为原点）
LABEL_X = 3 # 标签列 X 起点 (mm)
VALUE_X = 20 # 值列 X 起点 (mm)
START_Y = 22 # 第一行 Y 起点 (mm)
LINE_SPACING = 6.5 # 行间距 (mm)
FONT_SIZE_MM = 3.0 # 字体高度 (mm)
# 打印字段定义: (标签, 数据key, Y偏移序号)
PRINT_FIELDS: list[tuple[str, str, int]] = [
("产品名称：", "product_name", 0),
("产品型号：", "product_model", 1),
("公称通径：", "dn", 2),
("公称压力：", "pn", 3),
("工作温度：", "temperature", 4),
("适用介质：", "medium", 5),
("检验标准：", "check_standard", 6),
("检验工号：", "inspector_id", 7),
("出厂日期：", "manufacture_date", 8),
("产品编号：", "serial_number", 9),
]
[文档]
class PrinterManager(QObject if QObject is not None else object):
"""打印机管理器
封装 Windows 打印子系统，提供打印机枚举、状态监控、
GDI 精确文字打印、文件打印等功能。
分层说明：本模块归属 services 层，不应被 UI 库强制绑定导入。
在无 PySide6（headless / 无 X11）环境下，信号与监控定时器自动降级为空操作，
模块仍可导入与实例化，仅丢失 GUI 事件通知能力。
Signals（仅 Qt 环境生效）:
printer_added: 新增打印机 (name)
printer_removed: 移除打印机 (name)
printer_status_changed: 状态变化 (name, status)
print_started: 打印开始 (doc_name, job_id)
print_completed: 打印完成 (doc_name, job_id)
print_failed: 打印失败 (doc_name, error_msg)
progress_updated: 进度更新 (current, total)
"""
printer_added = _make_signal(str)
printer_removed = _make_signal(str)
printer_status_changed = _make_signal(str, PrinterStatus)
print_started = _make_signal(str, int)
print_completed = _make_signal(str, int)
print_failed = _make_signal(str, str)
progress_updated = _make_signal(int, int)
def __init__(self) -> None:
if QObject is not None:
super().__init__()
self.printers: dict[str, PrinterInfo] = {}
self.current_printer: str | None = None
self.pending_jobs: list[PrintJob] = []
self._init_printers()
self._start_monitor()
# ================================================================
# 初始化 & 枚举
# ================================================================
def _init_printers(self) -> None:
"""初始化打印机列表"""
try:
printers = win32print.EnumPrinters(
win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS,
None,
1,
)
default_printer = win32print.GetDefaultPrinter()
for printer in printers:
info = self._get_printer_info(printer[2], default_printer)
if info:
self.printers[info.name] = info
if info.name == default_printer:
self.current_printer = info.name
logger.debug(
f"打印机初始化完成: {len(self.printers)} 台, 默认={self.current_printer}",
)
except Exception as e:
logger.warning(f"打印机初始化失败: {e}")
def _get_printer_info(self, printer_name: str, default_printer: str) -> PrinterInfo | None:
"""获取单台打印机详细信息"""
try:
handle = win32print.OpenPrinter(printer_name)
info = win32print.GetPrinter(handle, 2)
win32print.ClosePrinter(handle)
status = self._get_printer_status(info)
paper_sizes = self._get_paper_sizes(printer_name)
return PrinterInfo(
name=printer_name,
port=info.get("pPortName", ""),
driver=info.get("pDriverName", ""),
status=status,
is_default=(printer_name == default_printer),
paper_sizes=paper_sizes,
)
except Exception as e:
logger.debug(f"获取打印机信息失败 [{printer_name}]: {e}")
return None
@staticmethod
def _get_printer_status(printer_info: dict) -> PrinterStatus:
"""解析打印机状态"""
status = printer_info.get("Status", 0)
if status & win32print.PRINTER_STATUS_OFFLINE:
return PrinterStatus.OFFLINE
if status & win32print.PRINTER_STATUS_PAPER_OUT:
return PrinterStatus.PAPER_OUT
if status & win32print.PRINTER_STATUS_PAUSED:
return PrinterStatus.PAUSED
if status & win32print.PRINTER_STATUS_ERROR:
return PrinterStatus.ERROR
if status & win32print.PRINTER_STATUS_PENDING_DELETION:
return PrinterStatus.PENDING
return PrinterStatus.READY
@staticmethod
def _get_paper_sizes(printer_name: str) -> list[str]:
"""获取打印机支持的纸张大小"""
paper_sizes = ["A4", "A5", "Letter"]
paper_sizes.extend(["60x100mm", "130x75mm"])
return paper_sizes
# ================================================================
# 监控
# ================================================================
def _start_monitor(self) -> None:
"""启动打印机状态监控（每10秒刷新）。
分层降级：无 PySide6 / QTimer 的 headless 环境下跳过监控，
避免 UI 库硬依赖阻断 services 层导入。
"""
if QTimer is None:
return
self.monitor_timer = QTimer()
self.monitor_timer.timeout.connect(self._refresh_status)
self.monitor_timer.start(10000)
def _refresh_status(self) -> None:
"""刷新所有打印机状态"""
try:
default_printer = win32print.GetDefaultPrinter()
except Exception:
return
for printer_name in list(self.printers.keys()):
new_info = self._get_printer_info(printer_name, default_printer)
if new_info:
old_status = self.printers[printer_name].status
if old_status != new_info.status:
self.printers[printer_name] = new_info
self.printer_status_changed.emit(printer_name, new_info.status)
# ================================================================
# 公共 API
# ================================================================
[文档]
def get_printers(self) -> list[PrinterInfo]:
"""获取所有打印机。
Returns:
list[PrinterInfo]: 当前已枚举的打印机信息列表。
"""
return list(self.printers.values())
[文档]
def get_printer_names(self) -> list[str]:
"""获取打印机名称列表。
Returns:
list[str]: 打印机名称列表。
"""
return list(self.printers.keys())
[文档]
def get_default_printer(self) -> str | None:
"""获取默认打印机名称。
Returns:
str | None: 默认打印机名称，获取失败时回退到当前打印机。
"""
try:
return win32print.GetDefaultPrinter()
except Exception:
return self.current_printer
[文档]
def set_default_printer(self, printer_name: str) -> bool:
"""设置默认打印机。
Args:
printer_name: 目标打印机名称。
Returns:
bool: 设置是否成功。
"""
try:
win32print.SetDefaultPrinter(printer_name)
self.current_printer = printer_name
for name, info in self.printers.items():
info.is_default = name == printer_name
return True
except Exception as e:
logger.warning(f"设置默认打印机失败: {e}")
return False
[文档]
def get_printer_status(self, printer_name: str | None = None) -> PrinterStatus:
"""获取打印机状态。
Args:
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
PrinterStatus: 对应打印机状态，未知时返回 PrinterStatus.UNKNOWN。
"""
name = printer_name or self.get_default_printer()
if name and name in self.printers:
return self.printers[name].status
return PrinterStatus.UNKNOWN
[文档]
def is_printer_ready(self, printer_name: str | None = None) -> bool:
"""检查打印机是否就绪。
Args:
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
bool: True 表示打印机处于就绪状态。
"""
return self.get_printer_status(printer_name) == PrinterStatus.READY
# ================================================================
# GDI 精确文字打印（核心方法）
# ================================================================
[文档]
def print_text_only(
self,
data: dict[str, Any],
serial_number: str,
printer_name: str | None = None,
) -> bool:
"""使用 win32 GDI 精确打印文字到预印卡纸
在 60mm × 100mm 预印卡纸的指定坐标位置输出文字。
卡纸已预印底板（边框、标题、品保章等），打印机只输出动态文字。
跳过空值字段，避免覆盖卡纸原有内容。
Args:
data: 打印数据字典，包含 product_name, product_model, dn, pn,
temperature, medium, check_standard, inspector_id, manufacture_date
serial_number: 产品编号（逐台打印时不同）
printer_name: 打印机名称，默认使用系统默认打印机
Returns:
bool: 打印是否成功
"""
hprinter = None
try:
printer = printer_name or self.get_default_printer()
if not printer:
logger.error("没有可用打印机")
return False
logger.debug(f"GDI 打印: 编号={serial_number}, 打印机={printer}")
# 打开打印机
hprinter = win32print.OpenPrinter(printer)
devmode = win32print.GetPrinter(hprinter, 2)["pDevMode"]
# 设置自定义纸张 60mm × 100mm
devmode.PaperSize = 256 # DMPAPER_USER
devmode.PaperWidth = int(PAPER_WIDTH_MM * 10) # 十分之一毫米
devmode.PaperLength = int(PAPER_HEIGHT_MM * 10)
hdc = win32ui.CreateDC()
hdc.CreatePrinterDC(printer)
# 开始打印任务
doc_name = f"合格证_{serial_number}"
hdc.StartDoc(doc_name)
hdc.StartPage()
# 获取设备分辨率
dpi_x = hdc.GetDeviceCaps(88) # LOGPIXELSX
def mm_to_px(mm_val: float) -> int:
return int(mm_val * dpi_x / 25.4)
# 创建逻辑字体
font_size_px = mm_to_px(FONT_SIZE_MM)
font = win32ui.CreateFont(
{
"name": "SimSun",
"height": font_size_px,
"weight": 400,
}
)
hdc.SelectObject(font)
# 设置透明背景 + 左对齐
hdc.SetTextAlign(0) # TA_LEFT | TA_TOP
hdc.SetBkMode(1) # TRANSPARENT
# 逐字段绘制
field_count = 0
for label, key, offset_idx in PRINT_FIELDS:
value = data.get(key, "") or ""
if key == "serial_number":
value = serial_number
# 跳过空值，避免打印空白覆盖卡纸原有内容
if not str(value).strip():
continue
y_mm = START_Y + offset_idx * LINE_SPACING
x_label_px = mm_to_px(LABEL_X)
x_value_px = mm_to_px(VALUE_X)
y_px = mm_to_px(y_mm)
# 绘制标签
hdc.TextOut(x_label_px, y_px, label)
# 绘制值
hdc.TextOut(x_value_px, y_px, str(value))
field_count += 1
hdc.EndPage()
hdc.EndDoc()
hdc.DeleteDC()
logger.info(f"GDI 打印成功: {serial_number} (共 {field_count} 个字段)")
return True
except Exception as e:
logger.error(f"GDI 打印失败 [{serial_number}]: {e}")
return False
finally:
if hprinter is not None:
with contextlib.suppress(Exception):
win32print.ClosePrinter(hprinter)
# ================================================================
# 文件打印
# ================================================================
[文档]
def print_file(
self,
file_path: str,
printer_name: str | None = None,
copies: int = 1,
) -> tuple[bool, int | None]:
"""打印文件（PDF/图片/文本）。
Args:
file_path: 待打印文件的完整路径。
printer_name: 目标打印机名称，为 None 时使用默认打印机。
copies: 打印份数。
Returns:
tuple[bool, int | None]: (是否成功, 作业 ID)，失败时作业 ID 为 None。
"""
if not os.path.exists(file_path):
self.print_failed.emit(file_path, "文件不存在")
return False, None
printer = printer_name or self.get_default_printer()
if not printer:
self.print_failed.emit(file_path, "未找到可用打印机")
return False, None
try:
ext = os.path.splitext(file_path)[1].lower()
if ext == ".pdf":
return self._print_pdf(file_path, printer, copies)
if ext in (".jpg", ".jpeg", ".png", ".bmp"):
return self._print_image(file_path, printer, copies)
if ext == ".txt":
return self._print_text_file(file_path, printer)
return self._print_with_shell(file_path, printer)
except Exception as e:
self.print_failed.emit(file_path, str(e))
return False, None
def _print_pdf(self, pdf_path: str, printer_name: str, copies: int) -> tuple[bool, int | None]:
"""通过 Adobe Reader 或 Edge 打印 PDF"""
try:
adobe_path = r"C:\Program Files\Adobe\Acrobat DC\Acrobat\Acrobat.exe"
if os.path.exists(adobe_path):
cmd = [adobe_path, "/t", pdf_path, printer_name]
else:
cmd = [
"msedge.exe",
"--headless",
f"--print-to-file={pdf_path.replace('.pdf', '_out.pdf')}",
]
subprocess.run(cmd, capture_output=True, text=True)
self.print_completed.emit(pdf_path, 0)
return True, 0
except Exception:
return self._print_with_shell(pdf_path, printer_name)
def _print_image(
self, image_path: str, printer_name: str, copies: int
) -> tuple[bool, int | None]:
"""通过 PIL + win32ui 打印图片"""
try:
from PIL import Image, ImageWin
image = Image.open(image_path)
printer_dc = win32ui.CreateDC()
printer_dc.CreatePrinterDC(printer_name)
printer_dc.StartDoc(os.path.basename(image_path))
printer_dc.StartPage()
printable_area = (
printer_dc.GetDeviceCaps(110),
printer_dc.GetDeviceCaps(111),
)
dib = ImageWin.Dib(image)
scaled_width = printable_area[0]
scaled_height = int(image.height * scaled_width / image.width)
if scaled_height > printable_area[1]:
scaled_height = printable_area[1]
scaled_width = int(image.width * scaled_height / image.height)
x = (printable_area[0] - scaled_width) // 2
y = (printable_area[1] - scaled_height) // 2
dib.draw(printer_dc.GetHandleOutput(), (x, y, x + scaled_width, y + scaled_height))
printer_dc.EndPage()
printer_dc.EndDoc()
printer_dc.DeleteDC()
self.print_completed.emit(image_path, 0)
return True, 0
except Exception as e:
logger.warning(f"图片打印失败: {e}")
return self._print_with_shell(image_path, printer_name)
def _print_text_file(self, text_path: str, printer_name: str) -> tuple[bool, int | None]:
"""通过记事本打印文本文件"""
try:
subprocess.run(["notepad.exe", "/p", text_path], capture_output=True)
self.print_completed.emit(text_path, 0)
return True, 0
except Exception:
return False, None
def _print_with_shell(self, file_path: str, printer_name: str) -> tuple[bool, int | None]:
"""通过 Shell 命令打印（兜底方案）"""
try:
ps_command = (
f'Copy-Item -Path "{file_path}" '
f'-Destination "\\\\{printer_name}\\$'
f"([System.IO.Path]::GetFileName('{file_path}'))\""
)
subprocess.run(["powershell", "-Command", ps_command], capture_output=True)
self.print_completed.emit(file_path, 0)
return True, 0
except Exception as e:
self.print_failed.emit(file_path, str(e))
return False, None
[文档]
def print_html(
self,
html_content: str,
printer_name: str | None = None,
copies: int = 1,
) -> tuple[bool, int | None]:
"""打印 HTML 内容（先转换为 PDF 再打印）。
Args:
html_content: HTML 字符串内容。
printer_name: 目标打印机名称，为 None 时使用默认打印机。
copies: 打印份数。
Returns:
tuple[bool, int | None]: (是否成功, 作业 ID)，失败时作业 ID 为 None。
"""
try:
with tempfile.NamedTemporaryFile(
mode="w", suffix=".html", delete=False, encoding="utf-8"
) as f:
f.write(html_content)
temp_file = f.name
pdf_file = temp_file.replace(".html", ".pdf")
cmd = [
"msedge.exe",
"--headless",
f"--print-to-file={pdf_file}",
f"--printer-name={printer_name or self.get_default_printer()}",
temp_file,
]
subprocess.run(cmd, capture_output=True)
if os.path.exists(pdf_file):
result = self.print_file(pdf_file, printer_name, copies)
os.unlink(pdf_file)
os.unlink(temp_file)
return result if "result" in dir() else (False, None)
except Exception as e:
self.print_failed.emit("HTML内容", str(e))
return False, None
# ================================================================
# 打印队列管理
# ================================================================
[文档]
def get_print_queue(self, printer_name: str | None = None) -> list[PrintJob]:
"""获取打印队列。
Args:
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
list[PrintJob]: 当前打印队列中的任务列表。
"""
jobs: list[PrintJob] = []
try:
name = printer_name or self.get_default_printer()
if name:
handle = win32print.OpenPrinter(name)
jobs_info = win32print.EnumJobs(handle, 0, 10, 1)
win32print.ClosePrinter(handle)
for job in jobs_info:
jobs.append(
PrintJob(
job_id=job["JobId"],
document_name=job["pDocument"],
printer_name=name,
pages=job.get("TotalPages", 0),
copies=job.get("Copies", 1),
status=self._get_job_status(job.get("Status", 0)),
submitted_time=str(job.get("Submitted", "")),
)
)
except Exception as e:
logger.debug(f"获取打印队列失败: {e}")
return jobs
@staticmethod
def _get_job_status(status_code: int) -> str:
if status_code & win32print.JOB_STATUS_PRINTING:
return "打印中"
if status_code & win32print.JOB_STATUS_PAUSED:
return "已暂停"
if status_code & win32print.JOB_STATUS_ERROR:
return "错误"
if status_code & win32print.JOB_STATUS_DELETING:
return "删除中"
if status_code & win32print.JOB_STATUS_SPOOLING:
return "后台处理"
return "等待中"
[文档]
def cancel_job(self, job_id: int, printer_name: str | None = None) -> bool:
"""取消打印任务。
Args:
job_id: 打印作业 ID。
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
bool: 取消是否成功。
"""
try:
name = printer_name or self.get_default_printer()
if name:
handle = win32print.OpenPrinter(name)
win32print.SetJob(handle, job_id, 0, None, win32print.JOB_CONTROL_CANCEL)
win32print.ClosePrinter(handle)
return True
except Exception as e:
logger.warning(f"取消打印任务失败: {e}")
return False
[文档]
def pause_printer(self, printer_name: str | None = None) -> bool:
"""暂停打印机。
Args:
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
bool: 暂停是否成功。
"""
try:
name = printer_name or self.get_default_printer()
if name:
handle = win32print.OpenPrinter(name)
win32print.SetPrinter(handle, 0, None, win32print.PRINTER_CONTROL_PAUSE)
win32print.ClosePrinter(handle)
return True
except Exception as e:
logger.warning(f"暂停打印机失败: {e}")
return False
[文档]
def resume_printer(self, printer_name: str | None = None) -> bool:
"""恢复打印机。
Args:
printer_name: 打印机名称，为 None 时使用默认打印机。
Returns:
bool: 恢复是否成功。
"""
try:
name = printer_name or self.get_default_printer()
if name:
handle = win32print.OpenPrinter(name)
win32print.SetPrinter(handle, 0, None, win32print.PRINTER_CONTROL_RESUME)
win32print.ClosePrinter(handle)
return True
except Exception as e:
logger.warning(f"恢复打印机失败: {e}")
return False
# ================================================================
# 设置持久化
# ================================================================
[文档]
def save_settings(self, settings_file: str = "printer_settings.json") -> None:
"""保存打印机设置到 JSON 文件。
Args:
settings_file: 设置文件路径，默认为 "printer_settings.json"。
"""
settings = {
"default_printer": self.get_default_printer(),
"printers": {name: info.to_dict() for name, info in self.printers.items()},
}
with open(settings_file, "w", encoding="utf-8") as f:
json.dump(settings, f, ensure_ascii=False, indent=2)
[文档]
def load_settings(self, settings_file: str = "printer_settings.json") -> None:
"""从 JSON 文件加载并设置默认打印机。
Args:
settings_file: 设置文件路径，默认为 "printer_settings.json"。
"""
if os.path.exists(settings_file):
with open(settings_file, encoding="utf-8") as f:
settings = json.load(f)
default_printer = settings.get("default_printer")
if default_printer:
self.set_default_printer(default_printer)
[文档]
@staticmethod
def cleanup_temp_files() -> None:
"""清理系统中本模块产生的临时 HTML/PDF 文件。"""
temp_dir = cfg("paths.temp.base", tempfile.gettempdir())
for file in os.listdir(temp_dir):
if file.startswith("tmp") and (file.endswith(".html") or file.endswith(".pdf")):
with contextlib.suppress(OSError):
os.unlink(os.path.join(temp_dir, file))