#!/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 import s2f 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值无效 """ # 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 def send(): c= " \n".join(msgcache[:3]) res = requests.post("http://push.nmhd.eu.org:12722/push/murmur", json={ "title": title, "description": description, "content": c, "token": "tt"}) print("nmhd",res) try: res = res.json()["success"] except: s="https://msgpusher.com/push/test111" # s="http://47.108.213.155:11722/push/murmur" res = requests.post(s, json={ "title": title, "description": description, "content": c, "token": "tt"}) print("msgpusher",s) # else: # if res["success"]: # return None # else: # return res["message"] if len(msgcache) >3: send() if len(msgcache): msgcache.pop(0) if len(msgcache): msgcache.pop(0) if len(msgcache): msgcache.pop(0) Timer(10,s2wx).start() elif (not len(content)) and len(msgcache): # 等效超时 send() if len(msgcache): msgcache.pop(0) if len(msgcache): msgcache.pop(0) Timer(10,s2wx).start() pushisbusy=0 # 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(): print("-------------------------↓",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): """更新终端库 addr:绑定tcp地址 ccid:终端号 clear:清空掉线终端""" if not clear: # 添加或更新客户端 pp[addr] = ccid p.clear if ccid not in p: t = f"""*信息:* 天通终端上线。 CCID:{ccid.decode()} 地址:{str(addr.getpeername())}""" print(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}" print(t) s2wx("","",t) 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 print("终端未上报心跳注册ccid。") return None """ +--------+--------------+----------+----------+--------------+------------------------+--------+ | ff | 长度 | 地址信息 | 信令 | 内容 | 校验位 | +--------+--------------+----------+----------+--------------+------------------------+--------+ | | 2Bytes | 4Bytes | 心跳 | 0x01 | 时间戳32bit（单位ms） | 1Bytes | + $TTMS +--------------+----------+----------+--------------+------------------------+--------+ | | 2Bytes | 4Bytes | 传输数据 | 0xaa | 载荷 | 1Bytes | +--------+--------------+----------+----------+--------------+------------------------+--------+ """ 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()) 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) print("服务器回复心跳包。") # 从缓存中匹配到数据才重发 # 仅有新终端上线时才重发 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]) # 成功则清空已发送成功的缓存数据 cache[ccid].pop(0) print("第", i + 1, "包缓存数据发送完成。") except: # 异常时退出循环，不用再尝试发送后续msgs print("第", i + 1, "包发送失败，停止尝试") s2wx("","",f"第{i + 1}包发送失败，停止尝试。") break if cache[ccid] == []: print("缓存数据全部发送完成。", 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:17]))) index=int.from_bytes(data[19:20],byteorder="big") cnt=int.from_bytes(data[20:21],byteorder="big") t=f"流水号20{sn} \n序号\[{index}/{cnt}]" print(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数据已缓存，待目标终端上线后重发。" print(t) s2wx("","",t) return 0 taddr = get_addr(tccid) print("匹配在线终端为：", tccid, "<--->", taddr) # 发送 try: add_timestamp() taddr.send(msg) print("数据发送成功。", 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) print("发送失败，终端可能已掉线。数据已缓存，待目标终端上线后重发。", taddr) s2wx("","",f"发送失败，终端{taddr}可能已掉线。 \n数据已缓存，待目标终端上线后重发。") return 0 def data_split(data): # 按帧头分割长数据 h = b"$TTMS" # b'\x24\x54\x54\x4d\x53' dlist = data.split(h)[1:] for i in range(len(dlist)): dlist[i] = h + dlist[i] print("解析为", len(dlist), "包数据。") return dlist def tt_decode(addr, data): """处理收到的数据 数据正常返回0 数据异常返回-1""" flag = -1 # 依据帧头和最小长度初判数据有效性 # 只要接收数据给的缓存足够大，即使是多条数据组成的长数据也默认帧格式是完整的，前几个字节必须是帧头 if data[:5] != b"$TTMS" or len(data) < 12: return -1 dlist = data_split(data) for data in dlist: 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*10) self.err = " " def handle(self): # 回调 add_timestamp() t=f"TCP客户端{str(self.client_address)}已连接，等待上报心跳注册ccid。" print(t) s2wx("天通消息","TCP客户端接入",t) conn = self.request # print(type(conn),conn.fd) while True: try: data = conn.recv(1024000) except socket.timeout: self.err = "心跳超时。" print(self.client_address, "超时。") break if not data: break add_timestamp() s2f.save2file(data) t = f"从{str(self.client_address)}<->[{get_ccid(conn).decode() if get_ccid(conn) else 'None'}]接收到新数据，长度{len(data)}字节。" print(t) print(data.hex(" ")) s2wx("","",t) if tt_decode(conn, data): self.err= "切断。" print(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) print(t) s2wx("","天通终端下线了",t) # 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}" print(t) s2wx("天通消息","TCP客户端断开",t) if __name__ == "__main__": # load_cache if load_cache(): cache = load_cache() socketserver.TCPServer.allow_reuse_address = True server = socketserver.ThreadingTCPServer(("", 7222), MyServer) add_timestamp() print("服务端初始化成功。", server.server_address) s2wx("天通消息","服务器上线","服务器端初始化完成，等待终端连接。") server.timeout = 10 server.serve_forever() """ 以下几种情况服务器会主动断开终端： 单次数据不以帧头开始 单次数据长度不足11 未上报心跳（注册ccid）而直接发数据 """