Python 實(shí)現(xiàn) WebSocket 通信的過(guò)程詳解

更新時(shí)間：2024年06月22日 12:22:28 作者：微軟技術(shù)分享

WebSocket是一種在Web應(yīng)用程序中實(shí)現(xiàn)雙向通信的協(xié)議,與傳統(tǒng)的HTTP請(qǐng)求-響應(yīng)模型不同,WebSocket允許服務(wù)器主動(dòng)向客戶端推送數(shù)據(jù),實(shí)現(xiàn)實(shí)時(shí)性和互動(dòng)性,這篇文章主要介紹了Python 實(shí)現(xiàn) WebSocket 通信的過(guò)程詳解,需要的朋友可以參考下

WebSocket是一種在Web應(yīng)用程序中實(shí)現(xiàn)雙向通信的協(xié)議。與傳統(tǒng)的HTTP請(qǐng)求-響應(yīng)模型不同，WebSocket允許服務(wù)器主動(dòng)向客戶端推送數(shù)據(jù)，實(shí)現(xiàn)實(shí)時(shí)性和互動(dòng)性。

以下是WebSocket的一些關(guān)鍵特點(diǎn)：

雙向通信：WebSocket提供了全雙工通信，允許服務(wù)器和客戶端同時(shí)發(fā)送和接收數(shù)據(jù)。相比起傳統(tǒng)的HTTP請(qǐng)求-響應(yīng)模型，WebSocket使得服務(wù)器可以主動(dòng)推送數(shù)據(jù)給客戶端，而不需要客戶端首先發(fā)起請(qǐng)求。

持久連接：與HTTP請(qǐng)求不同，WebSocket連接是持久的，即一旦建立連接，它將保持打開(kāi)狀態(tài)，直到其中一方關(guān)閉連接或發(fā)生錯(cuò)誤。

輕量級(jí)協(xié)議：WebSocket使用簡(jiǎn)單的消息傳遞協(xié)議，在傳輸數(shù)據(jù)時(shí)減少了額外的開(kāi)銷。它采用二進(jìn)制幀格式或文本幀格式來(lái)發(fā)送數(shù)據(jù)。

跨域支持：WebSocket支持跨域通信，允許在不同域名或端口之間建立連接和交換數(shù)據(jù)。

WebSocket在許多場(chǎng)景下非常有用，特別是實(shí)時(shí)通信和實(shí)時(shí)數(shù)據(jù)更新方面。它被廣泛應(yīng)用于在線聊天、實(shí)時(shí)游戲、股票市場(chǎng)行情、協(xié)同編輯和實(shí)時(shí)協(xié)作等應(yīng)用程序中。

在編程中，可以使用WebSocket API來(lái)建立WebSocket連接，并使用WebSocket的事件和方法來(lái)處理連接的打開(kāi)、關(guān)閉、錯(cuò)誤和消息等操作。常見(jiàn)的編程語(yǔ)言和框架提供了對(duì)WebSocket的支持，使得開(kāi)發(fā)者可以方便地實(shí)現(xiàn)WebSocket功能。

實(shí)現(xiàn)簡(jiǎn)單通信

這段代碼實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的WebSocket客戶端，通過(guò)與服務(wù)器建立連接，可以實(shí)現(xiàn)實(shí)時(shí)的雙向通信。需要注意的是，連接的URL和端口需要根據(jù)實(shí)際情況進(jìn)行修改，確保與服務(wù)器端的WebSocket服務(wù)端口一致。

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <script src="https://cdn.lyshark.com/jquery/3.5.1/jquery.min.js"></script>
</head>
<body>
<ul id="content"></ul>
<form class="form">
    <input type="text" placeholder="請(qǐng)輸入發(fā)送的消息" class="message" id="message"/>
    <input type="button" value="連接" id="connect" class="connect"/>
    <input type="button" value="發(fā)送" id="send" class="connect"/>
</form>
<script type="text/javascript">
    var oUl=document.getElementById('content');
    var oConnect=document.getElementById('connect');
    var oSend=document.getElementById('send');
    var websocket=null;
    oConnect.onclick=function(){
        websocket=new WebSocket('ws://127.0.0.1:10083');
        <!--客戶端鏈接后觸發(fā)-->
         websocket.onopen=function(){
             oUl.innerHTML+="<li>客戶端已連接</li>";
         }
         <!--收到消息后觸發(fā)-->
        websocket.onmessage=function(evt){
            oUl.innerHTML+="<li>"+evt.data+"</li>";
        }
        <!--關(guān)閉后觸發(fā)-->
        websocket.onclose=function(){
            oUl.innerHTML+="<li>客戶端已斷開(kāi)連接</li>";
        };
        <!--出錯(cuò)后觸發(fā)-->
        websocket.onerror=function(evt){
            oUl.innerHTML+="<li>"+evt.data+"</li>";
        };
    };
    oSend.onclick=function(){
        if(websocket){
             websocket.send($("#message").val())
        }
    }
</script>
</body>
</html>

后端的main.py執(zhí)行處理任務(wù),主要處理流程集中在handler_msg函數(shù)上，如下這段代碼實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的WebSocket服務(wù)器。它使用了Python的socket和threading模塊來(lái)創(chuàng)建服務(wù)器，并處理與客戶端的握手和數(shù)據(jù)交互。

代碼中的主要函數(shù)和功能包括：

get_headers(data): 從請(qǐng)求數(shù)據(jù)中獲取請(qǐng)求頭部信息，并將其轉(zhuǎn)換為字典格式。

parse_payload(payload): 對(duì)接收到的數(shù)據(jù)進(jìn)行解碼，還原出原始消息。

send_msg(conn, msg_bytes): 封裝并發(fā)送數(shù)據(jù)到瀏覽器。

recv_msg(conn): 從瀏覽器中接收數(shù)據(jù)，并解析出原始消息。

handler_accept(sock): 建立握手流程，處理與客戶端的連接和握手過(guò)程。在握手完成后，創(chuàng)建一個(gè)新的線程來(lái)處理與客戶端的數(shù)據(jù)交互。

handler_msg(connect): 處理與客戶端的數(shù)據(jù)交互。通過(guò)循環(huán)不斷接收客戶端發(fā)送的消息，并發(fā)送一個(gè)固定的回復(fù)消息。

主函數(shù)部分：創(chuàng)建一個(gè)socket對(duì)象并綁定到本地主機(jī)的端口10083上，然后開(kāi)始監(jiān)聽(tīng)客戶端的連接請(qǐng)求。當(dāng)有新的連接請(qǐng)求時(shí)，將其分配給handler_accept()函數(shù)進(jìn)行處理，并在新線程中處理與客戶端的數(shù)據(jù)交互。

該代碼實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的WebSocket服務(wù)器，可以接收來(lái)自客戶端的連接請(qǐng)求，并與客戶端進(jìn)行數(shù)據(jù)交互。需要注意的是，服務(wù)器綁定的IP地址和端口號(hào)可以根據(jù)實(shí)際需求進(jìn)行修改。

import socket,struct,hashlib,base64
import threading
# 獲取請(qǐng)求頭部數(shù)據(jù),并將請(qǐng)求頭轉(zhuǎn)換為字典
def get_headers(data):
    headers = {}
    data = str(data, encoding="utf-8")
    header, body = data.split("\r\n\r\n", 1)
    header_list = header.split("\r\n")
    for i in header_list:
        i_list = i.split(":", 1)
        if len(i_list) >= 2:
            headers[i_list[0]] = "".join(i_list[1::]).strip()
        else:
            i_list = i.split(" ", 1)
            if i_list and len(i_list) == 2:
                headers["method"] = i_list[0]
                headers["protocol"] = i_list[1]
                print("請(qǐng)求類型: {} 請(qǐng)求協(xié)議: {}".format(i_list[0],i_list[1]))
    return headers
# 接收數(shù)據(jù)時(shí)的解碼過(guò)程
def parse_payload(payload):
    payload_len = payload[1] & 127
    if payload_len == 126:
        mask = payload[4:8]
        decoded = payload[8:]
    elif payload_len == 127:
        mask = payload[10:14]
        decoded = payload[14:]
    else:
        mask = payload[2:6]
        decoded = payload[6:]
    # 將所有數(shù)據(jù)全部收集起來(lái)，對(duì)所有字符串編碼
    bytes_list = bytearray()
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]
        bytes_list.append(chunk)
    body = str(bytes_list, encoding='utf-8')
    return body
# 封裝并發(fā)送數(shù)據(jù)到瀏覽器
def send_msg(conn, msg_bytes):
    # 接收的第一個(gè)字節(jié)都是x81不變
    first_byte = b"\x81"
    length = len(msg_bytes)
    if length < 126:
        first_byte += struct.pack("B", length)
    elif length <= 0xFFFF:
        first_byte += struct.pack("!BH", 126, length)
    else:
        first_byte += struct.pack("!BQ", 127, length)
    msg = first_byte + msg_bytes
    conn.sendall(msg)
    return True
# 從瀏覽器中接收數(shù)據(jù)
def recv_msg(conn):
    data_recv = conn.recv(8096)
    if data_recv[0:1] == b"\x81":
        data_parse = parse_payload(data_recv)
        return data_parse
    return False
# 建立握手流程并創(chuàng)建 handler_msg 完成數(shù)據(jù)收發(fā)
def handler_accept(sock):
    while True:
        conn, addr = sock.accept()
        data = conn.recv(8096)
        headers = get_headers(data)
        # 對(duì)請(qǐng)求頭中的sec-websocket-key進(jìn)行加密
        response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \
                       "Upgrade:websocket\r\n" \
                       "Connection: Upgrade\r\n" \
                       "Sec-WebSocket-Accept: %s\r\n" \
                       "WebSocket-Location: ws://%s\r\n\r\n"
        # 加鹽操作,此處是H5規(guī)范定義好的
        magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
        if headers.get('Sec-WebSocket-Key'):
            value = headers['Sec-WebSocket-Key'] + magic_string
        # 對(duì)數(shù)據(jù)進(jìn)行加解密
        ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())
        response_str = response_tpl % (ac.decode('utf-8'), headers.get("Host"))
        # 相應(yīng)握手包數(shù)據(jù)
        conn.sendall(bytes(response_str, encoding="utf-8"))
        t = threading.Thread(target=handler_msg, args=(conn, ))
        t.start()
# 主函數(shù),用于實(shí)現(xiàn)數(shù)據(jù)交互
def handler_msg(connect):
        with connect as connect_ptr:
            while True:
                try:
                    recv = recv_msg(connect_ptr)
                    print("接收數(shù)據(jù): {}".format(recv))
                    send_msg(connect_ptr, bytes("hello lyshark", encoding="utf-8"))
                except Exception:
                    exit(0)
if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", 10083))
    sock.listen(5)
    t = threading.Thread(target=handler_accept(sock))
    t.start()

繪制動(dòng)態(tài)圖形

如下這段代碼實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的實(shí)時(shí)CPU數(shù)據(jù)展示頁(yè)面。通過(guò)WebSocket與服務(wù)器建立連接，并實(shí)時(shí)接收和展示來(lái)自服務(wù)器的CPU數(shù)據(jù)。需要注意的是，頁(yè)面中WebSocket連接的IP地址和端口號(hào)需要根據(jù)實(shí)際情況進(jìn)行修改，確保與服務(wù)器建立正確的連接。另外，使用ECharts庫(kù)來(lái)繪制折線圖需要保證頁(yè)面能夠正確加載ECharts的JavaScript文件。

代碼中的主要部分包括：

引入ECharts庫(kù)：使用<script>標(biāo)簽引入ECharts庫(kù)的JavaScript文件，以便在頁(yè)面中使用ECharts。

HTML結(jié)構(gòu)：在頁(yè)面中創(chuàng)建一個(gè)<div>元素，用于顯示折線圖。

JavaScript代碼：定義了一個(gè)名為display的函數(shù)，用于初始化和更新折線圖。該函數(shù)接收兩個(gè)參數(shù)：時(shí)間數(shù)組和CPU數(shù)據(jù)數(shù)組。通過(guò)調(diào)用ECharts的setOption方法，配置折線圖的樣式和數(shù)據(jù)，并將其顯示在指定的<div>元素中。

WebSocket通信：創(chuàng)建一個(gè)WebSocket對(duì)象，與服務(wù)器建立WebSocket連接。通過(guò)監(jiān)聽(tīng)onmessage事件，接收服務(wù)器發(fā)送的CPU數(shù)據(jù)。將接收到的數(shù)據(jù)解析為JSON格式，將時(shí)間和CPU數(shù)據(jù)分別添加到對(duì)應(yīng)的數(shù)組中。當(dāng)時(shí)間數(shù)組和CPU數(shù)據(jù)數(shù)組的長(zhǎng)度達(dá)到10時(shí)，刪除數(shù)組中最舊的數(shù)據(jù)，并調(diào)用display函數(shù)更新折線圖。

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <script type="text/javascript" src="https://cdn.lyshark.com/echarts/5.0.0/echarts.min.js"></script>
</head>
<body>
<center><div id="main" style="height:400px;width:90%;border:1px solid #eecc11;padding:10px;"></div></center>
    <script type="text/javascript" charset="UTF-8">
            var display = function(time,cpu) {
                var main = echarts.init(document.getElementById(("main")));
                var option = {
                    xAxis: {
                        type: 'category',
                        data: time
                    },
                    yAxis: {
                        type: 'value'
                    },
                    series: [{
                        type: 'line',
                        smooth:0.3,
                        symbol: 'none',
                        color: 'blue',
                        smooth: true,
                        areaStyle: {
                            color: '#0000CD',
                            origin: 'start',
                            opacity: 0.5
                        },
                        data: cpu
                    }]
                };
                main.setOption(option,true);
            };
    </script>
    <script type="text/javascript" charset="UTF-8">
            var ws=new WebSocket('ws://127.0.0.1:10083');
            var time =["","","","","","","","","",""];
            var cpu = [0,0,0,0,0,0,0,0,0,0];
            ws.onmessage=function(evt)
            {
                var recv = JSON.parse(evt.data);
                time.push(recv.response[0]);
                cpu.push(parseFloat(recv.response[1]));
                if(time.length >=10){
                    time.shift();
                    cpu.shift();
                    console.log("時(shí)間:" + time + " --> CPU數(shù)據(jù): " + cpu);
                    display(time,cpu)
                }
            }
    </script>
</body>
</html>

后臺(tái)部分我們主要代碼不需要?jiǎng)樱恍枰薷?code>handler_msg處理流程即可，代碼能夠接受客戶端的連接請(qǐng)求，并定時(shí)發(fā)送CPU數(shù)據(jù)給客戶端。需要注意的是，代碼中的sock.bind(("127.0.0.1", 10083))綁定的是本地地址和指定的端口號(hào)，需要根據(jù)實(shí)際情況進(jìn)行修改。另外，需要確保在服務(wù)器環(huán)境中安裝了psutil庫(kù)，以便獲取CPU使用率數(shù)據(jù)。

import socket,struct,hashlib,base64
import threading
# 獲取請(qǐng)求頭部數(shù)據(jù),并將請(qǐng)求頭轉(zhuǎn)換為字典
def get_headers(data):
    headers = {}
    data = str(data, encoding="utf-8")
    header, body = data.split("\r\n\r\n", 1)
    header_list = header.split("\r\n")
    for i in header_list:
        i_list = i.split(":", 1)
        if len(i_list) >= 2:
            headers[i_list[0]] = "".join(i_list[1::]).strip()
        else:
            i_list = i.split(" ", 1)
            if i_list and len(i_list) == 2:
                headers["method"] = i_list[0]
                headers["protocol"] = i_list[1]
                print("請(qǐng)求類型: {} 請(qǐng)求協(xié)議: {}".format(i_list[0],i_list[1]))
    return headers
# 接收數(shù)據(jù)時(shí)的解碼過(guò)程
def parse_payload(payload):
    payload_len = payload[1] & 127
    if payload_len == 126:
        mask = payload[4:8]
        decoded = payload[8:]
    elif payload_len == 127:
        mask = payload[10:14]
        decoded = payload[14:]
    else:
        mask = payload[2:6]
        decoded = payload[6:]
    # 將所有數(shù)據(jù)全部收集起來(lái)，對(duì)所有字符串編碼
    bytes_list = bytearray()
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]
        bytes_list.append(chunk)
    body = str(bytes_list, encoding='utf-8')
    return body
# 封裝并發(fā)送數(shù)據(jù)到瀏覽器
def send_msg(conn, msg_bytes):
    # 接收的第一個(gè)字節(jié)都是x81不變
    first_byte = b"\x81"
    length = len(msg_bytes)
    if length < 126:
        first_byte += struct.pack("B", length)
    elif length <= 0xFFFF:
        first_byte += struct.pack("!BH", 126, length)
    else:
        first_byte += struct.pack("!BQ", 127, length)
    msg = first_byte + msg_bytes
    conn.sendall(msg)
    return True
# 從瀏覽器中接收數(shù)據(jù)
def recv_msg(conn):
    data_recv = conn.recv(8096)
    if data_recv[0:1] == b"\x81":
        data_parse = parse_payload(data_recv)
        return data_parse
    return False
# 建立握手流程并創(chuàng)建 handler_msg 完成數(shù)據(jù)收發(fā)
def handler_accept(sock):
    while True:
        conn, addr = sock.accept()
        data = conn.recv(8096)
        headers = get_headers(data)
        # 對(duì)請(qǐng)求頭中的sec-websocket-key進(jìn)行加密
        response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \
                       "Upgrade:websocket\r\n" \
                       "Connection: Upgrade\r\n" \
                       "Sec-WebSocket-Accept: %s\r\n" \
                       "WebSocket-Location: ws://%s\r\n\r\n"
        # 加鹽操作,此處是H5規(guī)范定義好的
        magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
        if headers.get('Sec-WebSocket-Key'):
            value = headers['Sec-WebSocket-Key'] + magic_string
        # 對(duì)數(shù)據(jù)進(jìn)行加解密
        ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())
        response_str = response_tpl % (ac.decode('utf-8'), headers.get("Host"))
        # 相應(yīng)握手包數(shù)據(jù)
        conn.sendall(bytes(response_str, encoding="utf-8"))
        t = threading.Thread(target=handler_msg, args=(conn, ))
        t.start()
# 主函數(shù),用于實(shí)現(xiàn)數(shù)據(jù)交互
def handler_msg(conn):
    with conn as c:
        while True:
            try:
                times = time.strftime("%M:%S", time.localtime())
                data = psutil.cpu_percent(interval=None, percpu=True)
                print("處理時(shí)間: {} --> 處理負(fù)載: {}".format(times, data))
                send_msg(c, bytes(json.dumps({"response": [times, data]}), encoding="utf-8"))
                time.sleep(60)
            except Exception:
                exit(0)
if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", 10083))
    sock.listen(5)
    t = threading.Thread(target=handler_accept(sock))
    t.start()

實(shí)現(xiàn)動(dòng)態(tài)交互

新建index.html寫(xiě)入如下代碼，這段代碼實(shí)現(xiàn)了一個(gè)Web頁(yè)面，包含了一個(gè)可交互的終端，可以通過(guò)輸入地址范圍和執(zhí)行命令來(lái)批量執(zhí)行命令，并實(shí)時(shí)顯示執(zhí)行結(jié)果。需要注意的是，代碼中的WebSocket連接地址和端口號(hào)需要根據(jù)實(shí)際情況進(jìn)行修改，確保與后端WebSocket服務(wù)器的地址和端口號(hào)一致。

代碼中的主要部分包括：

引入CSS和JavaScript庫(kù)：頁(yè)面引入了xterm.css和xterm.js來(lái)加載和渲染終端的外觀樣式和功能，同時(shí)還引入了jQuery庫(kù)用于簡(jiǎn)化DOM操作。

創(chuàng)建終端實(shí)例：通過(guò)new Terminal()創(chuàng)建了一個(gè)終端實(shí)例term，設(shè)置了終端的行數(shù)、列數(shù)、換行符轉(zhuǎn)換、光標(biāo)閃爍等屬性。

WebSocket連接：通過(guò)new WebSocket()創(chuàng)建了一個(gè)WebSocket實(shí)例sock，并指定了與后端的WebSocket服務(wù)器的地址和端口號(hào)。

終端打開(kāi)和消息接收：在sock.onopen回調(diào)函數(shù)中，調(diào)用term.open()打開(kāi)終端，并在控制臺(tái)輸出WebSocket連接成功的消息。在sock.onmessage回調(diào)函數(shù)中，解析接收到的JSON數(shù)據(jù)，提取地址和狀態(tài)信息，并將其顯示在終端中。

執(zhí)行命令：通過(guò)$('#send_message').click()事件處理函數(shù)，獲取輸入框中的地址和命令信息，將其封裝成JSON格式，并通過(guò)WebSocket發(fā)送給后端。

<html>
<head>
    <meta charSet="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
    <link rel="stylesheet"  rel="external nofollow" />
    <script src="https://cdn.lyshark.com/xterm/xterm.js"></script>
    <script src="https://cdn.lyshark.com/jquery/3.5.1/jquery.min.js" type="text/javascript"></script>
</head>
<body>
<div id="terminal"></div>
<input type="text" id="address" placeholder="主機(jī)范圍 127.0.0.1-100" style="width:200px;height:40px"/>
<input type="text" id="command" placeholder="執(zhí)行命令 ls -lh " style="width:400px;height:40px"/>
<input type="button" id="send_message" value="批量執(zhí)行">
<!--實(shí)現(xiàn)格式化字符串-->
<script type="text/javascript">
    $.format = function(source, params)
    {
        if (arguments.length == 1) return function()
        {
            var args = $.makeArray(arguments);
            args.unshift(source);
            return $.format.apply(this, args);
        };
        if (arguments.length > 2 && params.constructor != Array)
        {
            params = $.makeArray(arguments).slice(1);
        }
        if (params.constructor != Array)
        {
            params = [params];
        }
        $.each(params,
        function(i, n)
        {
            source = source.replace(new RegExp("\\{" + i + "\\}", "g"), n);
        });
        return source;
    };
</script>
<!--打開(kāi)終端,并開(kāi)始執(zhí)行命令-->
<script type="text/javascript">
    $(function(){
        var window_width = $(window).width()-200;
        var window_height = $(window).height()-300;
        var term = new Terminal(
        {
            cols: Math.floor(window_width/9),
            rows: Math.floor(window_height/20),
            convertEol: true,
            cursorBlink:false,
        });
        var sock = new WebSocket("ws://127.0.0.1:10083");
        sock.onopen = function () {
            term.open(document.getElementById('terminal'));
            console.log('WebSocket Open');
        };
        sock.onmessage = function (recv) {
             var data = JSON.parse(recv.data);
             console.log(data['addr'] + ' -- ' + data['status']);
             var temp = "\x1B[1;3;35m 地址:[ {0} ] \x1B[0m --> \x1B[1;3;33m 狀態(tài):[ {1} ] \x1B[0m";
             var string = $.format(temp, data['addr'],data['status']);
             term.writeln(string);
        };
        $('#send_message').click(function () {
            var message ={"address":null,"command":null};
            message['address'] = $("#address").val();
            message['command'] = $("#command").val();
            var send_data = JSON.stringify(message);
            window.sock.send(send_data);
        });
        window.sock = sock;
    });
</script>

后端代碼如下所示，其中核心代碼handler_msg(conn)，用于處理接收到的消息，解析地址范圍，并針對(duì)每個(gè)地址執(zhí)行命令。執(zhí)行結(jié)果會(huì)通過(guò)WebSocket發(fā)送回客戶端，并在后臺(tái)打印出對(duì)應(yīng)的主機(jī)和執(zhí)行的命令。需要注意的是，代碼中的函數(shù)recv_msg()和send_msg()沒(méi)有提供具體的實(shí)現(xiàn)，你需要根據(jù)具體的需求自行實(shí)現(xiàn)這兩個(gè)函數(shù)。

CalculationIP(Addr_Count)函數(shù)用于解析地址范圍，并生成對(duì)應(yīng)的IP地址列表。參數(shù)Addr_Count是一個(gè)地址范圍字符串，如"127.0.0.1-100"。函數(shù)首先根據(jù)"-"符號(hào)將地址范圍字符串分割為起始IP和結(jié)束IP的兩部分。然后根據(jù)起始IP生成IP地址的前三個(gè)部分，即IP地址的頭部。接著根據(jù)起始IP和結(jié)束IP的范圍，循環(huán)生成完整的IP地址，并將其添加到結(jié)果列表中。最后返回生成的IP地址列表。

handler_msg(conn)函數(shù)是一個(gè)無(wú)限循環(huán)，用于處理消息并執(zhí)行命令。在循環(huán)中，首先使用eval(recv_msg(c))函數(shù)接收并解析JSON格式的消息數(shù)據(jù)。然后從消息中獲取地址和命令信息。接下來(lái)調(diào)用CalculationIP()函數(shù)生成地址列表。然后遍歷地址列表，對(duì)每個(gè)地址執(zhí)行命令，并將執(zhí)行結(jié)果發(fā)送回客戶端。執(zhí)行過(guò)程中，會(huì)打印出對(duì)應(yīng)的主機(jī)和執(zhí)行的命令。在每次執(zhí)行完命令后，程序會(huì)暫停1秒鐘。如果在執(zhí)行過(guò)程中發(fā)生異常，程序會(huì)退出。

import socket,struct,hashlib,base64
import threading,psutil
# 獲取請(qǐng)求頭部數(shù)據(jù),并將請(qǐng)求頭轉(zhuǎn)換為字典
def get_headers(data):
    headers = {}
    data = str(data, encoding="utf-8")
    header, body = data.split("\r\n\r\n", 1)
    header_list = header.split("\r\n")
    for i in header_list:
        i_list = i.split(":", 1)
        if len(i_list) >= 2:
            headers[i_list[0]] = "".join(i_list[1::]).strip()
        else:
            i_list = i.split(" ", 1)
            if i_list and len(i_list) == 2:
                headers["method"] = i_list[0]
                headers["protocol"] = i_list[1]
                print("請(qǐng)求類型: {} 請(qǐng)求協(xié)議: {}".format(i_list[0],i_list[1]))
    return headers
# 接收數(shù)據(jù)時(shí)的解碼過(guò)程
def parse_payload(payload):
    payload_len = payload[1] & 127
    if payload_len == 126:
        mask = payload[4:8]
        decoded = payload[8:]
    elif payload_len == 127:
        mask = payload[10:14]
        decoded = payload[14:]
    else:
        mask = payload[2:6]
        decoded = payload[6:]
    # 將所有數(shù)據(jù)全部收集起來(lái)，對(duì)所有字符串編碼
    bytes_list = bytearray()
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]
        bytes_list.append(chunk)
    body = str(bytes_list, encoding='utf-8')
    return body
# 封裝并發(fā)送數(shù)據(jù)到瀏覽器
def send_msg(conn, msg_bytes):
    # 接收的第一個(gè)字節(jié)都是x81不變
    first_byte = b"\x81"
    length = len(msg_bytes)
    if length < 126:
        first_byte += struct.pack("B", length)
    elif length <= 0xFFFF:
        first_byte += struct.pack("!BH", 126, length)
    else:
        first_byte += struct.pack("!BQ", 127, length)
    msg = first_byte + msg_bytes
    conn.sendall(msg)
    return True
# 從瀏覽器中接收數(shù)據(jù)
def recv_msg(conn):
    data_recv = conn.recv(8096)
    if data_recv[0:1] == b"\x81":
        data_parse = parse_payload(data_recv)
        return data_parse
    return False
# 建立握手流程并創(chuàng)建 handler_msg 完成數(shù)據(jù)收發(fā)
def handler_accept(sock):
    while True:
        conn, addr = sock.accept()
        data = conn.recv(8096)
        headers = get_headers(data)
        # 對(duì)請(qǐng)求頭中的sec-websocket-key進(jìn)行加密
        response_tpl = "HTTP/1.1 101 Switching Protocols\r\n" \
                       "Upgrade:websocket\r\n" \
                       "Connection: Upgrade\r\n" \
                       "Sec-WebSocket-Accept: %s\r\n" \
                       "WebSocket-Location: ws://%s\r\n\r\n"
        # 加鹽操作,此處是H5規(guī)范定義好的
        magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
        if headers.get('Sec-WebSocket-Key'):
            value = headers['Sec-WebSocket-Key'] + magic_string
        # 對(duì)數(shù)據(jù)進(jìn)行加解密
        ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())
        response_str = response_tpl % (ac.decode('utf-8'), headers.get("Host"))
        # 相應(yīng)握手包數(shù)據(jù)
        conn.sendall(bytes(response_str, encoding="utf-8"))
        t = threading.Thread(target=handler_msg, args=(conn, ))
        t.start()
def CalculationIP(Addr_Count):
    ret = []
    try:
        IP_Start = str(Addr_Count.split("-")[0]).split(".")
        IP_Heads = str(IP_Start[0] + "." + IP_Start[1] + "." + IP_Start[2] +".")
        IP_Start_Range = int(Addr_Count.split(".")[3].split("-")[0])
        IP_End_Range = int(Addr_Count.split("-")[1])
        for item in range(IP_Start_Range,IP_End_Range+1):
            ret.append(IP_Heads+str(item))
        return ret
    except Exception:
        return 0
def handler_msg(conn):
        with conn as c:
            while True:
                try:
                    ref_json = eval(recv_msg(c))
                    address = ref_json.get("address")
                    command = ref_json.get("command")
                    address_list = CalculationIP(address)
                    for ip in address_list:
                        response = {'addr': ip, 'status': 'success'}
                        print("對(duì)主機(jī): {} --> 執(zhí)行: {}".format(ip,command))
                        send_msg(c, bytes(json.dumps(response) , encoding="utf-8"))
                        time.sleep(1)
                except Exception:
                    exit(0)
if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", 10083))
    sock.listen(5)
    t = threading.Thread(target=handler_accept(sock))
    t.start()

到此這篇關(guān)于Python 實(shí)現(xiàn) WebSocket 通信的文章就介紹到這了,更多相關(guān)Python WebSocket 通信內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: