网络调试助手for mac

#-*-coding:utf8;-*- #qpy:2 #qpy:webapp:TCPUDP #qpy://localhost:8080/ import sys, os, re, json import socket, threading, time, Queue from bottle import route, run, redirect from bottle import static_file, template if True: #Android QPython os.chdir(os.path.dirname(__file__)) sys.path.append('/data/data/com.hipipal.qpyplus/files/lib/python2.7/site-packages/websocket_server-0.4-py2.7.egg') sys.path.append('/data/data/com.hipipal.qpyplus/files/lib/python2.7/site-packages/CherryPy-3.8.0-py2.7.egg') @route('/') @route('/<info>') def index(info=''): net.sock_close() return template('index.html', ip=net.local_ip, info=info) @route('/tcpServ/<ip>/<port:int>') def tcpServ(ip, port): if net.sock_open('tcpServ', (ip, port)): return template('main.html', title='TCP服务器') else: redirect('/' + 'TCP服务器开启失败') @route('/tcpClin/<ip>/<port:int>') def tcpClin(ip, port): if net.sock_open('tcpClin', (ip, port)): return template('main.html', title='TCP客户端') else: redirect('/' + '连接到TCP服务器失败') @route('/udpServ/<ip>/<port:int>') def udpServ(ip, port): if net.sock_open('udpServ', (ip, port)): return template('main.html', title='UDP服务器') else: redirect('/' + 'UDP服务器开启失败') @route('/udpClin/<ip>/<port:int>') def udpClin(ip, port): if net.sock_open('udpClin', (ip, port)): return template('main.html', title='UDP客户端') else: redirect('/' + '连接到UDP服务器失败') @route('/static/<filepath:path>') def server_static(filepath): return static_file(filepath, root='./static/') @route('/__exit') def __exit(): threading.Thread(target=websocketServer.server.shutdown).start() class TCPUDP(object): def __init__(self): self.local_ip = socket.gethostbyname(socket.gethostname()) self.hexShow = False self.hexSend = False threading.Thread(target=self.sock_recv).start() threading.Thread(target=self.sock_listen).start() def sock_open(self, mode, addr): self.mode, self.addr = mode, addr if self.mode == 'tcpServ': self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.sock.bind(addr) self.sock.listen(10) except Exception: return False else: return True elif self.mode == 'tcpClin': self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.sock.connect(addr) except Exception: return False else: return True elif self.mode == 'udpServ': self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: self.sock.bind(addr) except Exception: return False else: return True elif self.mode == 'udpClin': self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) return True def sock_send(self, str): map = json.loads(str) if map['type'] == 'cmd': if map.has_key('hexShow'): self.hexShow = map['hexShow'] elif map.has_key('hexSend'): self.hexSend = map['hexSend'] else: str = map['data'] if self.mode == 'tcpServ': if hasattr(self, 'clinSock'): self.clinSock.send(str) elif self.mode == 'tcpClin': self.sock.send(str) elif self.mode == 'udpServ': if hasattr(self, 'clinAddr'): self.sock.sendto(str, self.clinAddr) elif self.mode == 'udpClin': self.sock.sendto(str, self.addr) def sock_recv(self): while True: try: if self.mode == 'tcpServ': self.clinSock.setblocking(False) data = self.clinSock.recv(1024) elif self.mode == 'tcpClin': self.sock.setblocking(False) data = self.sock.recv(1024) elif self.mode == 'udpServ': self.sock.setblocking(False) data, addr = self.sock.recvfrom(1024) elif self.mode == 'udpClin': self.sock.setblocking(False) data, addr = self.sock.recvfrom(1024) except Exception as ex: pass else: if self.mode == 'udpServ': self.clinAddr = addr print data if hasattr(websocketServer, 'client'): websocketServer.send_message(data) time.sleep(0.1) def sock_close(self): try: self.sock.shutdown(socket.SHUT_RDWR) except Exception: pass try: self.sock.close() except Exception: pass if hasattr(self, 'clinSock'): delattr(self, 'clinSock') if hasattr(self, 'clinAddr'): delattr(self, 'clinAddr') def sock_listen(self): while 1: try: sock, addr = self.sock.accept() except Exception: pass else: self.clinSock = sock time.sleep(5) class WebsocketServer(object): def __init__(self): import websocket_server self.server = websocket_server.WebsocketServer(13254) self.server.set_fn_new_client(self.websocket_new_client) self.server.set_fn_message_received(self.message_received) threading.Thread(target=self.server.run_forever).start() def websocket_new_client(self, client, server): self.client = client def message_received(self, client, server, message): net.sock_send(message) def send_message(self, msg): self.server.send_message(self.client, msg) if __name__ == '__main__': net = TCPUDP() websocketServer = WebsocketServer() #建立socket # run(server='cherrypy', host='localhost', port='8080', debug=True) # run(host='localhost', port='8081', debug=True) run(host='localhost', port='8081', debug=True) #启动本地net assistant