HookCli/yyyy_js/python-server/WebServer.py

import asyncio
import websockets
import logging

# Initialize logging
logging.basicConfig(level=logging.INFO)

ip = "192.168.137.1"
port = 9999

status = 0  # 1连接 0失败
# 从同目录下读取props.properties文件，获取配置(ip、port)信息
with open("props.properties", "r") as f:
    global ip, port
    props = f.readlines()
    for prop in props:
        if prop.startswith("ip"):
            ip = prop.split("=")[1].strip()
        if prop.startswith("port"):
            port = prop.split("=")[1].strip()


async def echo(websocket, path):
    global status
    try:
        async for message in websocket:
            logging.info(f"Received message: {message}")
            # 将所有的message，写到一个ws.log文件中
            with open("ws.log", "a") as f:
                f.write(f"{message}\n")

            if status != 1:
                status = 1
                await websocket.send(f"Server says: 你好，服务端收到，连接建立成功")
    except websockets.ConnectionClosed as e:
        status = 0
        logging.warning(f"Connection closed: {e.reason}")
    except Exception as e:
        status = 0
        logging.error(f"Error: {e}", exc_info=True)


async def main():
    # SSL context for wss (optional, uncomment if you want to use wss)
    # ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    # localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
    # ssl_context.load_cert_chain(localhost_pem)

    # Remove ssl=ssl_context below if not using wss
    async with websockets.serve(echo, ip, port):  # , ssl=ssl_context):
        await asyncio.Future()  # run forever


if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Server stopped by KeyboardInterrupt")
    except Exception as e:
        logging.error(f"Server error: {e}", exc_info=True)