""" 低吸代理服务器 """ import http import json import logging import socket import socketserver import threading import time from http.server import BaseHTTPRequestHandler from urllib.parse import parse_qs # 禁用http.server的日志输出 import requests from utils import socket_util, tool logger = logging.getLogger("http.server") logger.setLevel(logging.CRITICAL) # 183.234.94.163/125.93.72.195 REAL_HOST, REAL_PORT = "183.234.94.163", 12881 class DataServer(BaseHTTPRequestHandler): # 禁用日志输出 def log_message(self, format, *args): pass def __get_params(self, url): ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()]) return ps_dict def do_GET(self): path = self.path whole_path = f"http://{REAL_HOST}:{REAL_PORT}{path}" response = requests.get(whole_path) self.send_response(response.status_code) if response.status_code == 200: # 发给请求客户端的响应数据 self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(response.text.encode()) else: self.wfile.write(response.text.encode()) def do_POST(self): path = self.path whole_path = f"http://{REAL_HOST}:{REAL_PORT}{path}" params = self.__parse_request() response = requests.post(whole_path, json=params) self.__send_response(response.text) def __send_response(self, data): # 发给请求客户端的响应数据 self.send_response(200) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(data.encode()) def __parse_request(self): params = {} datas = self.rfile.read(int(self.headers['content-length'])) _str = str(datas, encoding="gbk") # print(_str) try: params = json.loads(_str) except: pass return params class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): pass class ClientSocketManager: # 客户端类型 CLIENT_TYPE_COMMON = "common" socket_client_dict = {} socket_client_lock_dict = {} active_client_dict = {} @classmethod def add_client(cls, _type, rid, sk): if _type == cls.CLIENT_TYPE_COMMON: # 交易列表 if _type not in cls.socket_client_dict: cls.socket_client_dict[_type] = [] cls.socket_client_dict[_type].append((rid, sk)) cls.socket_client_lock_dict[rid] = threading.Lock() else: cls.socket_client_dict[_type] = (rid, sk) cls.socket_client_lock_dict[rid] = threading.Lock() @classmethod def acquire_client(cls, _type): if _type == cls.CLIENT_TYPE_COMMON: if _type in cls.socket_client_dict: # 根据排序活跃时间排序 client_list = sorted(cls.socket_client_dict[_type], key=lambda x: cls.active_client_dict.get(x[0]) if x[ 0] in cls.active_client_dict else 0, reverse=True) for d in client_list: if d[0] in cls.socket_client_lock_dict: try: if cls.socket_client_lock_dict[d[0]].acquire(blocking=False): return d except threading.TimeoutError: pass else: if _type in cls.socket_client_dict: try: d = cls.socket_client_dict[_type] if d[0] in cls.socket_client_lock_dict: if cls.socket_client_lock_dict[d[0]].acquire(blocking=False): return d except threading.TimeoutError: pass return None @classmethod def release_client(cls, rid): if rid in cls.socket_client_lock_dict: # 释放锁 cls.socket_client_lock_dict[rid].release() @classmethod def del_client(cls, rid): # 删除线程锁 if rid in cls.socket_client_lock_dict: cls.socket_client_lock_dict.pop(rid) # 删除sk for t in cls.socket_client_dict: if type(cls.socket_client_dict[t]) == list: for d in cls.socket_client_dict[t]: if d[0] == rid: cls.socket_client_dict[t].remove(d) try: d[1].close() except: pass break elif type(cls.socket_client_dict[t]) == tuple: if cls.socket_client_dict[t][0] == rid: cls.socket_client_dict.pop(t) try: t[1].close() except: pass break # 心跳信息 @classmethod def heart(cls, rid): cls.active_client_dict[rid] = time.time() @classmethod def del_invalid_clients(cls): # 清除长时间无心跳的客户端通道 for k in cls.active_client_dict.keys(): if time.time() - cls.active_client_dict[k] > 20: # 心跳时间间隔20s以上视为无效 cls.del_client(k) @classmethod def list_client(cls, type_=None): """ :param type_: :return:[(客户端ID, 是否锁定, 活跃时间)] """ _type = type_ if not _type: _type = cls.CLIENT_TYPE_COMMON client_list = sorted(cls.socket_client_dict[_type], key=lambda x: cls.active_client_dict.get(x[0]) if x[0] in cls.active_client_dict else 0, reverse=True) fdata = [] for client in client_list: active_time = cls.active_client_dict.get(client[0]) if active_time is None: active_time = 0 active_time = tool.to_time_str(int(active_time)) fdata.append( (client[0], cls.socket_client_lock_dict[client[0]].locked(), active_time)) return fdata class MyTCPServer(socketserver.TCPServer): def __init__(self, server_address, RequestHandlerClass): socketserver.TCPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate=True) # 如果使用异步的形式则需要再重写ThreadingTCPServer class MyThreadingTCPServer(socketserver.ThreadingMixIn, MyTCPServer): pass class MyBaseRequestHandle(socketserver.BaseRequestHandler): __inited = False def setup(self): self.__init() @classmethod def __init(cls): if cls.__inited: return True cls.__inited = True cls.__req_socket_dict = {} @classmethod def getRecvData(cls, skk): data = "" header_size = 10 buf = skk.recv(header_size) header_str = buf if buf: start_time = time.time() buf = buf.decode('utf-8') if buf.startswith("##"): content_length = int(buf[2:10]) received_size = 0 while not received_size == content_length: r_data = skk.recv(10240) received_size += len(r_data) data += r_data.decode('utf-8') else: data = skk.recv(1024 * 1024) data = buf + data.decode('utf-8') return data, header_str def handle(self): host = self.client_address[0] super().handle() sk: socket.socket = self.request while True: try: data, header = self.getRecvData(sk) if data: data_str = data # print("收到数据------", f"{data_str[:20]}......{data_str[-20:]}") data_json = None try: data_json = json.loads(data_str) except json.decoder.JSONDecodeError as e: # JSON解析失败 sk.sendall(socket_util.load_header(json.dumps( {"code": 100, "msg": f"JSON解析失败"}).encode( encoding='utf-8'))) continue type_ = data_json["type"] __start_time = time.time() try: if type_ == 'register': client_type = ClientSocketManager.CLIENT_TYPE_COMMON rid = data_json["rid"] ClientSocketManager.add_client(client_type, rid, sk) sk.sendall(json.dumps({"type": "register"}).encode(encoding='utf-8')) try: # print("客户端", ClientSocketManager.socket_client_dict) while True: result, header = self.getRecvData(sk) try: resultJSON = json.loads(result) if resultJSON["type"] == 'heart': # 记录活跃客户端 ClientSocketManager.heart(resultJSON['client_id']) except json.decoder.JSONDecodeError as e: print("JSON解析出错", result, header) if not result: sk.close() break time.sleep(1) except ConnectionResetError as ee: ClientSocketManager.del_client(rid) except Exception as e: logging.exception(e) elif data_json["type"] == "push_msg": # 华鑫内部服务器推送过来的信息，需要推送到注册的客户端上面 client_info_list = ClientSocketManager.socket_client_dict.get(ClientSocketManager.CLIENT_TYPE_COMMON) clients = [x[1] for x in client_info_list] if client_info_list: for client in client_info_list: try: client[1].sendall( socket_util.load_header( json.dumps(data_json).encode(encoding='utf-8'))) except Exception as e: logging.exception(e) ClientSocketManager.del_client(client[0]) break except Exception as e: pass finally: sk.sendall( socket_util.load_header( json.dumps({"code": 0}).encode(encoding='utf-8'))) else: # 断开连接 break # sk.close() except Exception as e: # log.logger_tuoguan_request_debug.exception(e) logging.exception(e) break def finish(self): super().finish() def run_socket_server(port): laddr = "0.0.0.0", port print("SocketServer is at: http://%s:%d/" % (laddr)) tcpserver = MyThreadingTCPServer(laddr, MyBaseRequestHandle) # 注意：参数是MyBaseRequestHandle tcpserver.serve_forever() def run(addr, port): handler = DataServer try: httpd = ThreadedHTTPServer((addr, port), handler) print("HTTP server is at: http://%s:%d/" % (addr, port)) httpd.serve_forever() except Exception as e: logging.exception(e) if __name__ == "__main__": threading.Thread(target=run_socket_server, args=(10009,), daemon=True).start() run("", 12008)