【python】云打印实现

这两天为了实现云打印功能找了很多相关的文章

记录一下这一篇，python云打印实现-朝花夕拾，代码通过监听文件夹有无产生新文件来判断是否执行，我尝试运行了下没问题，于是打算转载一下

程序运行结果

由于对方的代码和我实现的有点出入但都是先下载，后打印，只不过我的实通过mqtt服务器罢了，于是通过一步一步解析，我发现他这里有几个需要注意的，给大家和自己才个坑

1、settings.json 配置打印机和监控文件夹
2、不支持图片和pdf打印，有点难弄，后面处理了我会附上
3、不支持彩彩印规格这些
4、打印机用默认的即可，配置没多大作用，因为最后参数没有传过去

我改良后的

import time

import win32api
import win32print
from watchdog.observers import Observer
from watchdog.events import *
import json
import sys
import traceback
import os


def printer_loading(filename, printer):
    win32api.ShellExecute(
        0,
        "print",
        filename,
        '/d:"%s"' % win32print.GetDefaultPrinter(),
        ".",
        0
    )
'''
def printer_pic(pic_path, printer_name):
    # 物理宽度、高度
    PHYSICALWIDTH = 110
    PHYSICALHEIGHT = 111
    # 物理偏移位置
    PHYSICALOFFSETX = 112
    PHYSICALOFFSETY = 113
    printer_name = win32print.GetDefaultPrinter()
    hDC = win32ui.CreateDC()
    hDC.CreatePrinterDC(printer_name)
    printer_size = hDC.GetDeviceCaps(PHYSICALWIDTH), hDC.GetDeviceCaps(PHYSICALHEIGHT)
    # printer_margins = hDC.GetDeviceCaps (PHYSICALOFFSETX), hDC.GetDeviceCaps (PHYSICALOFFSETY)
    # 打开图片
    # ＃通过每个像素使它尽可能大
    # ＃页面不失真。
    bmp = Image.open(file_name)
    ratios = [1.0 * 1754 / bmp.size[0], 1.0 * 1240 / bmp.size[1]]
    scale = min(ratios)
    # ＃开始打印作业，并将位图绘制到
    # ＃按比例缩放打印机设备。
    hDC.StartDoc(file_name)
    hDC.StartPage()
    dib = ImageWin.Dib(bmp)
    scaled_width, scaled_height = [int(scale * i) for i in bmp.size]
    print(scaled_width, scaled_height)
    x1 = int((printer_size[0] - scaled_width) / 2)
    y1 = int((printer_size[1] - scaled_height) / 2)
    # 横向位置坐标
    x1 = 1580
    # 竖向位置坐标
    y1 = 30
    # 4倍为自适应图片实际尺寸打印
    x2 = x1 + bmp.size[0] * 4
    y2 = y1 + bmp.size[1] * 4
    dib.draw(hDC.GetHandleOutput(), (x1, y1, x2, y2))

    hDC.EndPage()
    hDC.EndDoc()
    hDC.DeleteDC()
'''
class FileEventHandler(FileSystemEventHandler):
    def __init__(self,printer):
        FileSystemEventHandler.__init__(self)
        self.printer = printer

    def on_created(self, event):
        print(event.is_directory)
        if event.is_directory:
            print("directory created:{0}".format(event.src_path))
        else:
            print("file created:{0}".format(event.src_path))
            if event.src_path.endswith("docx") or event.src_path.endswith("doc") or event.src_path.endswith("xlsx") or event.src_path.endswith("jpg") or event.src_path.endswith("pdf") or event.src_path.endswith("xls") or event.src_path.endswith("txt") or event.src_path.endswith("img"):
                if "~" not in event.src_path:
                    print(f"打印路径{event.src_path}")
                    print(f"打印机名字{self.printer}")
                    printer_loading(event.src_path, self.printer)

# import time
# win32api.ShellExecute(0, 'open', 'C:\\Program Files (x86)\\Allway Sync\\Bin\\syncappw.exe', '','',1)

if __name__ == "__main__":

    try:
        observers = []
        print("当前检测到的打印机:")
        maxL = 2
        Num = 2
        for i in range(1, 10):
            # print(i)
            r = win32print.EnumPrinters(i)
            l = len(r)
            if l > maxL:
                maxL = l
                Num = i

        for i, p in enumerate(list(win32print.EnumPrinters(Num))):
            print(f"\t{i}:{p[1]}\n")

        # f = open('.ttings.txt', 'r', encoding="utf-8")
        settings = {}
        # print("当前配置:")
        # for line in f.readlines():
        #     k, v = line.split(",")
        #     settings[k.strip()] = v.strip()
        #     print(f"\t{k}:{v}")



        print("当前配置:")
        with open("./settings.json", 'rb') as load_f:
            ds = json.load(load_f)
            for d in ds:
                settings[d['url'].strip()] = d["printer"].strip()
                print(f"\t{d}")
        pass
        for path, printer in settings.items():
            observer = Observer()
            print("打印机名"+printer)
            event_handler = FileEventHandler(printer)
            observer.schedule(event_handler, path, True)
            observer.start()
            observers.append(observer)

        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            for observer in observers:
                observer.stop()

        for observer in observers:
            observer.join()
    except Exception as e:
        print(e)
        traceback.print_exc()
        input()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

完整代码在这里：xxxxx

---

我顺藤摸瓜发现我们启动的方法不一样，我的是通过

 w = DispatchEx('kwps.Application')
 w.Visible = 0  # 不打开软件
 w.DisplayAlerts = 0  # 不报错
doc = w.Documents.Open(f)
1
2
3
4

而他的核心代码是通过

 # 打印输出
                    def printer_loading(filename, printer):
                        win32api.ShellExecute(
                            0,
                            "print",
                            filename,
                            '/d:"%s"' % win32print.GetDefaultPrinter(),
                            ".",
                            0
                        )
                    # 打印
1
2
3
4
5
6
7
8
9
10
11

暂时只知道第一个代码需要安装wps，第二个未测试

client_mqtt版（需要配合mqtt数据使用，不建议下载，自己用，需要安装wps-珠海zf版本）

client_mqtt_plus（需要配合mqtt数据使用，不建议下载，自己用，需要安装wps-珠海zf版本）

朝花夕拾-改良版（可以下载）