http协议接口
对于http协议接口，requests库用起来最方便，因此采用requests库来做http协议的接口测试。安装requests：
pip install requests1GET请求示例
输出结果："{"status":1002,"desc":"invilad-citykey"}"
2带参数的GET请求示例
参数：city城市名称
输出结果："{"data":{"yesterday":{"量化接口,date":"31日星期二","high":"高温27℃","fx":"南风","low":"低温17℃","fl":"","type":"小雨"},"city":"北京","forecast":[{"量化接口,date":"1日星期三","high":"高温29℃","fengli":"","low":"低温18℃","fengxiang":"南风","type":"晴"},{"date":"2日星期四","high":"高温30℃","fengli":"","low":"低温20℃","fengxiang":"南风","type":"晴"},{"date":"3日星期五","high":"高温30℃","fengli":"","low":"低温22℃","fengxiang":"南风","type":"阴"},{"date":"4日星期六","high":"高温26℃","fengli":"","low":"低温21℃","fengxiang":"南风","type":"中雨"},{"date":"5日星期天","high":"高温26℃","fengli":"","low":"低温20℃","fengxiang":"东风","type":"小雨"}],"ganmao":"感冒易发期，外出请适当调整衣物，注意补充水分。","wendu":"23"},"status":1000,"desc":"OK"}"
假如服务对请求中city的一致性进行校验，header中需要传入Authorization，为city值的md5
输出结果："{"data":{"yesterday":{"date":"31日星期二","high":"高温35℃","fx":"南风","low":"低温25℃","fl":"","type":"晴"},"city":"合肥","forecast":[{"date":"1日星期三","high":"高温30℃","fengli":"","low":"低温24℃","fengxiang":"北风","type":"小雨"},{"date":"2日星期四","high":"高温26℃","fengli":"","low":"低温23℃","fengxiang":"西北风","type":"阴"},{"date":"3日星期五","high":"高温29℃","fengli":"","low":"低温23℃","fengxiang":"东风","type":"多云"},{"date":"4日星期六","high":"高温29℃","fengli":"","low":"低温24℃","fengxiang":"东风","type":"小雨"},{"date":"5日星期天","high":"高温31℃","fengli":"","low":"低温24℃","fengxiang":"东北风","type":"小雨"}],"ganmao":"感冒低发期，天气舒适，请注意多吃蔬菜水果，多喝水哦。","wendu":"26"},"status":1000,"desc":"OK"}"
3POST请求示例
Websocket协议接口
安装websockets库
pip install websockets下面是一个NLP接口的例子，客户端连接服务端，发送文本，接收返回结果，其中文本需要进行base64加密。
import uuid
import websockets
async def send_msg(websocket):
text = "我要查余额"
bs = str(base64.b64encode(text.encode("utf-8")), "utf-8")
sid = uuid.uuid4()
timestamp = str(int(time.time()) * 1000)
msg = "{"app":"test","sn":"123","timeStamp":"" + timestamp + "","version":"1.0.0","data":"" + bs + "","
""sid":"" + str(sid) + ""} "
print(msg)
await websocket.send(msg)
rev_text = await websocket.recv()
print(rev_text)
async def client():
async with websockets.connect("ws://xxx.xxx.com/nlp") as websocket:
await send_msg(websocket)
if __name__ == "__main__":
asyncio.get_event_loop().run_until_complete(client())执行结果示例：{"appid":"test","sn":"123","timeStamp":"1630586234000","version":"0.0","data":"5oiR6KaB5p+l6K+i5L2Z6aKd",sid":"4c6f5a64-2374-4616-9385-376c0d25bc1e"}{"code":"0000","msg":"succ","nlp":"{"sceneKey":"","sid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","uid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","text":"我要查询余额",,"semantic":[{"intent":"querybalance","intentName":"查询余额","intentCnxt":"","template":"{Query}余额","score":0.865519,"slots":[]}],"dialog":{"status":0,"uid":"","reaction":{"text":""},"slots":null},"answer":{"answerType":"","text":"","emotion":"","topic":"","topicID":"","type":"","question":{"question":"","question_ws":""}}}","extendData":"","sid":"d28c7631-54d9-493e-8db8-f67a5ecc2a25","last":true}
grpc协议接口
grpc协议需要Python5版本以上，安装所需的库：
pip install grpcio
pip install grpcio-tools
下面以虚拟主播服务接口为例，服务使用了grpc双向流进行交互，与服务端建立连接后，会返回一个rtmp地址，推送虚拟主播视频流，向服务端发送文本，会驱动虚拟主播播报文本。
1编写avatar.proto文件
syntax = "proto3";
package protocol;
message AvatarRequest {
string app = 1;
string timestamp = 2;
string version = 3;
string anchorId = 4;
string vcn = 5;
bytes data = 6;
int32 width = 7;
int32 height = 8;
string type=9;
}
message AvatarResponse {
string code = 1;
string msg = 2;
string type = 3;
bytes data = 4;
string status = 5;
string sessionId = 6;
}
service AvatarInteraction {
rpc Interact (stream AvatarRequest) returns (stream AvatarResponse) {
}
}2使用工具生成grpc代码
python-mgrpc_tools.protoc-I.--python_out=.--grpc_python_out=../avatar.proto最终生成avatar_pbpy和avatar_pb2_grpc.py两个文件
3grpc请求示例
将上述两个文件，引入到项目代码中
import sys
import threading
import time
import grpc
import avatar_pb2 as pb
import avatar_pb2_grpc as avatar
class avatarClient(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.status = False
self.serverUrl = ""
self.anchorId = "1001"
self.vcn = "xiaoyuan"
self.app = "testapp"
self.type = "0"
self.sendTxt = False
self.text = ""
self.connectType = "1001"
self.pingType = "-1000"
self.width = 1080
self.height = 1920
self.sid = ""
self.print_res = printRes()
self.lastPingTime = int(time.time() * 1000)
# 连接服务
def connect_sever(self):
return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version="1.0.0",
anchorId=self.anchorId, vcn=self.vcn, type=self.connectType, width=self.width,
height=self.height)
# 发送文本
def send_text(self):
return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version="1.0.0",
anchorId=self.anchorId, vcn=self.vcn, type="3", data=str(self.text).encode("utf-8"))
# 发送ping
def send_ping(self):
return pb.AvatarRequest(app=self.app, timestamp=str(int(time.time() * 1000)), version="1.0.0",
anchorId=self.anchorId, vcn=self.vcn, type=self.pingType)
def send_message(self):
yield self.connect_sever()
#间隔10秒,发送ping,进行连接保活
while True:
if self.sendTxt:
yield self.send_text()
self.sendTxt = False
if (int(time.time() * 1000) - int(self.lastPingTime)) < 10000:
continue
else:
yield self.send_ping()
self.lastPingTime = int(time.time() * 1000)
def run(self):
self.status = True
if self.serverUrl == "":
return
with grpc.insecure_channel(self.serverUrl) as channel:
stub = avatar.AvatarInteractionStub(channel)
self.channel = channel
responses = stub.Interact(self.send_message())
self.print_res.setClient(responses, self)
self.print_res.start()
time.sleep(1)
while True:
if self.print_res.runStatus == False:
break
client.text = input("input your text:
")
client.sendTxt = True
time.sleep(1)
def stop(self):
try:
self.status = False
self.print_res.runStatus = False
time.sleep(1)
except KeyboardInterrupt:
pass
class printRes(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.runStatus = True
def setClient(self, responses, client):
self.responses = responses
self.client = client
def run(self):
try:
for response in self.responses:
print(response)
if not self.runStatus:
sys.exit(0)
if response.code != "000000":
self.runStatus = False
sys.exit(0)
except Exception:
print("err")
if __name__ == "__main__":
client = avatarClient()
client.serverUrl = "xxx.xxx.xxx.xxx:xxxx"
client.start()打造自己的测试工具
在编辑器中执行测试，有时候交互性不太友好，因此我们可以打造一个带交互界面的测试工具。QT是跨平台C++库的集合，它实现高级API来访问现代桌面和移动系统的许多方面。这些服务包括定位和定位服务、多媒体、NFC和蓝牙连接、基于铬的web浏览器以及传统的UI开发。PyQt5是一套针对QtV5的全面Python绑定。它被实现为超过35个扩展模块，并使Python可以作为一个替代的应用程序开发语言C++在所有支持的平台上包括IOS和Android。
1安装PyQt5
pip install PyQt5如果要使用像界面进行设计，可以安装QtDesigner形界面开发工具，相关配置可以自行搜索。
pip install PyQt5-tools2打造websocket测试工具
下面以websocket接口为例，实现一个nlp的测试工具
# -*- coding: utf-8 -*-
import sys
import asyncio
import base64
import time
import uuid
import websockets
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtWidgets import *
class Ui_mainWindow(object):
def __init__(self, mainWindow):
self.mainWindow = mainWindow
self.setupUi()
def setupUi(self):
self.mainWindow.setObjectName("mainWindow")
self.mainWindow.setWindowModality(QtCore.Qt.WindowModal)
self.mainWindow.resize(600, 300)
self.mainWindow.setWindowTitle("NLP测试工具")
self.centralwidget = QtWidgets.QWidget(self.mainWindow)
self.centralwidget.setObjectName("centralwidget")
self.widget1 = QtWidgets.QWidget(self.centralwidget)
self.widget1.setGeometry(QtCore.QRect(10, 30, 400, 25))
self.widget1.setObjectName("widget1")
self.horizontalLayout = QtWidgets.QHBoxLayout(self.widget1)
self.horizontalLayout.setContentsMargins(0, 0, 0, 0)
self.horizontalLayout.setObjectName("horizontalLayout")
self.serverUrlLabel = QtWidgets.QLabel(self.widget1)
self.serverUrlLabel.setObjectName("label")
self.serverUrlLabel.setText("服务地址:")
self.horizontalLayout.addWidget(self.serverUrlLabel)
self.serverUrlLineEdit = QtWidgets.QLineEdit(self.widget1)
self.serverUrlLineEdit.setEnabled(True)
self.serverUrlLineEdit.setText("ws://xxx..xxx.xxx/nlp")
self.serverUrlLineEdit.setMinimumWidth(320)
self.serverUrlLineEdit.setMaximumWidth(320)
self.serverUrlLineEdit.setObjectName("serverUrlLineEdit")
self.horizontalLayout.addWidget(self.serverUrlLineEdit, stretch=500, alignment=QtCore.Qt.AlignLeft)
self.sendTextWidget = QtWidgets.QWidget(self.centralwidget)
self.sendTextWidget.setGeometry(QtCore.QRect(0, 70, 500, 70))
self.sendTextWidget.setObjectName("sendTextWidget")
self.sendTextLayout = QtWidgets.QHBoxLayout(self.sendTextWidget)
self.sendTextLayout.setContentsMargins(0, 0, 0, 0)
self.sendTextLayout.setObjectName("sendTextLayout")
self.sendTextLabel = QtWidgets.QLabel(self.sendTextWidget)
self.sendTextLabel.setContentsMargins(20, 0, 0, 0)
self.sendTextLabel.setText("文本:")
self.sendTextLayout.addWidget(self.sendTextLabel)
self.sendTextTextEdit = QtWidgets.QTextEdit(self.sendTextWidget)
self.sendTextTextEdit.setObjectName("sendTextTextEdit")
self.sendTextLayout.addWidget(self.sendTextTextEdit)
self.sendTextButton = QtWidgets.QPushButton(self.sendTextWidget)
self.sendTextButton.setObjectName("sendText")
self.sendTextButton.setText("发送")
self.sendTextButton.clicked.connect(self.send_text)
self.sendTextLayout.addWidget(self.sendTextButton)
self.widget3 = QtWidgets.QWidget(self.centralwidget)
self.widget3.setGeometry(QtCore.QRect(0, 160, 420, 90))
self.widget3.setObjectName("widget3")
self.logLayout = QtWidgets.QHBoxLayout(self.widget3)
self.logLayout.setContentsMargins(0, 0, 0, 0)
self.logLayout.setObjectName("logLayout")
self.logLabel = QtWidgets.QLabel(self.widget3)
self.logLabel.setObjectName("logLabel")
self.logLabel.setContentsMargins(20, 0, 0, 0)
self.logLabel.setLineWidth(30)
self.logLabel.setText("返回:")
self.logLayout.addWidget(self.logLabel)
self.logTextEdit = QtWidgets.QTextEdit(self.widget3)
self.logTextEdit.setContentsMargins(0, 0, 0, 0)
self.logTextEdit.setObjectName("logTextEdit")
self.logLayout.addWidget(self.logTextEdit)
self.mainWindow.setCentralWidget(self.centralwidget)
self.statusbar = QtWidgets.QStatusBar(self.mainWindow)
self.statusbar.setObjectName("statusbar")
self.mainWindow.setStatusBar(self.statusbar)
QtCore.QMetaObject.connectSlotsByName(self.mainWindow)
def send_text(self):
self.websocketClient = WebsocketClient()
self.websocketClient.url = self.serverUrlLineEdit.text()
self.websocketClient.text = self.sendTextTextEdit.toPlainText()
result = self.websocketClient.run()
self.logTextEdit.setText(result)
class WebsocketClient:
def __init__(self):
self.url = ""
async def send_msg(self, websocket):
if self.text == "":
return "empty text"
bs = str(base64.b64encode(self.text.encode("utf-8")), "utf-8")
sid = uuid.uuid4()
timestamp = str(int(time.time()) * 1000)
msg = "{"app":"test","sn":"123","timeStamp":"" + timestamp + "","version":"1.0.0","data":"" + bs + "","
""sid":"" + str(sid) + ""} "
await websocket.send(msg)
rev_text = await websocket.recv()
self.result = rev_text
self.text = ""
return rev_text
async def connect(self):
try:
async with websockets.connect(self.url) as websocket:
res = await self.send_msg(websocket)
return res
except Exception as e:
return str(e)
def run(self):
return asyncio.get_event_loop().run_until_complete(self.connect())
if __name__ == "__main__":
app = QApplication(sys.argv)
window = QMainWindow()
box = Ui_mainWindow(window)
window.show()
sys.exit(app.exec_())
文章为作者独立观点，不代表 股票程序化软件自动交易接口观点
相关文章
股民评论