• Welcome to the world's largest Chinese hacker forum

    Welcome to the world's largest Chinese hacker forum, our forum registration is open! You can now register for technical communication with us, this is a free and open to the world of the BBS, we founded the purpose for the study of network security, please don't release business of black/grey, or on the BBS posts, to seek help hacker if violations, we will permanently frozen your IP and account, thank you for your cooperation. Hacker attack and defense cracking or network Security

    business please click here: Creation Security  From CNHACKTEAM

200 行,一个河岸计算窗口后台实现服务,例子


Recommended Posts

直接上代码

导入系统

从输入导入字典,列表

从实现导入网

进口异步

从函数工具导入包装

来自信号灯输入信号

从PySide6导入QtCore作为质量控制

从PySide6导入模块作为栅极电荷

从PySide6导入QtWidgets作为q波

定义获取(路径):

定义_f1(fn):

@ wrapps(fn)

异步定义_f2(self: '路线',*参数,* *千瓦):

带自锁的异步:

self.count=1

if self.count=self.max:

self.max=self.count

ret=await fn(self,*args,**kw)

带自锁的异步:

self.count -=1

返回浸水使柔软

_f2.path=path

_f2.method='get '

返回_f2

return _f1

课程路线:

def __init__(self,app: web .应用)-无:

self.lock=asyncio .锁定()

self.count=0

self.max=0

自助添加路线(应用程序)

@get('/')

异步定义get_g1(自身,请求):

return {1: 2,2: 3,3: '中国'}

def add_routes(self,app: web .申请):

get_lst=[]

以自我的名义. dir__():

if name.startswith('get_'):

att=getattr(self,name)

如果可调用(电话号码):

断言hasattr(属性,“路径”)

断言hasattr(属性,“方法”)

path=getattr(att,' path ')

get_lst.append(web.get(path,att))

app.add_routes(get_lst)

异步定义记录器_mdw(应用程序,处理程序):

异步定义_h1(请求:网站.请求):

打印(请求方法,请求路径)

返回等待处理程序(请求)

return _h1

异步定义响应_mdw(应用程序,处理程序):

异步定义_h1(请求:网站.请求):

ret=await处理程序(请求)

打印(类型(ret))

if isinstance(ret,web .回应):

打印(3)

返回浸水使柔软

elif isinstance(ret,(str,int,float)):

打印(1)

返回网页.Response(text=str(ret))

elif isinstance(ret,(Dict,List)):

返回web.json_response(ret)

else:

打印(4)

返回网页.Response(text=str(ret))

return _h1

支援车(Suport Vehicle)类(qc .QObject):

sg=信号()

def __init__(self,loop: asyncio .抽象事件循环,ft: asyncio .未来)-无:

超级()。__init__()

自循环=循环

self.ft=ft

异步定义运行_服务器(自身):

服务器=web .应用程序()

服务器。中间件。append(logger _ mdw)

服务器。中间件。append(response _ mdw)

路线(服务器)

转轮=web .AppRunner服务器)

# 3 -

await runner.setup()

站点=web .钙钛石(转轮,' 127.0.0.1 ',8080)

await site.start()

print(f'======在http://{网站上提供host}:{site ._port}/======')

返回站点

异步定义服务器_容器(自身):

site=await self.run_server()

wait self.ft #等待约定的信号传过来

等待site.stop()

self.sg.send()

定义开始(自身):

自我。循环。run _ until _ complete(自身。服务器容器()

定义停止(自我):

自我。循环。call _ soon _ threadsafe(self。制成set _ result,无)

纽思级(qc .q线程):

v: sv

定义运行(自身):

loop=asyncio.new_event_loop()

ft=asyncio .未来(循环=循环)

self.v=sv(回路,英尺)

self.v.moveToThread(self)

尝试:

self.v.start()

e:除外

打印(五)

最后:

打印('服务器完成..')

定义停止(自我):

如果hasattr(self,' v ')且self.v不是无:

self.v。停止()

兆瓦级(qw .QDialog):

def __init__(self) - None:

超级()。__init__()

ly=qw .QHBoxLayout()

self.setLayout(ly)

self.b1=qw .q按钮("开始")

自我。B1。点击了。连接(自我。on _ B1 _ clicked)

ly.addWidget(self.b1)

self.resize(200,200)

self.th=None

def on_b1_clicked(self):

self.b1.setEnabled(False)

如果自我价值是无:

self.th=newth()

自我。th。开始了。连接(自我。B1 _更新)

自我。th。完成了。连接(自我。B1 _更新)

self.th.start()

else:

self.th.stop()

定义b1 _更新(自身):

如果自我价值为没有人或自我。这就完成了():

self.th=None

self.b1.setEnabled(真)

self.b1.setText('开始)

else:

self.b1.setEnabled(真)

self.b1.setText('停止)

def closeEvent(self,arg__1: qg .QCloseEvent) - None:

如果自我价值不是无:

self.on_b1_clicked()

arg__1.ignore()

else:

arg__1.accept()

if __name__=='__main__':

app=qw .QApplication()

ss=mw()

ss.show()

app.exec()

sys.exit()

Link to comment
Share on other sites