1117 lines
40 KiB
Python
1117 lines
40 KiB
Python
import os
|
||
import re
|
||
import io
|
||
import sys
|
||
import threading
|
||
import clr
|
||
import time
|
||
import json
|
||
import queue
|
||
import shutil
|
||
import sqlite3
|
||
import logging
|
||
import tempfile
|
||
import pywintypes
|
||
import hashlib
|
||
import win32com.client
|
||
import win32print
|
||
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QLabel, QComboBox, QCheckBox, QLineEdit, QAction, QMenu, \
|
||
QMessageBox, QPushButton, QVBoxLayout, QHBoxLayout, QFileDialog, QDialog
|
||
from PyQt5.QtCore import Qt, QCoreApplication, QTimer, QPropertyAnimation
|
||
from PyQt5.QtGui import QPixmap, QIntValidator
|
||
from threading import RLock
|
||
|
||
from flask import Flask, request
|
||
|
||
flask_app = Flask(__name__)
|
||
|
||
|
||
def _flask_thread():
|
||
try:
|
||
flask_app.run(host='0.0.0.0', port=7746)
|
||
except:
|
||
pass
|
||
|
||
|
||
flask_thread = threading.Thread(target=_flask_thread)
|
||
|
||
sys.path.append(os.path.dirname(__file__))
|
||
clr.AddReference('BarTender')
|
||
try:
|
||
import BarTender
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def killThread(thread):
|
||
import ctypes
|
||
try:
|
||
if thread.is_alive():
|
||
return ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), ctypes.py_object(BaseException)) == 1
|
||
else:
|
||
return True
|
||
except BaseException:
|
||
pass
|
||
|
||
|
||
def cp(src: str, dst: str):
|
||
"""
|
||
Support file or directory.
|
||
"""
|
||
return shutil.copytree(src, dst) and True if os.path.isdir(src) else shutil.copy(src, dst) and True
|
||
|
||
|
||
def mv(src: str, dst: str):
|
||
"""
|
||
Support file or directory.
|
||
"""
|
||
return shutil.move(src, dst) and True
|
||
|
||
|
||
def rm(src: str):
|
||
"""
|
||
Support file or directory.
|
||
"""
|
||
shutil.rmtree(src) if os.path.isdir(src) else os.remove(src)
|
||
return True
|
||
|
||
|
||
def rename(path: str, filename: str):
|
||
"""
|
||
Support file or directory.
|
||
"""
|
||
if filename.find('/') != -1:
|
||
raise Exception('File name, catalog name or scroll grammar incorrect grammar.')
|
||
return os.rename(path, os.path.dirname(path) + '/' + filename) or True
|
||
|
||
|
||
def exists(path: str):
|
||
"""
|
||
Support file or directory.
|
||
"""
|
||
return os.path.exists(path)
|
||
|
||
|
||
def mkdirs(path: str):
|
||
"""
|
||
Support multi-level directory.
|
||
"""
|
||
return os.makedirs(path)
|
||
|
||
|
||
def _fd(f):
|
||
return f.fileno() if hasattr(f, 'fileno') else f
|
||
|
||
|
||
if os.name == 'nt':
|
||
import msvcrt
|
||
from ctypes import (sizeof, c_ulong, c_void_p, c_int64, Structure, Union, POINTER, windll, byref)
|
||
from ctypes.wintypes import BOOL, DWORD, HANDLE
|
||
|
||
LOCK_SH = 0x0
|
||
LOCK_NB = 0x1
|
||
LOCK_EX = 0x2
|
||
LOCK_UN = 0x9
|
||
|
||
if sizeof(c_ulong) != sizeof(c_void_p):
|
||
ULONG_PTR = c_int64
|
||
else:
|
||
ULONG_PTR = c_ulong
|
||
|
||
PVOID = c_void_p
|
||
|
||
|
||
class _OFFSET(Structure):
|
||
_fields_ = [
|
||
('Offset', DWORD),
|
||
('OffsetHigh', DWORD)
|
||
]
|
||
|
||
|
||
class _OFFSET_UNION(Union):
|
||
_fields_ = [
|
||
('_offset', _OFFSET),
|
||
('Pointer', PVOID)
|
||
]
|
||
_anonymous_ = ['_offset']
|
||
|
||
|
||
class OVERLAPPED(Structure):
|
||
_fields_ = [
|
||
('Internal', ULONG_PTR),
|
||
('InternalHigh', ULONG_PTR),
|
||
('_offset_union', _OFFSET_UNION),
|
||
('hEvent', HANDLE)
|
||
]
|
||
_anonymous_ = ['_offset_union']
|
||
|
||
|
||
LPOVERLAPPED = POINTER(OVERLAPPED)
|
||
LockFileEx = windll.kernel32.LockFileEx
|
||
LockFileEx.restype = BOOL
|
||
LockFileEx.argtypes = [HANDLE, DWORD, DWORD, DWORD, DWORD, LPOVERLAPPED]
|
||
UnlockFileEx = windll.kernel32.UnlockFileEx
|
||
UnlockFileEx.restype = BOOL
|
||
UnlockFileEx.argtypes = [HANDLE, DWORD, DWORD, DWORD, LPOVERLAPPED]
|
||
|
||
|
||
def flock(f, flags):
|
||
hfile = msvcrt.get_osfhandle(_fd(f))
|
||
overlapped = OVERLAPPED()
|
||
if flags == LOCK_UN:
|
||
ret = UnlockFileEx(
|
||
hfile,
|
||
0,
|
||
0,
|
||
0xFFFF0000,
|
||
byref(overlapped)
|
||
)
|
||
else:
|
||
ret = LockFileEx(
|
||
hfile,
|
||
flags,
|
||
0,
|
||
0,
|
||
0xFFFF0000,
|
||
byref(overlapped)
|
||
)
|
||
return bool(ret)
|
||
else:
|
||
try:
|
||
import fcntl
|
||
|
||
LOCK_SH = fcntl.LOCK_SH
|
||
LOCK_NB = fcntl.LOCK_NB
|
||
LOCK_EX = fcntl.LOCK_EX
|
||
LOCK_UN = fcntl.LOCK_UN
|
||
except (ImportError, AttributeError):
|
||
LOCK_EX = LOCK_SH = LOCK_NB = 0
|
||
|
||
|
||
def flock(f, flags):
|
||
return flags == LOCK_UN
|
||
else:
|
||
def flock(f, flags):
|
||
return fcntl.flock(_fd(f), flags) == 0
|
||
|
||
|
||
def getJson(file, data=None):
|
||
if os.path.exists(file):
|
||
try:
|
||
return json.loads(open(file=file, mode='r', encoding='utf-8').read())
|
||
except json.decoder.JSONDecodeError:
|
||
return data
|
||
return data
|
||
|
||
|
||
def putJson(file, data=None):
|
||
with open(file=file, mode='w', encoding='utf-8') as f:
|
||
flock(f, LOCK_EX)
|
||
res = f.write(json.dumps(data, indent=4, ensure_ascii=True))
|
||
flock(f, LOCK_UN)
|
||
return res
|
||
|
||
|
||
class SQLite3:
|
||
__conn__ = None
|
||
__curr__ = None
|
||
__data__ = None
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
if args or kwargs:
|
||
self.connect(*args, **kwargs)
|
||
|
||
def _check_connection(self):
|
||
if self.__curr__ is None:
|
||
raise Exception('Connection does not exist.')
|
||
|
||
@staticmethod
|
||
def _sqlite_dict_factory(cursor, row):
|
||
d = {}
|
||
for idx, col in enumerate(cursor.description):
|
||
d[col[0]] = row[idx]
|
||
return d
|
||
|
||
def connect(self, database):
|
||
if self.__conn__:
|
||
raise Exception('Already Connect.')
|
||
self.__conn__ = sqlite3.connect(database, timeout=10, check_same_thread=False)
|
||
self.__conn__.row_factory = self._sqlite_dict_factory
|
||
self.__curr__ = self.__conn__.cursor()
|
||
return self
|
||
|
||
def execute(self, *args, **kwargs):
|
||
self._check_connection()
|
||
self.__curr__.execute(*args, **kwargs)
|
||
self.__conn__.commit()
|
||
return self
|
||
|
||
def rownums(self):
|
||
self._check_connection()
|
||
return self.__curr__.rowcount
|
||
|
||
def fetchall(self):
|
||
self._check_connection()
|
||
return self.__curr__.fetchall()
|
||
|
||
def fetchone(self):
|
||
self._check_connection()
|
||
return self.__curr__.fetchone()
|
||
|
||
def close(self):
|
||
if self.__curr__ or self.__conn__:
|
||
self.__curr__.close()
|
||
self.__conn__.close()
|
||
self.__curr__ = None
|
||
self.__conn__ = None
|
||
return True
|
||
|
||
|
||
class Ditto:
|
||
def __init__(self, db_file, rlock=None, class_name='Ditto-Default', limit_time=None, limit_rows=None):
|
||
self.rLock = rlock or RLock()
|
||
if limit_time is not None and not limit_time >= 0: raise Exception('Range error.')
|
||
if limit_rows is not None and not limit_rows >= 0: raise Exception('Range error.')
|
||
self.class_name = class_name
|
||
self.limit_rows = limit_rows
|
||
self.limit_time = limit_time
|
||
try:
|
||
self.database = SQLite3(db_file)
|
||
self.init_database()
|
||
except sqlite3.DatabaseError:
|
||
self.database.close() and os.remove(db_file)
|
||
raise Exception('The database was damaged and has been reset.')
|
||
|
||
def init_database(self):
|
||
with self.rLock:
|
||
self.database.execute('''
|
||
CREATE TABLE IF NOT EXISTS "ditto" (
|
||
"id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
|
||
"class" TEXT(64) COLLATE NOCASE,
|
||
"value" TEXT(256) COLLATE NOCASE,
|
||
"time" integer(12)
|
||
);
|
||
''')
|
||
|
||
def fresh(self):
|
||
with self.rLock:
|
||
if self.limit_time is not None: self.database.execute('''
|
||
DELETE FROM ditto WHERE class='%s' AND time < %s;
|
||
''' % (self.class_name, (int(time.time()) - self.limit_time)))
|
||
if self.limit_rows is not None: self.database.execute('''
|
||
DELETE FROM ditto WHERE id IN (SELECT id FROM (SELECT id FROM ditto WHERE class='%s' ORDER BY time DESC LIMIT %s,100));
|
||
''' % (self.class_name, self.limit_rows))
|
||
|
||
def count(self):
|
||
with self.rLock:
|
||
self.fresh()
|
||
self.database.execute('''
|
||
SELECT COUNT(1) AS rows FROM ditto WHERE class='%s';
|
||
''' % (self.class_name,))
|
||
return self.database.fetchone()['rows']
|
||
|
||
def clear(self):
|
||
with self.rLock:
|
||
self.fresh()
|
||
self.database.execute('''
|
||
DELETE FROM ditto WHERE class='%s';
|
||
''' % (self.class_name,))
|
||
return self.database.rownums()
|
||
|
||
def query(self, value):
|
||
with self.rLock:
|
||
self.fresh()
|
||
self.database.execute('''
|
||
SELECT COUNT(1) AS rows FROM ditto WHERE class='%s' AND value='%s';
|
||
''' % (self.class_name, value))
|
||
return self.database.fetchone()['rows']
|
||
|
||
def addit(self, value):
|
||
with self.rLock:
|
||
if self.query(value) != 0: return False
|
||
self.database.execute('''
|
||
INSERT INTO ditto (\"class\", \"value\", \"time\") VALUES ('%s', '%s', '%s');
|
||
''' % (self.class_name, value, int(time.time())))
|
||
self.fresh()
|
||
return True
|
||
|
||
|
||
class LoggerFileHandler:
|
||
def __init__(self, log_file: str, mode: str = 'a', level: str = None, fmt: str = None):
|
||
self.log, self.mod = log_file, mode
|
||
self.lev, self.fmt = level, fmt
|
||
self.sty = '%'
|
||
|
||
|
||
class LoggerConsHandler:
|
||
def __init__(self, level: str = None, fmt: str = None):
|
||
self.lev, self.fmt = level, fmt
|
||
self.sty = '%'
|
||
|
||
|
||
class Logger:
|
||
logger = None
|
||
levels = {
|
||
'CRITICAL': logging.CRITICAL,
|
||
'FATAL': logging.FATAL,
|
||
'ERROR': logging.ERROR,
|
||
'WARNING': logging.WARNING,
|
||
'WARN': logging.WARN,
|
||
'INFO': logging.INFO,
|
||
'DEBUG': logging.DEBUG,
|
||
'NOTSET': logging.NOTSET,
|
||
'D': logging.DEBUG,
|
||
'I': logging.INFO,
|
||
'W': logging.WARNING,
|
||
'E': logging.ERROR,
|
||
'F': logging.FATAL
|
||
}
|
||
default_format = '{asctime} - {name} - {levelname[0]}: {message}'
|
||
default_format_style = '{'
|
||
handler_list = []
|
||
|
||
def __init__(
|
||
self,
|
||
name: str = 'default',
|
||
default_level='DEBUG',
|
||
fh: LoggerFileHandler = None,
|
||
ch: LoggerConsHandler = None,
|
||
add_default_handler=False
|
||
):
|
||
ch = LoggerConsHandler() if add_default_handler and not ch else ch
|
||
if fh and not isinstance(fh, LoggerFileHandler):
|
||
raise TypeError('The parameter fh must be <LoggerFileHandler> type.')
|
||
if ch and not isinstance(ch, LoggerConsHandler):
|
||
raise TypeError('The parameter ch must be <LoggerConsHandler> type.')
|
||
|
||
self.logger = logging.getLogger(name)
|
||
self.logger.setLevel(self.levels[default_level])
|
||
|
||
if fh:
|
||
fhandler = logging.FileHandler(filename=fh.log, mode=fh.mod, encoding='utf-8')
|
||
self.handler_list.append(fhandler)
|
||
fhandler.setLevel(self.levels[fh.lev or default_level])
|
||
fh.fmt = fh.fmt or self.default_format
|
||
fh.sty = '{' if '%' not in fh.fmt else '%'
|
||
fhandler.setFormatter(logging.Formatter(fmt=fh.fmt, style=fh.sty))
|
||
self.logger.addHandler(fhandler)
|
||
if ch:
|
||
chandler = logging.StreamHandler()
|
||
self.handler_list.append(chandler)
|
||
chandler.setLevel(self.levels[ch.lev or default_level])
|
||
ch.fmt = ch.fmt or self.default_format
|
||
ch.sty = '{' if '%' not in ch.fmt else '%'
|
||
chandler.setFormatter(logging.Formatter(fmt=ch.fmt, style=ch.sty))
|
||
self.logger.addHandler(chandler)
|
||
|
||
self.d = self.logger.debug
|
||
self.i = self.logger.info
|
||
self.w = self.logger.warning
|
||
self.e = self.logger.error
|
||
self.f = self.logger.fatal
|
||
self.c = self.logger.critical
|
||
|
||
|
||
class Setting(dict):
|
||
def __init__(self, setting_file: str, setting_default: dict):
|
||
self.data_file = os.path.abspath(setting_file)
|
||
super().__init__(getJson(self.data_file, setting_default))
|
||
|
||
def __getitem__(self, item):
|
||
if item in self:
|
||
return super().__getitem__(item)
|
||
else:
|
||
return None
|
||
|
||
def __setitem__(self, key, value):
|
||
super().__setitem__(key, value)
|
||
putJson(self.data_file, self)
|
||
|
||
|
||
class BarTenderPrint:
|
||
def __init__(self, logger: Logger):
|
||
self.logger = logger
|
||
self.label = None
|
||
self.print = None
|
||
self.sheet = None
|
||
self.bt_app = BarTender.Application()
|
||
self.bt_app.Visible = False
|
||
self.bt_format = None
|
||
self.printer_list = [printer[2] for printer in win32print.EnumPrinters(win32print.PRINTER_ENUM_LOCAL)][::-1]
|
||
self.printer_current_index = 0
|
||
self.template_source_primary = {}
|
||
self.logger.i('%s%s' % ('Init', ''))
|
||
|
||
def set_label(self, data: str):
|
||
self.label = os.path.abspath(data)
|
||
self.logger.i('%s%s' % ('Open the label template ', self.label))
|
||
for _ in range(2):
|
||
if (self.bt_format is not None) == 1:
|
||
self.bt_format.Close(BarTender.BtSaveOptions.btDoNotSaveChanges)
|
||
self.bt_format = self.bt_app.Formats.Open(self.label, False, '')
|
||
self.template_source_primary = {}
|
||
self.set_print(self.printer_current_index or 0)
|
||
self.set_sheet(self.sheet or 1)
|
||
|
||
def set_print(self, data: int):
|
||
self.printer_current_index = data
|
||
self.print = self.printer_list[self.printer_current_index]
|
||
self.logger.i('%s%s' % ('Printer set to ', self.print))
|
||
if (self.bt_format is not None) == 1:
|
||
self.bt_format.PrintSetup.Printer = self.print
|
||
|
||
def set_sheet(self, data: int):
|
||
self.sheet = int(data)
|
||
self.logger.i('%s%s' % ('Number of copies ', self.sheet))
|
||
if (self.bt_format is not None) == 1:
|
||
self.bt_format.PrintSetup.IdenticalCopiesOfLabel = self.sheet
|
||
|
||
def set_data_source(self, name, value):
|
||
if name not in self.template_source_primary.keys():
|
||
self.template_source_primary[name] = self.bt_format.GetNamedSubStringValue(name)
|
||
self.logger.i('%s%s' % ('Set data source ', '%s=%s' % (name, value)))
|
||
self.bt_format.SetNamedSubStringValue(name, value)
|
||
|
||
def start_printing(self, content: str):
|
||
for k, v in {'A': content}.items():
|
||
self.set_data_source(k, v)
|
||
if (self.bt_format is not None) == 1:
|
||
self.logger.i('%s%s' % ('Printing ', 'normal'))
|
||
self.bt_format.PrintOut(False, False)
|
||
|
||
def start_printing_template(self):
|
||
for k, v in self.template_source_primary.items():
|
||
self.set_data_source(k, v)
|
||
if (self.bt_format is not None) == 1:
|
||
self.logger.i('%s%s' % ('Printing ', 'template'))
|
||
self.bt_format.PrintOut(False, False)
|
||
|
||
def generate_preview(self):
|
||
path = os.path.abspath(
|
||
os.path.join(tempfile.gettempdir(), '%s%s.png' % ('preview_', int(round(time.time() * 1000)))))
|
||
if (self.bt_format is not None) == 1:
|
||
self.logger.i('%s%s' % ('Generate preview to ', path))
|
||
self.bt_format.ExportToFile(path, 'PNG', BarTender.BtColors.btColors24Bit,
|
||
BarTender.BtResolution.btResolutionPrinter,
|
||
BarTender.BtSaveOptions.btDoNotSaveChanges)
|
||
return path
|
||
|
||
def quit(self):
|
||
try:
|
||
self.bt_format is not None and self.bt_format.Close(BarTender.BtSaveOptions.btDoNotSaveChanges)
|
||
self.bt_app is not None and self.bt_app.Quit(BarTender.BtSaveOptions.btDoNotSaveChanges)
|
||
except Exception:
|
||
pass
|
||
self.bt_format = None
|
||
self.bt_app = None
|
||
self.logger.i('%s%s' % ('Quit', ''))
|
||
|
||
|
||
class ScanVerify:
|
||
def __init__(self, logger, rule_file):
|
||
self.logger = logger
|
||
self.checker_list = [['无', '.*']]
|
||
self.checker_list.extend(getJson(os.path.abspath(rule_file), []))
|
||
self.checker_current_index = 0
|
||
|
||
def set_check(self, data: int):
|
||
self.checker_current_index = data
|
||
self.logger.i('%s%s' % ('Checker set to ', self.checker_list[self.checker_current_index][0]))
|
||
|
||
def verify(self, content: str):
|
||
return bool(re.match(self.checker_list[self.checker_current_index][1], content))
|
||
|
||
|
||
class CustomLineEdit(QLineEdit):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet('font-size: 12px; font-family: \'Microsoft YaHei\'; color: #000000;')
|
||
|
||
def contextMenuEvent(self, event):
|
||
menu = QMenu(self)
|
||
action_c = menu.addAction('复制')
|
||
action_p = menu.addAction('粘贴')
|
||
action_c.triggered.connect(self.copy)
|
||
action_p.triggered.connect(self.paste)
|
||
menu.popup(self.mapToGlobal(event.pos()))
|
||
|
||
|
||
class CustomLineEditNoPopup(QLineEdit):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet('font-size: 12px; font-family: \'Microsoft YaHei\'; color: #656565;')
|
||
|
||
def contextMenuEvent(self, event):
|
||
pass
|
||
|
||
|
||
class CustomPushButton(QPushButton):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet('font-size: 12px; font-family: \'Microsoft YaHei\'; color: #0C0C0C;')
|
||
|
||
|
||
class CustomLabel(QLabel):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet(
|
||
'font-size: 12px; font-family: \'Microsoft YaHei\'; color: #0C0C0C; border: 1px solid #0C0C0C;')
|
||
|
||
|
||
class CustomComboBox(QComboBox):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet('font-size: 12px; font-family: \'Microsoft YaHei\'; color: #0C0C0C;')
|
||
|
||
|
||
class CustomCheckBox(QCheckBox):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setStyleSheet('font-size: 12px; font-family: \'Microsoft YaHei\'; color: #0C0C0C;')
|
||
|
||
|
||
class CustomHBoxLayout(QHBoxLayout):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setAlignment(Qt.AlignTop | Qt.AlignLeft)
|
||
|
||
|
||
class CustomVBoxLayout(QVBoxLayout):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.setAlignment(Qt.AlignTop | Qt.AlignLeft)
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
def __init__(self, logger: Logger):
|
||
super().__init__()
|
||
self.app_name = '标签打印'
|
||
self.app_version = ('1.0.5', '20241022', 'zhaoyafan', 'zhaoyafan@foxmail.com', 'https://www.fanscloud.net/')
|
||
self.logger = logger
|
||
self.toast = ToastNotification()
|
||
self.setting = Setting(
|
||
os.path.abspath(os.path.join(tempfile.gettempdir(), '%s.config' % self.md5(__file__)[:16])), {})
|
||
self.bt = BarTenderPrint(logger=self.logger)
|
||
self.sv = ScanVerify(logger=self.logger,
|
||
rule_file=os.path.abspath(os.path.join(os.path.dirname(__file__), 'rules.json')))
|
||
self.ditto = Ditto(
|
||
db_file=os.path.abspath(os.path.join(tempfile.gettempdir(), '%s.db' % self.md5(__file__)[:16])),
|
||
class_name='LabelPrint', limit_time=7776000, limit_rows=100000)
|
||
self.last_opened_template = ['', '']
|
||
self.input_convert_letter = 0
|
||
self.input_scan_prohibited_enter = 0
|
||
self.task_queue = queue.Queue()
|
||
self.task_timer = QTimer(self)
|
||
self.task_timer.timeout.connect(self.task_execute)
|
||
self.task_timer.start(100)
|
||
self.setWindowTitle(self.app_name)
|
||
self.setGeometry(0, 0, 600, 400)
|
||
self.statusBar()
|
||
self.menubar = self.menuBar()
|
||
filemenu = self.menubar.addMenu('设置')
|
||
self.blockRepeatAction = QAction('禁止重复打印', self, checkable=True)
|
||
self.clearRecordAction = QAction('清空打印记录', self)
|
||
filemenu.addActions(
|
||
[
|
||
self.blockRepeatAction,
|
||
self.clearRecordAction
|
||
]
|
||
)
|
||
moremenu = self.menubar.addMenu('更多')
|
||
self.designStateAction = QAction('编辑模式', self, checkable=True)
|
||
self.createShortAction = QAction('发送快捷方式到桌面', self)
|
||
self.aboutWindowAction = QAction('关于', self)
|
||
self.blockRepeatAction.triggered.connect(self.blockRepeatActionFunction)
|
||
self.clearRecordAction.triggered.connect(self.clearRecordActionFunction)
|
||
self.designStateAction.triggered.connect(self.designStateActionFunction)
|
||
self.createShortAction.triggered.connect(self.createShortActionFunction)
|
||
self.aboutWindowAction.triggered.connect(self.aboutWindowActionFunction)
|
||
moremenu.addActions(
|
||
[
|
||
self.designStateAction,
|
||
self.createShortAction,
|
||
self.aboutWindowAction
|
||
]
|
||
)
|
||
# 定义布局
|
||
m_layout_0 = CustomVBoxLayout()
|
||
l_layout_0 = CustomVBoxLayout()
|
||
l_layout_1 = CustomHBoxLayout()
|
||
l_layout_2 = CustomVBoxLayout()
|
||
r_layout_0 = CustomVBoxLayout()
|
||
r_layout_1 = CustomHBoxLayout()
|
||
r_layout_2 = CustomHBoxLayout()
|
||
r_layout_3 = CustomVBoxLayout()
|
||
l_layout_0.addLayout(l_layout_1)
|
||
l_layout_0.addLayout(l_layout_2)
|
||
r_layout_0.addLayout(r_layout_1)
|
||
r_layout_0.addLayout(r_layout_2)
|
||
r_layout_0.addLayout(r_layout_3)
|
||
m_layout_0.addLayout(l_layout_0)
|
||
m_layout_0.addLayout(r_layout_0)
|
||
|
||
# Select file
|
||
self.butSTP = CustomPushButton('选择文件')
|
||
self.butSTP.setFixedSize(82, 27)
|
||
self.butSTP.clicked.connect(self.on_open_template_file)
|
||
l_layout_1.addWidget(self.butSTP)
|
||
# Template name
|
||
self.edtNTP = CustomLineEditNoPopup('')
|
||
self.edtNTP.setReadOnly(True)
|
||
self.edtNTP.setFixedSize(280, 27)
|
||
l_layout_1.addWidget(self.edtNTP)
|
||
# Test printing
|
||
self.butTPT = CustomPushButton('测试打印')
|
||
self.butTPT.setFixedSize(82, 27)
|
||
self.butTPT.clicked.connect(self.on_test_print)
|
||
l_layout_1.addWidget(self.butTPT)
|
||
# Print preview
|
||
self.labPRV = CustomLabel()
|
||
self.labPRV.setAlignment(Qt.AlignCenter)
|
||
self.labPRV.setFixedSize(455, 256)
|
||
l_layout_2.addWidget(self.labPRV)
|
||
# Printer
|
||
self.labNPT = CustomLabel('打印机选择')
|
||
self.labNPT.setFixedSize(82, 27)
|
||
r_layout_1.addWidget(self.labNPT)
|
||
# Select printer
|
||
self.slcSPT = CustomComboBox()
|
||
self.slcSPT.addItems(self.bt.printer_list)
|
||
self.slcSPT.setFixedSize(213, 27)
|
||
self.slcSPT.currentIndexChanged.connect(self.on_printer_changed)
|
||
r_layout_1.addWidget(self.slcSPT)
|
||
# Number of copies
|
||
self.labPQT = CustomLabel('打印份数')
|
||
self.labPQT.setFixedSize(82, 27)
|
||
r_layout_1.addWidget(self.labPQT)
|
||
self.edtNQT = CustomLineEditNoPopup('1')
|
||
self.edtNQT.setFixedSize(60, 27)
|
||
self.edtNQT.textChanged.connect(self.on_copies_editing_changed)
|
||
validator = QIntValidator(1, 999)
|
||
self.edtNQT.setValidator(validator)
|
||
r_layout_1.addWidget(self.edtNQT)
|
||
# Input verification
|
||
self.labNVR = CustomLabel('扫描验证')
|
||
self.labNVR.setFixedSize(82, 27)
|
||
r_layout_2.addWidget(self.labNVR)
|
||
# Select input verification
|
||
self.slcSVR = CustomComboBox()
|
||
self.slcSVR.addItems([data[0] for data in self.sv.checker_list])
|
||
self.slcSVR.setFixedSize(213, 27)
|
||
self.slcSVR.currentIndexChanged.connect(self.on_checker_changed)
|
||
r_layout_2.addWidget(self.slcSVR)
|
||
# Convert letter
|
||
self.chkCVL = CustomCheckBox('强制大写')
|
||
self.chkCVL.setFixedSize(72, 27)
|
||
self.chkCVL.stateChanged.connect(self.on_convert_letter_changed)
|
||
r_layout_2.addWidget(self.chkCVL)
|
||
# Disable enter
|
||
self.chkDET = CustomCheckBox('禁用回车')
|
||
self.chkDET.setFixedSize(72, 27)
|
||
self.chkDET.stateChanged.connect(self.on_prohibit_enter_changed)
|
||
r_layout_2.addWidget(self.chkDET)
|
||
# Scan
|
||
self.edtSCN = CustomLineEdit('')
|
||
self.edtSCN.setStyleSheet(
|
||
'QLineEdit {font-size: 28px; font-family: \'Microsoft YaHei\'; color: #000000; background-color: #FFFFCC;}')
|
||
self.edtSCN.setFixedSize(455, 45)
|
||
r_layout_3.addWidget(self.edtSCN)
|
||
# Print button
|
||
self.butSPT = CustomPushButton('打印')
|
||
self.butSPT.setFixedSize(455, 45)
|
||
self.butSPT.clicked.connect(self.on_start_printing)
|
||
self.edtSCN.returnPressed.connect(self.on_print_scan_enter)
|
||
r_layout_3.addWidget(self.butSPT)
|
||
|
||
# 显示界面
|
||
central_widget = QWidget()
|
||
central_widget.setLayout(m_layout_0)
|
||
self.setCentralWidget(central_widget)
|
||
self.setFixedSize(self.minimumSizeHint())
|
||
screen_rect = QApplication.desktop().availableGeometry()
|
||
window_rect = self.geometry()
|
||
self.move(int((screen_rect.width() - window_rect.width()) * 0.5),
|
||
int((screen_rect.height() - window_rect.height()) * 0.5))
|
||
self.edtSCN.setFocus()
|
||
self.show()
|
||
self.load_setting()
|
||
|
||
def closeEvent(self, event):
|
||
self.bt.quit()
|
||
killThread(flask_thread)
|
||
event.accept()
|
||
|
||
def task_push(self, func, args):
|
||
self.task_queue.put([func, args])
|
||
|
||
def task_execute(self):
|
||
while True:
|
||
if not self.task_queue.empty():
|
||
task = self.task_queue.get()
|
||
try:
|
||
task[0](*task[1])
|
||
except:
|
||
pass
|
||
self.task_queue.task_done()
|
||
else:
|
||
break
|
||
|
||
def load_setting(self):
|
||
# load blockRepeat
|
||
self.blockRepeatAction.setChecked(bool(self.setting['blockRepeat']))
|
||
# load printer
|
||
self.slcSPT.setCurrentIndex(
|
||
next((index for index, item in enumerate(self.bt.printer_list) if item == self.setting['printer']), 0))
|
||
# load checker
|
||
self.slcSVR.setCurrentIndex(self.setting['checker'] or 0)
|
||
# load printCopies
|
||
self.edtNQT.setText(str(self.setting['printCopies'] or 1))
|
||
# load template
|
||
tempfile_last = self.setting['template']
|
||
if (tempfile_last and os.path.exists(tempfile_last)) == 1:
|
||
self.load_template(tempfile_last)
|
||
else:
|
||
self.setting['template'] = ''
|
||
|
||
def blockRepeatActionFunction(self, checked):
|
||
self.setting['blockRepeat'] = bool(checked)
|
||
|
||
def clearRecordActionFunction(self):
|
||
count = self.ditto.clear()
|
||
if (count > 0) == 1:
|
||
self.toast.show_toast('已成功清空%s条打印记录' % (count,))
|
||
|
||
def designStateActionFunction(self, checked):
|
||
if (not self.bt.bt_app) == 1:
|
||
return None
|
||
if (checked is True) == 1:
|
||
t1 = '1.在编辑模式中您可以动态对模板进行修改；'
|
||
t2 = '2.保存模板将会覆盖原文件永久生效；'
|
||
t3 = '3.修改完成后手动关闭编辑模式；'
|
||
QMessageBox.information(self, '提示', '%s\n%s\n%s' % (t1, t2, t3))
|
||
self.bt.bt_app.Visible = bool(1)
|
||
else:
|
||
self.bt.bt_app.Visible = bool(0)
|
||
filename = self.last_opened_template[0]
|
||
if (filename == '' or not os.path.exists(filename)) == 1:
|
||
return None
|
||
last_md5 = self.last_opened_template[1]
|
||
temp_md5 = self.md5(open(self.last_opened_template[0], 'rb'))
|
||
QMessageBox.information(self, '提示', '模板文件已被修改：\n\n%s' % filename) if (last_md5 != temp_md5) == 1 else None
|
||
self.last_opened_template[1] = temp_md5
|
||
|
||
def createShortActionFunction(self):
|
||
target = '%s.exe' % (os.path.splitext(__file__)[0],)
|
||
shortcut_path = self.path_expandvars('%%USERPROFILE%%\\Desktop\\%s.lnk' % (self.app_name,))
|
||
if (os.path.exists(target)) == 1:
|
||
self.create_shortcut(target, shortcut_path, 1)
|
||
QMessageBox.information(self, '提示', '%s' % ('已成功发送快捷方式到桌面',))
|
||
|
||
def aboutWindowActionFunction(self):
|
||
_c = self.app_version
|
||
if (not _c) == 0:
|
||
QMessageBox.information(self, '关于', "" 'version: %s, build: %s, author: %s, email: %s, site: %s' % (
|
||
_c[0], _c[1], _c[2], _c[3], '<a href=\'%s\'>%s</a>' % (_c[4], _c[4])))
|
||
|
||
@staticmethod
|
||
def md5(input_data):
|
||
if isinstance(input_data, bytes):
|
||
return hashlib.md5(input_data).hexdigest()
|
||
if isinstance(input_data, str):
|
||
return hashlib.md5(bytes(input_data, encoding='utf-8')).hexdigest()
|
||
md5_object = hashlib.md5()
|
||
while True:
|
||
data = input_data.read(io.DEFAULT_BUFFER_SIZE)
|
||
if data:
|
||
md5_object.update(data)
|
||
else:
|
||
return md5_object.hexdigest()
|
||
|
||
@staticmethod
|
||
def path_expandvars(path):
|
||
resolve = os.path.expandvars(path)
|
||
if (re.search('%[A-Za-z0-9_]+%', resolve)) is not None:
|
||
raise Exception('Variable analysis failed')
|
||
return resolve
|
||
|
||
@staticmethod
|
||
def create_shortcut(target, shortcut_path, run_as_admin):
|
||
shell = win32com.client.Dispatch('WScript.Shell')
|
||
try:
|
||
os.remove(shortcut_path)
|
||
except FileNotFoundError:
|
||
pass
|
||
shortcut = shell.CreateShortCut(shortcut_path)
|
||
shortcut.TargetPath = target
|
||
if shortcut_path.endswith('.lnk'):
|
||
shortcut.Arguments = ''
|
||
shortcut.WorkingDirectory = os.path.dirname(target)
|
||
if target.startswith('\\\\'):
|
||
shortcut.WorkingDirectory = ''
|
||
shortcut.save()
|
||
if shortcut_path.endswith('.lnk') and run_as_admin:
|
||
with open(shortcut_path, 'r+b') as f:
|
||
if os.path.isfile(shortcut_path):
|
||
f.seek(21, 0)
|
||
f.write(b'\x22\x00')
|
||
|
||
def set_preview(self, preview_image: str):
|
||
if (preview_image and os.path.exists(preview_image)) == 1:
|
||
pixmap = QPixmap(preview_image)
|
||
pixmap = pixmap.scaled(self.labPRV.size(), aspectRatioMode=Qt.KeepAspectRatio,
|
||
transformMode=Qt.SmoothTransformation)
|
||
self.labPRV.setPixmap(pixmap)
|
||
if (tempfile.gettempdir() in preview_image) == 1:
|
||
rm(preview_image)
|
||
|
||
def set_template_name(self, name):
|
||
if (isinstance(name, str)) == 1:
|
||
self.edtNTP.setText(name)
|
||
|
||
def load_template(self, file):
|
||
try:
|
||
if (file != '' and os.path.exists(file)) == 1:
|
||
self.bt.set_label(file)
|
||
self.set_template_name(os.path.splitext(os.path.basename(file))[0])
|
||
self.set_preview(self.bt.generate_preview())
|
||
self.last_opened_template[0] = file
|
||
self.last_opened_template[1] = self.md5(open(file, 'rb'))
|
||
self.toast.show_toast('模板加载成功')
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
def on_open_template_file(self):
|
||
filename, _ = QFileDialog.getOpenFileName(self, '打开文件', os.path.dirname(self.last_opened_template[0]),
|
||
'BarTender 文档 (*.btw)')
|
||
if (filename and os.path.exists(filename)) == 1:
|
||
self.load_template(filename)
|
||
self.setting['template'] = filename
|
||
|
||
def on_test_print(self):
|
||
try:
|
||
if (self.bt.bt_format is not None) == 1:
|
||
self.bt.start_printing_template()
|
||
self.set_preview(self.bt.generate_preview())
|
||
return None
|
||
self.toast.show_toast('请先加载模板文件')
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
def on_printer_changed(self, index):
|
||
try:
|
||
self.bt.set_print(index)
|
||
self.set_preview(self.bt.generate_preview())
|
||
self.setting['printer'] = str(self.bt.print)
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
def on_checker_changed(self, index):
|
||
try:
|
||
self.sv.set_check(index)
|
||
self.setting['checker'] = index
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
def on_copies_editing_changed(self, text):
|
||
try:
|
||
if (text == '' or int(text) == 0) == 1:
|
||
text = '1'
|
||
self.edtNQT.setText(text)
|
||
else:
|
||
copies = int(text)
|
||
self.bt.set_sheet(copies)
|
||
self.setting['printCopies'] = copies
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
def on_convert_letter_changed(self, checked):
|
||
if (checked == 2) == 1:
|
||
self.input_convert_letter = 1
|
||
self.toast.show_toast('强制大写开启')
|
||
else:
|
||
self.input_convert_letter = 0
|
||
self.toast.show_toast('强制大写关闭')
|
||
|
||
def on_prohibit_enter_changed(self, checked):
|
||
if (checked == 2) == 1:
|
||
self.input_scan_prohibited_enter = 1
|
||
self.toast.show_toast('禁用回车开启')
|
||
else:
|
||
self.input_scan_prohibited_enter = 0
|
||
self.toast.show_toast('禁用回车关闭')
|
||
|
||
def on_print_scan_enter(self):
|
||
self.input_scan_prohibited_enter or self.on_start_printing()
|
||
|
||
def on_start_printing(self, content=None):
|
||
edit = self.edtSCN
|
||
if (self.input_convert_letter == 1) == 1:
|
||
text = edit.text().strip().upper()
|
||
else:
|
||
text = edit.text().strip()
|
||
if (not content) == 0:
|
||
text = str(content).strip()
|
||
edit.setText(text)
|
||
if (self.bt.bt_format is None) == 1:
|
||
self.toast.show_toast('未加载模板文件')
|
||
edit.selectAll()
|
||
return None
|
||
if (text == '') == 1:
|
||
self.toast.show_toast('输入的内容为空')
|
||
edit.selectAll()
|
||
return None
|
||
if (not self.sv.verify(text)) == 1:
|
||
self.toast.show_toast('输入的内容不符合验证规则')
|
||
edit.selectAll()
|
||
return None
|
||
if (not self.ditto.addit(text) and self.setting['blockRepeat']) == 1:
|
||
QMessageBox.critical(self, '错误', '重复打印：\n\n%s' % (text,))
|
||
edit.selectAll()
|
||
return None
|
||
try:
|
||
self.bt.start_printing(text)
|
||
self.set_preview(self.bt.generate_preview())
|
||
edit.clear()
|
||
except pywintypes.com_error as e:
|
||
t = str(e.args[1])
|
||
m = str(e.args[2][2] if isinstance(e.args[2], tuple) else e.args[2])
|
||
if (len(t) > 5) == 1:
|
||
m = '%s\n\n%s' % (t, m)
|
||
t = '错误'
|
||
QMessageBox.critical(self, t, m)
|
||
except AttributeError:
|
||
QMessageBox.critical(self, '错误', '调用服务失败')
|
||
except Exception as e:
|
||
QMessageBox.critical(self, '错误', '%s' % e)
|
||
|
||
|
||
class ToastNotification(QDialog):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowFlags(Qt.FramelessWindowHint | Qt.WindowStaysOnTopHint | Qt.WindowType.Tool)
|
||
layout = QVBoxLayout()
|
||
layout.setContentsMargins(5, 705, 5, 5)
|
||
|
||
self.label = QLabel("")
|
||
self.label.setStyleSheet("font-family: 'Microsoft YaHei'; font-size: 24px; color: #ffffff;")
|
||
layout.addWidget(self.label)
|
||
self.setLayout(layout)
|
||
self.setAttribute(Qt.WA_TranslucentBackground)
|
||
self.setStyleSheet("background-color: rgba(65, 65, 65, 180); border-radius: 5px; padding: 10px 16px 10px 16px")
|
||
|
||
self.adjustSize()
|
||
|
||
self.timer = QTimer(self)
|
||
self.timer.setInterval(2000)
|
||
self.timer.timeout.connect(self.close)
|
||
|
||
self.fade_animation = QPropertyAnimation(self, b"windowOpacity")
|
||
self.fade_animation.setDuration(500)
|
||
|
||
def show_toast(self, message):
|
||
self.close()
|
||
self.label.setText(message)
|
||
self.timer.isActive() and self.timer.stop()
|
||
self.timer.start()
|
||
self.adjustSize()
|
||
self.fade_animation.setStartValue(0.0)
|
||
self.fade_animation.setEndValue(1.0)
|
||
self.show()
|
||
self.fade_animation.start()
|
||
|
||
def closeEvent(self, event):
|
||
if self.timer.isActive():
|
||
self.timer.stop()
|
||
super().closeEvent(event)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
if (os.path.basename(__file__).lower().endswith('.int')) == 1:
|
||
QCoreApplication.addLibraryPath(
|
||
os.path.abspath(os.path.join(os.path.dirname(__file__), 'site-packages/PyQt5/Qt5/plugins')))
|
||
f_lock = open(file=os.path.abspath(os.path.join(tempfile.gettempdir(), '%s.lock' % hashlib.md5(
|
||
bytes(__file__, encoding='utf-8')).hexdigest()[:16])), mode='w', encoding='utf-8')
|
||
if (not flock(f_lock, LOCK_EX | LOCK_NB)) == 1:
|
||
app = QApplication(sys.argv)
|
||
msg = QMessageBox()
|
||
msg.setIcon(QMessageBox.Warning)
|
||
msg.setText('程序已经在运行中')
|
||
msg.setWindowTitle('提示')
|
||
msg.setStandardButtons(QMessageBox.Ok)
|
||
msg.exec_()
|
||
app.exit(1)
|
||
sys.exit(1)
|
||
try:
|
||
win32com.client.Dispatch('BarTender.Application')
|
||
except pywintypes.com_error:
|
||
app = QApplication(sys.argv)
|
||
msg = QMessageBox()
|
||
msg.setIcon(QMessageBox.Critical)
|
||
msg.setText('初始化失败，请检查BarTender是否已安装。')
|
||
msg.setWindowTitle('错误')
|
||
msg.setStandardButtons(QMessageBox.Ok)
|
||
msg.exec_()
|
||
app.exit(1)
|
||
sys.exit(1)
|
||
app = QApplication(sys.argv)
|
||
window_main = MainWindow(logger=Logger(name='main', fh=None, ch=LoggerConsHandler()))
|
||
|
||
|
||
@flask_app.route('/print', methods=['POST'])
|
||
def api_print():
|
||
content = request.json['data']
|
||
if (isinstance(content, str) is True) == 1:
|
||
threading.Thread(target=window_main.task_push, args=[window_main.on_start_printing, [content]]).start()
|
||
return 'Printing task sent'
|
||
else:
|
||
return 'Error'
|
||
|
||
|
||
flask_thread.start()
|
||
sys.exit(app.exec_())
|