133 lines
4.9 KiB
Python
133 lines
4.9 KiB
Python
import os
|
|
import pystray
|
|
import threading
|
|
from PIL import Image
|
|
from pystray import MenuItem
|
|
import time
|
|
# import requests
|
|
from datetime import datetime
|
|
import json
|
|
from configparser import ConfigParser
|
|
import socket
|
|
import glob
|
|
|
|
class SnPost:
|
|
|
|
def __init__(self, server, port, format_str) -> None:
|
|
self.is_one_instance = False
|
|
self.is_working = False
|
|
self.icon = None # 下标
|
|
self.server = server
|
|
self.port = port
|
|
self.format_str = format_str
|
|
self.check_one_instance()
|
|
|
|
|
|
def check_one_instance(self):
|
|
try:
|
|
global s
|
|
s = socket.socket()
|
|
host = socket.gethostname()
|
|
s.bind((host, 20309))
|
|
self.is_one_instance = True
|
|
except Exception:
|
|
self.is_one_instance = False
|
|
|
|
def close(self):
|
|
self.is_working = False
|
|
self.icon.stop()
|
|
|
|
def format_data(self, data):
|
|
|
|
format_list = format_str.split(',')
|
|
data = data.replace('"', '').replace('\r', '').replace('\n', '')
|
|
data_list = data.split(',')
|
|
post_data = {'TestVaule':[]}
|
|
ALL_ELEMENT = ["loss", "SiO2", "Al2O3", '"Fe2O3"', "CaO", "MgO", "MnO", "TiO2", "SO3", "P2O5", "Cl-", "K2O", "Na2O", "R2O", "f-CaO", "KH", "KH-", "SM", "IM", "LSF", "C2S", "C3S", "C3A", "C4AF", "SUM"]
|
|
HAS_ELEMENT = []
|
|
try:
|
|
for index, value in enumerate(format_list):
|
|
if value == 'SampleNo':
|
|
post_data['SampleNo'] = data_list[index]
|
|
elif value == 'TestDate':
|
|
dt = datetime.strptime(data_list[index], '%m/%d/%Y %I:%M:%S %p')
|
|
post_data['TestDate'] = dt.strftime('%Y-%m-%d %H:%M')
|
|
elif value == 'TestPro':
|
|
post_data['TestVaule'].append({'检测项目':data_list[index]})
|
|
else:
|
|
HAS_ELEMENT.append(value)
|
|
value_p = data_list[index]
|
|
if value_p.startswith('.'):
|
|
value_p = '0' + value_p
|
|
post_data['TestVaule'].append({value:value_p})
|
|
difference = set(ALL_ELEMENT).difference(set(HAS_ELEMENT))
|
|
for i in difference:
|
|
post_data['TestVaule'].append({i:None})
|
|
return json.dumps(post_data, ensure_ascii=False)
|
|
except:
|
|
import traceback
|
|
print(traceback.format_exc())
|
|
self.icon.notify(message='数据格式解析失败', title="ctc数据推送")
|
|
return None
|
|
|
|
|
|
def run(self):
|
|
if not self.is_one_instance:
|
|
return
|
|
|
|
menu = (MenuItem('退出', self.close), )
|
|
image = Image.open("favicon.ico")
|
|
title = "ctc数据推送"
|
|
self.icon = pystray.Icon(title, image, "ctc数据推送", menu)
|
|
threading.Thread(target=self.icon.run, daemon=True).start()
|
|
|
|
server_addr = (server, int(port))
|
|
self.is_working = True
|
|
filename_pattern = "*.dat"
|
|
# 获取当前工作目录
|
|
current_dir = os.getcwd()
|
|
while self.is_working:
|
|
files = glob.glob(os.path.join(current_dir, filename_pattern))
|
|
print(files)
|
|
for file in files:
|
|
# file_path = os.path.join(currentPath, file)
|
|
# if file_path.endswith('.txt'):
|
|
# with open(file_path, 'r', encoding="utf-8") as f:
|
|
# post_data = json.load(f)
|
|
# for i in post_data['TestVaule']:
|
|
# for k in i:
|
|
# if i[k] == 'null':
|
|
# i[k] = None
|
|
with open(file, 'r') as f:
|
|
post_data = self.format_data(f.read())
|
|
if post_data:
|
|
try:
|
|
# requests.post(url=serverUrl, json=post_data)
|
|
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
tcp_socket.settimeout(3)
|
|
tcp_socket.connect(server_addr)
|
|
tcp_socket.sendall(post_data.encode('utf-8'))
|
|
os.rename(file, file+'_bak')
|
|
except Exception as e:
|
|
err = '服务器错误:{}'.format(e.__str__()[0:80])
|
|
self.icon.notify(message=err, title="ctc数据推送")
|
|
finally:
|
|
tcp_socket.close()
|
|
else:
|
|
return
|
|
time.sleep(3)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
conf = ConfigParser() # 需要实例化一个ConfigParser对象
|
|
conf.read('conf.ini')
|
|
# serverUrl = conf['ctc']['serverUrl']
|
|
server = conf['ctc']['server']
|
|
port = conf['ctc']['port']
|
|
format_str = conf['ctc']['format']
|
|
snpost = SnPost(server, port, format_str)
|
|
snpost.run()
|
|
|
|
# 打包命令
|
|
# pyinstaller -F -w -i favicon.ico ctc数据推送.py
|