所以我一直在开发一个带有基于模块的http请求的异步python scraper。为此，我一直在使用 requests 和 importlib，并且我想制作一个简单的小 GUI，可以使用请求的状态代码进行更新。我做到了。
一切都很好，但我似乎遇到了一个问题，因为请求已成功发送，GUI 显示了，但它仅在发送所有请求后才显示，而不是在发送请求时动态显示
我一直在研究教程和 Qtimer，但在我见过的所有教程和帮助线程中，例如:
我尝试根据我的情况实现代码，但我唯一能做的就是在发送请求的同时显示 GUI，但它保持卡住状态(不响应)，直到所有请求已完成
import trio
from asks import Session
import importlib
from PyQt5.QtWidgets import QLabel, QMainWindow, QApplication, QWidget, QVBoxLayout
from PyQt5 import QtCore
import qdarkstyle
app = QApplication(sys.argv)
app.setStyleSheet(qdarkstyle.load_stylesheet_pyqt5())
module = importlib.import_module("get_module")
good = 0
bad = 0
total = 0
class Menu(QMainWindow):
def __init__(self):
global good, bad, total
super().__init__()
self.setWindowTitle("Status Codes")
self.central_widget = QWidget()
self.setCentralWidget(self.central_widget)
lay = QVBoxLayout(self.central_widget)
self.resize(500, 350)
ok = QLabel("200: <font color='green'>0</font>")
ok.setAlignment(QtCore.Qt.AlignHCenter)
bad = QLabel("400: <font color='yellow'>0</font>")
bad.setAlignment(QtCore.Qt.AlignHCenter)
total = QLabel("Total: <font color='#00FF00'>0</font>")
total.setAlignment(QtCore.Qt.AlignHCenter)
r_total, r_good, r_bad = self.check()
QtCore.QTimer.singleShot(1000, lambda: self.updateLabels(r_total, r_good, r_bad))
lay.addWidget(ok)
lay.addWidget(bad)
lay.addWidget(total)
self.show()
def check(self):
async def worker1(s):
global ok
global bad
global total
if module.method.lower() == "get":
r = await s.get(module.request(), params=module.settings())
elif module.method.lower() == "post":
r = await s.post(module.request(), data=module.settings())
if any(x in r.status_code for x in module.error):
print("BAD -- " + module.request())
r_total += 1
r_invalid += 1
else:
print("GOOD -- " + module.request())
r_total += 1
r_valid += 1
print(r.text)
async def worker2(s):
global ok
global bad
global total
if module.method.lower() == "get":
r = await s.get(module.request(), params=module.settings())
elif module.method.lower() == "post":
r = await s.post(module.request(), data=module.settings())
if any(x in r.status_code for x in module.error):
print("BAD -- " + module.request())
r_total += 1
r_invalid += 1
else:
print("GOOD -- " + module.request())
r_total += 1
r_valid += 1
print(r.text)
async def example():
s = Session(connections=module.connections)
for i in range(10):
async with trio.open_nursery() as nursery:
nursery.start_soon(worker1, s)
nursery.start_soon(worker2, s)
trio.run(example)
print("Total:", r_total)
print("Total good:", r_valid)
print("Total bad:", r_invalid)
return r_total, r_valid, r_invalid
def updateLabels(self, r_total, r_card, r_invalid):
good.setText("200: <font color='green'>%s</font>" % (r_valid))
bad.setText("400: <font color='#00FF00'>%s</font>" % (r_invalid))
total.setText("Total: <font color='#F40D30'>%s</font>" % (r.total))
if __name__ == '__main__':
ex = Menu()
sys.exit(app.exec_())
现在我希望它能显示 GUI，并动态(或每 1 秒)200、400 和总标签显示已发出多少个请求，以及有多少个返回 200 和 400。
但是，它会显示它们(它确实显示总数、总数 200 和总数 400)，但只有在所有请求完成后才会显示，而不是动态显示
最佳答案
异步任务会阻塞 GUI 的线程，因此这些任务必须在另一个线程中执行，并通过信号将其发送到 GUI，这也使我们能够将业务逻辑和 GUI 分开，从而拥有更清晰的代码:
get_module.py
# coding: utf8
#======================= IMPORT AREA =======================
#here's where you import any module needed for the website
from random import randint
#===========================================================
#====================== SETTINGS AREA ======================
#here's where you declare the settings of the website such
#as method, error key, success key, custom settings, etc...
name = "GET Test Config"
method = 'GET' #Method is either GET, POST or CUSTOM
error = ['400', '401', '402', '403', '404', '405', '406', '407', '408', '409', '410', '411', '412', '413', '414', '415', '416', '417', '418', '421', '422', '423', '424', '425', '426', '428', '429', '431', '451','500', '501', '502', '503', '504', '505', '506', '507', '508', '510', '511']
connections = 5
#===========================================================
#====================== DEFINITON AREA =====================
#here's where the definitions are made.
#There's 2 defs:
#def request(): Which returns the url (with or without modifications)
#def settings(): returns the data to send in the GET request
#====== SETTINGS AREA ======
def request():
url = "https://httpbin.org/anything"
return url
def settings():
data = {'example':'example'}
return (data)
#===========================================================
main.py
import importlib
import multio
import qdarkstyle
import trio
from PyQt5 import QtCore, QtWidgets
from asks import Session
module = importlib.import_module("get_module")
class TaskWorker(QtCore.QObject):
totalChanged = QtCore.pyqtSignal(int)
validChanged = QtCore.pyqtSignal(int)
invalidChanged = QtCore.pyqtSignal(int)
def __init__(self, parent=None):
super(TaskWorker, self).__init__(parent)
self._total = 0
self._valid = 0
self._invalid = 0
@QtCore.pyqtProperty(int, notify=totalChanged)
def total(self):
return self._total
@total.setter
def total(self, value):
if self._total == value:
return
self._total = value
self.totalChanged.emit(self._total)
@QtCore.pyqtProperty(int, notify=validChanged)
def valid(self):
return self._valid
@valid.setter
def valid(self, value):
if self._valid == value:
return
self._valid = value
self.validChanged.emit(self._valid)
@QtCore.pyqtProperty(int, notify=invalidChanged)
def invalid(self):
return self._invalid
@invalid.setter
def invalid(self, value):
if self._invalid == value:
return
self._invalid = value
self.invalidChanged.emit(self._invalid)
@QtCore.pyqtSlot()
def check(self):
async def worker1(s):
if module.method.lower() == "get":
r = await s.get(module.request(), params=module.settings())
elif module.method.lower() == "post":
r = await s.post(module.request(), data=module.settings())
# if any(x in r.status_code for x in module.error):
if str(r.status_code) in module.error:
print("BAD -- " + module.request())
self.total += 1
self.invalid += 1
else:
print("GOOD -- " + module.request())
self.total += 1
self.valid += 1
print(r.text)
async def worker2(s):
if module.method.lower() == "get":
r = await s.get(module.request(), params=module.settings())
elif module.method.lower() == "post":
r = await s.post(module.request(), data=module.settings())
# if any(x in r.status_code for x in module.error):
if str(r.status_code) in module.error:
print("BAD -- " + module.request())
self.total += 1
self.invalid += 1
else:
print("GOOD -- " + module.request())
self.total += 1
self.valid += 1
print(r.text)
async def example():
s = Session(connections=module.connections)
for i in range(40):
async with trio.open_nursery() as nursery:
nursery.start_soon(worker1, s)
nursery.start_soon(worker2, s)
multio.init("trio")
trio.run(example)
class Menu(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("Status Codes")
self.central_widget = QtWidgets.QWidget()
self.setCentralWidget(self.central_widget)
lay = QtWidgets.QVBoxLayout(self.central_widget)
self.resize(500, 350)
self.ok = QtWidgets.QLabel(
"200: <font color='green'>0</font>",
alignment=QtCore.Qt.AlignHCenter,
)
self.bad = QtWidgets.QLabel(
"400: <font color='yellow'>0</font>",
alignment=QtCore.Qt.AlignHCenter,
)
self.total = QtWidgets.QLabel(
"Total: <font color='#00FF00'>0</font>",
alignment=QtCore.Qt.AlignHCenter,
)
lay.addWidget(self.ok)
lay.addWidget(self.bad)
lay.addWidget(self.total)
thread = QtCore.QThread(self)
thread.start()
self.worker = TaskWorker()
self.worker.moveToThread(thread)
self.worker.totalChanged.connect(self.updateTotal)
self.worker.validChanged.connect(self.updateValid)
self.worker.invalidChanged.connect(self.updateInvalid)
QtCore.QTimer.singleShot(0, self.worker.check)
@QtCore.pyqtSlot(int)
def updateTotal(self, total):
self.total.setText("Total: <font color='#F40D30'>%s</font>" % (total))
@QtCore.pyqtSlot(int)
def updateValid(self, valid):
self.ok.setText("200: <font color='green'>%s</font>" % (valid))
@QtCore.pyqtSlot(int)
def updateInvalid(self, invalid):
self.bad.setText("400: <font color='#00FF00'>%s</font>" % (invalid))
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
app.setStyleSheet(qdarkstyle.load_stylesheet_pyqt5())
ex = Menu()
ex.show()
sys.exit(app.exec_())
关于python - 让 PyQt5 标签在发送 http 请求时动态更新，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/55641561/