612 lines
18 KiB
Python
Executable File
612 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# coding:utf-8
|
||
import socketserver
|
||
import subprocess
|
||
import pickle
|
||
from collections import defaultdict
|
||
import datetime
|
||
import time
|
||
import requests
|
||
import socket
|
||
# import json
|
||
from threading import Timer
|
||
from loguru import logger
|
||
import os
|
||
|
||
|
||
|
||
def gettm():
|
||
"""
|
||
返回时戳和日期
|
||
"""
|
||
t= datetime.datetime.now()
|
||
d = str(t)
|
||
return d,d[:10]
|
||
|
||
def save2file(data,isR=1):
|
||
"""
|
||
存储数据至文件
|
||
"""
|
||
t,d = gettm()
|
||
|
||
p = '/'.join(['./trlog',d])
|
||
if not os.path.exists(p):
|
||
os.makedirs(p)
|
||
|
||
l=f"{'[R]' if isR else '[T]'} - {t} --- {data}\n"
|
||
f=open('/'.join(['./trlog',d,'trlog.txt']),'a')
|
||
f.write(l)
|
||
f.close
|
||
|
||
|
||
|
||
SERVER = "http://www.pushplus.plus/send"
|
||
TOKEN = "ff328cba923a4225bc4acd0086a9014c"
|
||
# SERVER = "http://push.nmhd.eu.org:12722/push/murmur"
|
||
# TOKEN = "tt"
|
||
|
||
def pp2wx(title,content,description):
|
||
"""通过pushplus发送至微信，有频次限制.
|
||
title:标题
|
||
content:内容"""
|
||
|
||
ppmsg = {"token":TOKEN,
|
||
"template":"json",
|
||
"channel":"wechat",
|
||
"description": description,
|
||
}
|
||
ppmsg["title"] = title
|
||
ppmsg["content"] = content
|
||
# ppmsg["description"]= content
|
||
requests.post(SERVER, json=ppmsg)
|
||
|
||
msgcache=[]
|
||
pushisbusy=0
|
||
def s2wx(title="",description="", content=""):
|
||
"""通过自建msgpusher发送至微信
|
||
当前title值无效
|
||
"""
|
||
# logger.debug(content)
|
||
# return
|
||
# 受API限制，频繁发送会失败
|
||
# 拟合并发送，满3条已发送，超时时间10s
|
||
if len(content):
|
||
t=f"""
|
||
**时戳：**
|
||
{datetime.datetime.now()}
|
||
**信息：**
|
||
{content}"""
|
||
msgcache.append(t)
|
||
|
||
global pushisbusy
|
||
if pushisbusy:
|
||
Timer(10,s2wx).start()
|
||
return
|
||
pushisbusy=1
|
||
le = len(msgcache)
|
||
def send():
|
||
c= " \n".join(msgcache[:3])
|
||
try:
|
||
res = requests.post("http://8.137.110.19:4014/push/murmur", json={
|
||
"title": title,
|
||
"description": description,
|
||
"content": c,
|
||
"channel":"all-TT",
|
||
"token": "gjvL2gMZqj326isgafmz"})
|
||
logger.debug("msgpusher {}",res)
|
||
except:
|
||
logger.error("msgpusher error");
|
||
|
||
try:
|
||
res = requests.post("https://push.020824.xyz/push/murmur", json={
|
||
"title": title,
|
||
"description": description,
|
||
"content": c,
|
||
"channel":"all-TT",
|
||
"token": "gjvL2gMZqj326isgafmz"})
|
||
logger.debug("msgpusher {}",res)
|
||
except:
|
||
logger.error("msgpusher error");
|
||
|
||
|
||
|
||
|
||
|
||
# try:
|
||
# try:
|
||
# res = res.json()
|
||
# except:
|
||
# s="http://push.nmhd.eu.org:12722/push/murmur"
|
||
# # s="http://47.108.213.155:11722/push/murmur"
|
||
# res = requests.post(s, json={
|
||
# "title": title,
|
||
# "description": description,
|
||
# "content": c,
|
||
# "token": "tt"})
|
||
# print("nmhd",s)
|
||
# else:
|
||
# if res["success"]:
|
||
# return None
|
||
# else:
|
||
# return res["message"]
|
||
|
||
if le >2:
|
||
send()
|
||
msgcache.pop(0)
|
||
msgcache.pop(0)
|
||
msgcache.pop(0)
|
||
# Timer(10,s2wx).start()
|
||
elif (not len(content)) and le:
|
||
# 等效超时
|
||
send()
|
||
for i in range(le):
|
||
msgcache.pop(0)
|
||
# Timer(10,s2wx).start()
|
||
pushisbusy=0
|
||
Timer(10,s2wx).start()
|
||
# GET 方式
|
||
# res = requests.get(f"{SERVER}/push/{USERNAME}?title={title}"
|
||
# f"&description={description}&content={content}&token={TOKEN}")
|
||
|
||
# POST 方式
|
||
# res = requests.post("http://push.nmhd.eu.org:12722/push/murmur", json={
|
||
# "title": title,
|
||
# "description": description,
|
||
# "content": f"""
|
||
# *时戳：*
|
||
# _{datetime.datetime.now()}_
|
||
# *信息：*
|
||
# {content}""",
|
||
# "token": "tt"
|
||
# })
|
||
# try:
|
||
# res = res.json()
|
||
# except:
|
||
# return
|
||
# else:
|
||
# if res["success"]:
|
||
# return None
|
||
# else:
|
||
# return res["message"]
|
||
Timer(10,s2wx).start()
|
||
|
||
cache = defaultdict(list)
|
||
offs = 0 # 偏移量
|
||
|
||
# 保存终端信息，ccid=地址
|
||
p = {}
|
||
# 地址 ==ccid
|
||
pp = {}
|
||
|
||
|
||
def add_timestamp():
|
||
logger.info("-------------------------↓ {} ↓ -------------------------",datetime.datetime.now() )
|
||
|
||
|
||
def save_cache(dic):
|
||
"""缓存未发送成功的消息
|
||
仅在目标终端未上线或发送失败时缓存
|
||
发送失败的一般原因是终端掉线但未及时更新在线列表"""
|
||
|
||
# f=open("./cache.txt",mode='w+')
|
||
# f.write(str(dic))
|
||
# f.close
|
||
pickle.dump(dic, open("./cache.txt", "wb"))
|
||
|
||
|
||
def load_cache():
|
||
"""程序启动时加载缓存"""
|
||
try:
|
||
# f=open("./cache.txt")
|
||
# c = eval(f.read())
|
||
# f.close()
|
||
c = pickle.load(open("./cache.txt", "rb"))
|
||
return c
|
||
except:
|
||
return None
|
||
|
||
|
||
def update_pairs(addr, ccid, clear=0,cnt=[0]):
|
||
"""更新终端库
|
||
addr:绑定tcp地址
|
||
ccid:终端号
|
||
clear:清空掉线终端
|
||
cnt:心跳包数量"""
|
||
|
||
if not clear:
|
||
# 添加或更新客户端
|
||
pp[addr] = ccid
|
||
p.clear
|
||
|
||
if ccid not in p:
|
||
# 新终端上线，推送在线消息
|
||
cnt[0] = 0
|
||
t = f"""*信息:*
|
||
天通终端上线。
|
||
CCID:{ccid.decode()}
|
||
地址:{str(addr.getpeername())}"""
|
||
logger.info(t)
|
||
s2wx("","天通终端上线",t)
|
||
# p[ccid] = addr
|
||
|
||
else:
|
||
del p[ccid]
|
||
del pp[addr]
|
||
|
||
# 打印在线终端
|
||
# for i in p:
|
||
# print(i, "<--->", p[i])
|
||
od= []
|
||
for i in pp:
|
||
p[pp[i]] = i
|
||
# od.append(f"{pp[i].decode()}<--->{i.getpeername()}")
|
||
# 可能客户端未断开，导致多个连接指向相同的ccid，故需更新后再遍历打印
|
||
for i in p:
|
||
od.append(f"{i.decode()}<--->{p[i].getpeername()}")
|
||
# print('\r\n'.join(zd))
|
||
clents = ' \n'+'\r\n'.join(od) if len(od) else ""
|
||
t=f"在线终端已更新，数量{len(od)}。{clents}"
|
||
logger.info(t)
|
||
# cnt[0]是一个列表，它的默认值只会在函数定义时初始化一次，因此它的值在多次调用时保持不变
|
||
# cnt[0]为10的倍数
|
||
if cnt[0]%10 == 0:
|
||
s2wx("","",t)
|
||
cnt[0] += 1
|
||
|
||
def get_addr(ccid):
|
||
"""获取ccid对应的地址"""
|
||
|
||
try:
|
||
# 如果有直接返回
|
||
return p[ccid]
|
||
except Exception:
|
||
# 没有就返回None
|
||
return None
|
||
|
||
|
||
def get_ccid(addr):
|
||
try:
|
||
# 如果有直接返回
|
||
return pp[addr]
|
||
except Exception:
|
||
# 没有就返回None
|
||
logger.warning("终端未上报心跳注册ccid。")
|
||
return None
|
||
|
||
|
||
"""
|
||
+--------+--------------+----------+----------+--------------+------------------------+--------+
|
||
| ff | 长度 | 地址信息 | 信令 | 内容 | 校验位 |
|
||
+--------+--------------+----------+----------+--------------+------------------------+--------+
|
||
| | 2Bytes | 4Bytes | 心跳 | 0x01 | 时间戳32bit（单位ms） | 1Bytes |
|
||
+ $TTMS +--------------+----------+----------+--------------+------------------------+--------+
|
||
| | 2Bytes | 4Bytes | 传输数据 | 0xaa | 载荷 | 1Bytes |
|
||
+--------+--------------+----------+----------+--------------+------------------------+--------+
|
||
"""
|
||
|
||
# 回复ACK给终端，使用命令0xAA，载荷为四字节0xFFFFFFFF
|
||
def ack2client(addr,data):
|
||
msg = bytearray(data[:])
|
||
msg[11 + offs] = 0xAA
|
||
# 覆写32bit时戳为FF
|
||
msg[12 + offs:16 + offs] = int(0xFFFFFFFF).to_bytes(4,byteorder="big")
|
||
# 计算校验和
|
||
crc = 0
|
||
for i in msg[:-1]:
|
||
crc = crc ^ i
|
||
msg[-1] = crc
|
||
addr.send(msg)
|
||
# save2file(f"[SERVER --> {get_ccid(addr).decode() if get_ccid(addr) else 'None'}][{len(msg)}] {msg.hex(' ')}",0)
|
||
# logger.debug("服务器回复ACK。")
|
||
|
||
|
||
def tt_hh(addr, data):
|
||
"""处理心跳"""
|
||
|
||
# 更新pairs
|
||
ccid = data[7 + offs : 11 + offs]
|
||
update_pairs(addr, ccid)
|
||
# 组帧再返回
|
||
msg = bytearray(data[:])
|
||
msg[:5] = b"$TTMX"
|
||
val = "%04x" % int(time.time()) #取Unix时间戳，以s为单位
|
||
tmp=[]
|
||
for i in range(0,len(val),2):
|
||
tmp.append(int(val[i:i+2],16))
|
||
msg[12:-1] = tmp
|
||
# 计算校验和
|
||
crc = 0
|
||
for i in msg[:-1]:
|
||
crc = crc ^ i
|
||
msg[-1] = crc
|
||
add_timestamp()
|
||
addr.send(msg)
|
||
save2file(f"[SERVER --> {get_ccid(addr).decode() if get_ccid(addr) else 'None'}][{len(msg)}] {msg.hex(' ')}",0)
|
||
logger.debug("服务器回复心跳包。")
|
||
ack2client(addr,msg)
|
||
|
||
# 从缓存中匹配到数据才重发
|
||
# 仅有新终端上线时才重发
|
||
if cache and ccid in cache:
|
||
msgs = cache[ccid][:] # 浅拷贝
|
||
sccid = msgs[0][7 + offs : 11 + offs]
|
||
print("有", len(msgs), "包缓存数据待发。", sccid, "-->", ccid)
|
||
s2wx("","",f"有{len(msgs)}包缓存数据待发。 \n{sccid} --> {ccid}")
|
||
for i in range(len(msgs)):
|
||
try:
|
||
add_timestamp()
|
||
# 按缓存先后顺序发送
|
||
addr.send(msgs[i])
|
||
save2file(f"[SERVER --> {get_ccid(addr).decode() if get_ccid(addr) else 'None'}][{len(msgs[i])}] {msgs[i].hex(' ')}")
|
||
time.sleep(1)
|
||
# 成功则清空已发送成功的缓存数据
|
||
cache[ccid].pop(0)
|
||
logger.debug(f"第 {i + 1} 包缓存数据发送完成。")
|
||
except:
|
||
# 异常时退出循环，不用再尝试发送后续msgs
|
||
logger.error(f"第 {i + 1} 包发送失败，停止尝试")
|
||
s2wx("","",f"第{i + 1}包发送失败，停止尝试。")
|
||
break
|
||
if cache[ccid] == []:
|
||
logger.info(f"缓存数据全部发送完成。 {sccid} --> {ccid}")
|
||
s2wx("","",f"缓存数据全部发送完成。 \n{sccid} --> {ccid}")
|
||
del cache[ccid]
|
||
save_cache(cache)
|
||
|
||
return 0
|
||
|
||
|
||
def check_valid(data):
|
||
# 检查数据有效性
|
||
|
||
return bytearray(data[:])
|
||
|
||
def get_info(data):
|
||
# print(type(data[12]))
|
||
o=0
|
||
def h2s(x):
|
||
# x= int.from_bytes(x)
|
||
return str(x)
|
||
pass
|
||
sn = "_".join(list(map(h2s,data[12:19])))
|
||
index=int.from_bytes(data[20:21],byteorder="big")
|
||
cnt=int.from_bytes(data[21:22],byteorder="big")
|
||
t=f"流水号20{sn} \n序号\[{index}/{cnt}]"
|
||
logger.info(t)
|
||
s2wx("","",t)
|
||
|
||
|
||
def tt_trans(addr, data):
|
||
"""发送数据到指定ccid"""
|
||
|
||
get_info(data)
|
||
sccid = get_ccid(addr)
|
||
if not sccid:
|
||
return -1
|
||
tccid = data[7 + offs : 11 + offs]
|
||
# taddr = get_addr(tccid)
|
||
# 组帧
|
||
msg = check_valid(data)
|
||
msg[:5] = b"$TTMX"
|
||
msg[7 + offs : 11 + offs] = sccid
|
||
|
||
# 计算校验和
|
||
crc = 0
|
||
for i in msg[:-1]:
|
||
crc = crc ^ i
|
||
msg[-1] = crc
|
||
|
||
if tccid not in p:
|
||
# 未找到匹配的在线终端
|
||
|
||
# 更新缓存
|
||
# {tccid1=[msg1,msg2,...],...}
|
||
cache[tccid].append(msg)
|
||
save_cache(cache)
|
||
|
||
t=f"目标终端{tccid.decode()}未在线。 \n数据已缓存，待目标终端上线后重发。"
|
||
logger.info(t)
|
||
s2wx("","",t)
|
||
return 0
|
||
taddr = get_addr(tccid)
|
||
logger.info(f"匹配在线终端为：{tccid} <---> {taddr}")
|
||
|
||
# 发送
|
||
try:
|
||
add_timestamp()
|
||
taddr.send(msg)
|
||
save2file(f"[SERVER --> {get_ccid(taddr).decode() if get_ccid(taddr) else 'None'}][{len(msg)}] {msg.hex(' ')}",0)
|
||
logger.info(f"数据发送成功。 {sccid} --> {tccid}")
|
||
s2wx("","",f"匹配到在线终端, 数据发送成功。 \n{sccid.decode()} ---> {tccid.decode()}")
|
||
except Exception:
|
||
# 发送失败
|
||
taddr.close()
|
||
# 更新pairs，清空对应终端
|
||
update_pairs(taddr, tccid, 1)
|
||
# 更新缓存
|
||
# {tccid1=[msg1,msg2,...],...}
|
||
cache[tccid].append(msg)
|
||
save_cache(cache)
|
||
|
||
logger.error("发送失败，终端{}可能已掉线。数据已缓存，待目标终端上线后重发。", taddr)
|
||
s2wx("","",f"发送失败，终端{taddr}可能已掉线。 \n数据已缓存，待目标终端上线后重发。")
|
||
return 0
|
||
|
||
def isInlst(tar,lst):
|
||
"""
|
||
在列表中查找目标子列表
|
||
|
||
返回匹配的列表
|
||
"""
|
||
index = []
|
||
for x in range(len(lst)):
|
||
if lst[x:x+len(tar)] == tar :
|
||
index.append(x)
|
||
|
||
rst = []
|
||
for x in range(len(index)):
|
||
if x<len(index)-1:
|
||
rst.append(lst[index[x]:index[x+1]])
|
||
if x == len(index)-1:
|
||
rst.append(lst[index[-1]:])
|
||
|
||
return rst
|
||
|
||
# 便于拼合数据的缓存
|
||
recCache = {}
|
||
def data_split(addr):
|
||
# 按帧头分割长数据
|
||
|
||
h = b"$TTMS" # b'\x24\x54\x54\x4d\x53'
|
||
dlist = isInlst([0x24,0x54,0x54,0x4d,0x53],recCache[addr])
|
||
# print(dlist)
|
||
|
||
#校验每帧数据
|
||
for i in range(len(dlist)):
|
||
flen = (dlist[i][5]<<8)+dlist[i][6]
|
||
if flen != len(dlist[i]) :
|
||
logger.warning("第{}数据帧不匹配，应收{}<-->实收{}",i+1,flen,len(dlist[i]))
|
||
|
||
# 清空缓存
|
||
recCache[addr].clear()
|
||
|
||
# 因为无帧尾，需添加帧是否完整的校验
|
||
if len(dlist) :
|
||
# 目前情形多为两包粘包，故最后一包肯定不完整
|
||
# logger.debug(bytes(dlist[-1]).hex(" "))
|
||
flen = (dlist[-1][5]<<8)+dlist[-1][6]
|
||
if flen <= len(dlist[-1]) :
|
||
recCache[addr].extend(dlist[-1][flen:])
|
||
for x in range(len(dlist[-1])-flen):
|
||
dlist[-1].pop()
|
||
else:
|
||
recCache[addr].extend(dlist[-1][:])
|
||
dlist.pop()
|
||
|
||
logger.debug(f"解析为 {len(dlist)} 包数据。")
|
||
if len(recCache[addr]):
|
||
logger.warning("缓存数据为{}",bytes(recCache[addr]).hex(" "))
|
||
return dlist[:]
|
||
|
||
|
||
def tt_decode(addr, data):
|
||
"""处理收到的数据
|
||
|
||
数据正常返回0
|
||
数据异常返回-1"""
|
||
# 追加至缓存末尾
|
||
if addr not in recCache:
|
||
recCache[addr]=[]
|
||
recCache[addr].extend(data)
|
||
|
||
flag = -1
|
||
|
||
dlist = data_split(addr)
|
||
if len(dlist)==0:
|
||
logger.warning("not enough data")
|
||
|
||
for data in dlist:
|
||
data = bytes(data)
|
||
cmd = data[11 + offs]
|
||
# 回退到if-elif而不用match，因服务器默认版本3.8
|
||
if cmd == 0x01:
|
||
# 处理心跳
|
||
flag = tt_hh(addr, data)
|
||
elif cmd == 0xAA:
|
||
# 处理数据
|
||
if tt_trans(addr, data):
|
||
flag = -1
|
||
break
|
||
else:
|
||
flag = 0
|
||
else:
|
||
break
|
||
return flag
|
||
|
||
class MyServer(socketserver.BaseRequestHandler):
|
||
def setup(self):
|
||
self.request.settimeout(60*1.5)
|
||
self.err = " "
|
||
self.recDataBuff=[]
|
||
|
||
def handle(self): # 回调
|
||
add_timestamp()
|
||
t=f"TCP客户端{str(self.client_address)}已连接，等待上报心跳注册ccid。"
|
||
logger.info("TCP客户端{}已连接，等待上报心跳注册ccid。",str(self.client_address))
|
||
s2wx("天通消息","TCP客户端接入",t)
|
||
conn = self.request
|
||
# print(type(conn),conn.fd)
|
||
|
||
while True:
|
||
try:
|
||
data = conn.recv(1024000)
|
||
except socket.timeout:
|
||
self.err = "心跳超时。"
|
||
logger.warning("{self.client_address}超时。")
|
||
break
|
||
if not data:
|
||
break
|
||
add_timestamp()
|
||
self.recDataBuff.extend(data)
|
||
# logger.debug("[R] {}",data.hex(" "))
|
||
save2file(f"[{get_ccid(conn).decode() if get_ccid(conn) else 'None'} --> SERVER][{len(data)}] {data.hex(' ')}")
|
||
t = f"从{str(self.client_address)}<->[{get_ccid(conn).decode() if get_ccid(conn) else 'None'}]接收到新数据，长度{len(data)}字节。"
|
||
logger.info(t)
|
||
logger.debug("[R] - {}",data.hex(" "))
|
||
# print(data.hex(" "))
|
||
# 心跳报文17字节，避免推送心跳报文刷屏
|
||
if len(data) > 17:
|
||
s2wx("","",t)
|
||
# print("--",len(self.recDataBuff))
|
||
# print(self.recDataBuff[:])
|
||
if tt_decode(conn, data):
|
||
self.err= "切断。"
|
||
# logger.warning(f"{self.client_address}疑是非法连接")
|
||
# conn.close()
|
||
# break
|
||
|
||
def finish(self):
|
||
|
||
if self.request in pp:
|
||
t = f"""终端下线了。{self.err}
|
||
CCID:{pp[self.request].decode()}
|
||
地址:{str(self.client_address)}"""
|
||
update_pairs(self.request,get_ccid(self.request),1)
|
||
logger.info(t)
|
||
s2wx("","天通终端下线了",t)
|
||
recCache[self.request].clear()
|
||
# print("终端",self.client_address,"下线了。",self.err)
|
||
# update_pairs(self.request,get_ccid(self.request),1)
|
||
else:
|
||
t= f"TCP客户端{str(self.client_address)}已断开连接。{self.err}"
|
||
logger.info(t)
|
||
recCache[self.request].clear()
|
||
# s2wx("天通消息","TCP客户端断开",t)
|
||
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_,t = gettm()
|
||
if not os.path.exists('./debug'):
|
||
os.mkdir('./debug')
|
||
logger.add('/'.join(['./debug',t,'/log_{time}.log']))
|
||
# load_cache
|
||
if load_cache():
|
||
cache = load_cache()
|
||
socketserver.TCPServer.allow_reuse_address = True
|
||
server = socketserver.ThreadingTCPServer(("", 7222), MyServer)
|
||
add_timestamp()
|
||
logger.info("服务端{}初始化成功。", server.server_address)
|
||
s2wx("天通消息","服务器上线","服务器端初始化完成，等待终端连接。")
|
||
server.timeout = 10
|
||
server.serve_forever()
|
||
|
||
|
||
"""
|
||
以下几种情况服务器会主动断开终端：
|
||
单次数据不以帧头开始
|
||
单次数据长度不足11
|
||
未上报心跳（注册ccid）而直接发数据
|
||
"""
|