websocket工具开发

一、PySimpleGUI简介

        Python语言是一种功能强大的通用编程语言,它也是一种流行的GUI(图形用户界面)开发语言。Python GUI技术可以用于创建各种类型的应用程序,包括桌面应用程序、游戏、数据可视化工具等等,PySimpleGUI是一个用于创建简单GUI应用程序的Python库,它使用了Tkinter、Qt或wxPython GUI库创建用户界面。PySimpleGUI提供了各种小部件和布局管理器,使开发者可以轻松创建各种类型的简单GUI应用程序。

文件名称 文件介绍
build_packger.py 打包文件。
GameInterface.py 代码逻辑文件,处理软件服务逻辑并返回数据给客户端。
RunGame.py 客户端代码文件,处理客户页面交互。
zrunLogs.py 日志代码文件,实时获取保存工具产生的日志信息。
yaml_server.py 读取或修改工具的配置信息。
ToolsConfig.yaml 配置文件

二、GameInterface代码

#!/usr/bin/python
# -*- coding: utf-8 -*-

#@Users: LiMu
#@Files:GameInterface.py
#@Times: 2022/9/30 
#@Software:PyCharm

import os
import sys
import time
import uuid
import json
import websocket
import zlib
from jsonpath import jsonpath

import zrunLogs


class ConnectWebsocket(object):
    #Constructor initialization properties
    def __init__(self,Url,ServerId,GameName):
        self.ConnectServer = websocket.create_connection(Url)
        self.ServerId = ServerId
        self.GameName = GameName

    #Request method for game login interface
    def GameLogin(self,ModuleName:str,MethodName:str,UserName:str):
        #Assembly login parameters
        self.login_parameters = {"HandleCode": str(uuid.uuid4()).replace("-",""),"ModuleName": ModuleName,"MethodName":MethodName,"PartnerId": "1001","ServerId": self.ServerId,"PlayerId": "","Session": "","GameVersionId": "1212121","ResourceVersionName": "2232a","Data":[UserName]}
        #Send parameters to the server
        self.ConnectServer.send(json.dumps(self.login_parameters))
        while True:
            #Accept the returned information and process it
            if self.GameName == "ShootingVulture":
                recvs = self.ConnectServer.recv()
                zlibs = zlib.decompress(recvs)
                response = json.loads(zlibs)
            elif self.GameName == "BrightSword":
                recvs = self.ConnectServer.recv()
                response = json.loads(recvs)
            elif self.GameName == "SwordImmortals":
                import msgpack
                from lz4 import frame
                recvs = self.ConnectServer.recv()
                framess = frame.decompress(recvs)
                response = msgpack.loads(framess, strict_map_key=False)
                print(response)
                break
            else:
                response = self.ConnectServer.recv()
                print(response)
                break

            #Filter the request information returned
            if response.get("HandleCode") == self.login_parameters.get("HandleCode"):
                self.playerid = jsonpath(response,"$.Value.PlayerId")[0]
                self.sessionid = jsonpath(response,"$.Value.Session")[0]
                print("Login request parameters:{}".format(self.login_parameters))
                print("Login return information:{}".format(response))
                zrunLogs.getLog().comment_log().debug(self.login_parameters)
                zrunLogs.getLog().comment_log().debug(response)
                print("=" * 160)
                return response

    #Game service interface request method
    def PlayGame(self,ModuleAndMethod:list,Data:list):
        #Parameters of assembly salesman
        play_parameters = {"HandleCode": str(uuid.uuid4()).replace("-",""),"ModuleName":ModuleAndMethod[0],"MethodName":ModuleAndMethod[1],"PartnerId": "1001","ServerId": self.ServerId,"PlayerId":self.playerid,"Session":self.sessionid,"GameVersionId": "1212121","ResourceVersionName": "2232a","Data":Data}
        #Send parameters to the server
        self.ConnectServer.send(json.dumps(play_parameters))
        while True:
            #Accept the returned information and process it
            if self.GameName == "ShootingVulture":
                recvs = self.ConnectServer.recv()
                zlibs = zlib.decompress(recvs)
                response = json.loads(zlibs)
            elif self.GameName == "BrightSword":
                recvs = self.ConnectServer.recv()
                response = json.loads(recvs)
            elif self.GameName == "SwordImmortals":
                import msgpack
                from lz4 import frame
                recvs = self.ConnectServer.recv()
                framess = frame.decompress(recvs)
                response = msgpack.loads(framess, strict_map_key=False)
                print(response)
                break
            else:
                response = self.ConnectServer.recv()
                print(response)
                break


            #Filter the request information returned
            if play_parameters.get("HandleCode") == response.get("HandleCode"):
                print("API request parameters:{}".format(play_parameters))
                print("API returns information:{}".format(response))
                zrunLogs.getLog().comment_log().debug(play_parameters)
                zrunLogs.getLog().comment_log().debug(response)
                print("=" * 160)
                return response

    def GameOver(self):
        self.ConnectServer.close()

if __name__ == '__main__':
    pass

三、RunGame代码

#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Users: LiMu
# @Files:RunGame.py
# @Times: 2023/6/7
# @Software:PyCharm


import os
import json
import time
import random
import PySimpleGUI as sg
from yaml_server import yaml_read,yaml_update
from jsonpath import jsonpath
from GameInterface import ConnectWebsocket



layoutText = [[sg.Menu([["&使用说明","使用说明"],["&更新公告","更新公告"],["&关于我们","关于我们"]],font=400)]]

layInput = [[sg.Text("请求地址:",font=500),sg.InputText(yaml_read()["amumu"]["ServerUrl"],font=500,key="ServerUrl"),
             sg.Text("服务编号:",font=500),sg.InputText(yaml_read()["amumu"]["ServerId"],font=500,key="ServerId"),
             sg.Text("登录玩家:", font=500), sg.InputText("沐歌", font=500,size=(40,1),key="UserName"),
             sg.Button("保存配置",font=500,key="SaveConfiguration")],
            [sg.Text("前置模块:",font=500),sg.InputText("Player",font=500,key="Methoda1"),sg.Text("前置方法:",font=500),
             sg.InputText("AddGameRecoure",font=500,key="Methoda2"),sg.Text("前置参数:",font=500),
             sg.InputText(["1112,1112,1000"],key="Position",size=(40, 1),font=500),sg.CBox("启用哇",font=500,default=False,key="OpenPosition")],
            [sg.Text("业务模块:",font=500), sg.InputText("Player",font=500,key="Methodb1"), sg.Text("业务方法:",font=500),
             sg.InputText("AddGameRecoure",font=500,key="Methodb2"),sg.Text("选择游戏:",font=500),
             sg.InputCombo(["ShootingVulture","BrightSword","SwordImmortals"],font=600,key="gamename",default_value="ShootingVulture"),
            sg.Text("登录状态:",font=500),sg.InputText("尚未登录",font=500,text_color="red",size=(40,1),disabled=True,key="StatusMsg")
             ],
            [sg.Text("业务参数:",font=500), sg.InputText(["1112,1112,1000"],key="Business",size=(155, 4),font=500),
            sg.Button("发送参数", font=500,key="SendParameters")
            ],[sg.Text("断言语句:",font=500),sg.InputText(["$.StatusMsg","$.Value"],font=500,size=(155, 5),key="AssertionsStatement"),
               sg.Button("发放资源",font=500,key="ReleaseResources")],
            [sg.Multiline(["断言信息返回"],key="Assertions",size=(200,10),autoscroll=True)],
            [sg.Multiline("前置信息返回",key="Positionlog",size=(300,10),autoscroll=True)],
            [sg.Multiline("业务信息返回",key="Businesslog",size=(300,30),autoscroll=True)]
            ]



layout = layoutText+layInput

windows = sg.Window("游戏接口测试工具",layout,size=(1440,900),resizable=False) #定义主题标题和布局
windows.set_icon("leidaradar.svg")

while True:
    event,values = windows.read()
    if event == None:
        break


    if event == "SaveConfiguration":
        ServerId = str(values["ServerId"])
        ServerUrl = str(values["ServerUrl"])
        yaml_update(ServerId,ServerUrl)

    if event == "使用说明":
        DescriptionText1 = '''
        1、游戏以此是xx、xx、xx,默认xx。\n
        2、点击保存配置只会保存请求地址和服务编号(服务器ID)。\n
        3、请求参数可以元祖(a,b,c),列表[a,b,c]。\n
        4、阵列值"a","b","c",1,2,3表示一个字符串。\n
        5、复杂的参数["a",1,["b",2],{"key":"values"}]。\n
        6、断言格式"$.StatusMsg","$..playid",无数量限制。\n
        7、发放资源会默认发放5个英雄和升级资源。
        '''
        sg.Popup(DescriptionText1,title="使用说明")

    if event == "关于我们":
        DescriptionText2 = '''
        作者:沐歌\n
        扣扣:229421064\n
        版本:Version1.0.2\n
        '''
        sg.Popup(DescriptionText2,title="关于我们")

    if event == "更新公告":
        DescriptionText3 = '''
        版本更新内容:\n
        1、新增XX游戏模块。\n
        2、新增资源发放模块。\n
        3、修改BUG参数为空值报错。\n
        4、修改BUG参数为列表报错。
        '''
        sg.Popup(DescriptionText3,title="更新公告")

    if event == "SendParameters":
        AssertionsStatement = eval(values.get("AssertionsStatement"))
        # windows["Position"].Update(sendPosition)
        # windows["Business"].Update(sendBusiness)

        try:
            ConnectServer = ConnectWebsocket(values.get("ServerUrl"),values.get("ServerId"),values.get("gamename"))
            returnlogin = ConnectServer.GameLogin("Player","Login_ForDebugTest",values.get("UserName"))
            if jsonpath(returnlogin,"$.StatusMsg")[0] == "成功":
                windows["StatusMsg"].Update("登录成功",text_color="Green")

            if values.get("OpenPosition") == True:
                aii = []
                sendPosition = eval(values.get("Position"))
                if sendPosition == []:
                    aii = sendPosition
                elif type(sendPosition) == tuple:
                    for i in sendPosition:
                        aii.append(i)
                else:
                    aii.append(sendPosition)

                RePosition = ConnectServer.PlayGame([values.get("Methoda1"), values.get("Methoda2")], aii)
                windows.Element('Positionlog').Update(RePosition)

            bii = []
            sendBusiness = eval(values.get("Business"))
            if sendBusiness == []:
                bii = sendBusiness
            elif type(sendBusiness) == tuple or type(sendBusiness) == list:
                for j in sendBusiness:
                    bii.append(j)
            else:
                bii.append(sendBusiness)

            Responses = ConnectServer.PlayGame([values.get("Methodb1"), values.get("Methodb2")], bii)

            if jsonpath(Responses,"$.StatusMsg")[0] == "成功":
                cii = []
                Assertionslist = []
                if type(AssertionsStatement) == tuple:
                    for k in AssertionsStatement:
                        cii.append(str(k))
                    AssertionsStatement = cii
                for y in AssertionsStatement:
                    ayy = "{}={}".format(y,jsonpath(Responses,y))
                    Assertionslist.append(ayy)
                windows.Element('Assertions').Update(Assertionslist)
            else:
                windows.Element('Assertions').Update(f'StatusMsg={jsonpath(Responses,"$.StatusMsg")}')

            windows.Element('Businesslog').Update(Responses)
        except Exception as e:
            windows.Element('Businesslog').Update(e)

    if event == "ReleaseResources":
        if values.get("gamename")== "ShootingVulture":
            from Sd3AddPlayerResources import AddResources
            addresources = AddResources(values.get("ServerUrl"), values.get("ServerId"))
            addresources.PlayerLogin(values.get("UserName"))
            addresources.PveChapter()
            addresources.PlayerCharge("1000")
            addresources.AddRec(["1509,15093113,2000", "1509,15092107,2000", "1509,15092503,2000", "1509,15091604,2000",
                                 "1509,15091110,2000", "1519,15190125,2100"])
            addresources.GoodsUse(["15093113", "15092107", "15092503", "15091604", "15091110"], "100")
            addresources.AddRec(["1605,16050527,100", "1605,16055110,650", "1510,15100075,100", "1605,16050414,100",
                                 "1605,16050282,100", "1605,16050021,2000", "1605,16050297,300"])
            addresources.AddRec(["1111,1111,1000", "1112,1112,100000000", "1114,1114,200000000", "1605,16058004,1000"])
            addresources.GameOver()
            sg.Popup("资源发放成功,请重新登录游戏查看,感谢您的支持!", title="提示信息")
        elif values.get("gamename")== "BrightSword":
            from Lj6AddPlayerResources import AddResources
            addresources = AddResources(values.get("ServerUrl"), values.get("ServerId"))
            addresources.PlayerLogin(values.get("UserName"))
            addresources.GiveHero([12011103, 12012206, 12011201, 12012401, 12012204, 12011102], 360)
            addresources.AddRec(["1113*2000000", "16050001*5000", "16030068*2000", "16030013*1000", "16160001*1"])
            addresources.AddRec(["30000001*1", "30000002*1", "1129*100000", "25000010*6", "16050326*20"])
            addresources.GameOver()
            sg.Popup("资源发放成功,请重新登录游戏查看,感谢您的支持!",title="提示信息")
        else:
            sg.Popup("XX资源发放功能尚未开发,敬请期待!",title="温馨提示")

windows.close()

四、zrunLogs代码

#!/usr/bin/python
# -*- coding: utf-8 -*-

#@Users: LiMu
#@Files:get_logs.py
#@Times: 2021/12/26
#@Software:PyCharm

import os
import sys
import time
from loguru import logger


class getLog:
    @staticmethod
    def comment_log():
        get_time = time.strftime('%Y-%m-%d', time.localtime())
        log_dir = f"{os.path.dirname(os.path.abspath(__file__))}/Logs/{get_time}.log"
        #handler_id=None时关闭调试行打印
        logger.remove(handler_id=None)
        logger.add(log_dir,format="行号.{line} | 时间:{time:YYYY-MM-DD HH:mm:ss} | 路径:{module} | 级别:{level} | 信息:{message}",backtrace=True,diagnose=True,level="INFO",encoding="UTF-8")
        return logger

    #错误日志调试
    @staticmethod
    def try_log():
        try:
            4/0
        except Exception as e:
            return e

if __name__ == '__main__':
    getLog.comment_log().debug("this a debug!")
    getLog.comment_log().info("this a info!")
    getLog.comment_log().success("this a success!")
    getLog.comment_log().warning("this a warning!")
    getLog.comment_log().critical("this a critical!")
    # getLog.comment_log().exception(getLog.try_log())

五、yaml_server代码

#!/usr/bin/python
# -*- coding: utf-8 -*-

#@Users: LiMu
#@Files:yaml_server.py
#@Times: 2021/12/23 
#@Software:PyCharm

import os
import sys
import time
import yaml

#使用yaml读取yaml文件
def yaml_read()->dict:
    with open("ToolsConfig.yaml",encoding="utf-8") as files:
        read_data = yaml.load(files,Loader=yaml.Loader)
        return read_data

def yaml_update(ServerId:str,Urls:str):
    with open("ToolsConfig.yaml",encoding="utf-8") as ReadFiles:
        configtext = yaml.load(ReadFiles,Loader=yaml.Loader)

        configtext["amumu"]["devices"] = ServerId
        configtext["amumu"]["IsReal"] = Urls


    with open("ToolsConfig.yaml",'w',encoding='utf-8') as WriteFiles:
        yaml.dump(configtext,WriteFiles)


if __name__ == '__main__':
    yaml_update("127.0.0.1:7555","True")
    readss = yaml_read()
    print(readss["amumu"]["devices"])
    print(readss["amumu"]["IsReal"])
    print(type(readss["amumu"]["IsReal"]))
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容