lsy/practice_code
0
0
Fork 0
Code Pull Requests Actions Packages Projects Releases Wiki Activity

状态栏没有二次修改,无法发送

Browse Source
This commit is contained in:
lsy2246 2024-04-12 00:56:58 +08:00
parent d48d38851b
commit 2bfbb8862b
11 changed files with 141 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
python/test/Cilent/Login.py
View File

@ -1,7 +1,8 @@
import wx
import threading
import time
from tkinter import messagebox
from .Session_server import *
class LoginFrame(wx.Frame):
@ -78,9 +79,10 @@ class LoginFrame(wx.Frame):
self.Layout()
class LoginPanel(wx.Panel):
class LoginPanel(wx.Panel, Session_server):
def __init__(self, parent):
super().__init__(parent)
Session_server.__init__(self)
# 主盒子
main_box = wx.BoxSizer(wx.VERTICAL)
@ -119,16 +121,25 @@ class LoginPanel(wx.Panel):
bottom_box = wx.BoxSizer(wx.VERTICAL)
# 登录按钮
login_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子布局
login_label = wx.Button(self, size=(200, 50), label='登录')
self.login_label = wx.Button(self, size=(200, 50), label='登录')
login_box.AddStretchSpacer() # 添加一个可伸缩的空间，将登录按钮推到垂直中间
login_box.Add(login_label, 0, wx.ALIGN_CENTER_HORIZONTAL) # 将登录按钮添加到垂直盒子中
login_box.Add(self.login_label, 0, wx.ALIGN_CENTER_HORIZONTAL) # 将登录按钮添加到垂直盒子中
login_box.AddStretchSpacer() # 再次添加一个可伸缩的空间，将登录按钮推到垂直中间
bottom_box.Add(login_box, 1, wx.ALIGN_CENTER_VERTICAL | wx.ALIGN_CENTER_HORIZONTAL) # 将垂直盒子添加到底部盒子中，设置垂直和水平居中对齐
main_box.Add(bottom_box, 1, wx.EXPAND) # 将底部盒子添加到主盒子中，使其填满剩余空间
# 按钮绑定
self.Bind(wx.EVT_BUTTON, self.send_login_button, self.login_label)
self.SetSizer(main_box)
def send_login_button(self, event):
account = self.account_text.GetValue().strip()
password = self.password_text.GetValue().strip()
content = {'account': account, 'password': password}
target = "服务器"
genre = "登录"
self.send_server(genre=genre, target=target, content=content)
class RegisterPanel(wx.Panel):

56
python/test/Cilent/Session_server.py
View File

@ -1,47 +1,43 @@
import string
import time
import socket
from threading import Thread
import random
import json
import threading
class Session_server:
def __init__(self, ip_socker, port_socker):
self.ip_socker = ip_socker
self.port_socker = port_socker
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
def __init__(self):
self.ip_socker = "127.0.0.1"
self.port_socker = 8000
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_status = False # 服务器状态
def Link_server(self):
Verification_code = ""
for _ in range(20):
if random.choice([True, False]):
Verification_code += str(random.randint(0, 9))
else:
Verification_code += random.choice(string.ascii_letters)
data = {"genre": "test", "source": "Link_server", "data": Verification_code,
"datetime": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
data_json = json.dumps(data)
self.server_socket.sendto(data_json.encode("utf-8"), (self.ip_socker, self.port_socker))
receive_content_json = self.server_socket.recv(1024).decode('utf-8')
receive_content = json.loads(receive_content_json)
if receive_content["data"] == Verification_code:
self.server_status = True # 服务器状态
Receive_server_Thread = Thread(target=self.Receive_server)
Receive_server_Thread.daemon = True
Receive_server_Thread.start()
self.link_server_Thread = threading.Thread(target=self.link_server)
self.link_server_Thread.start()
self.receive_server_Thread = threading.Thread(target=self.receive_server)
def Receive_server(self):
def link_server(self):
while not self.server_status:
try:
time.sleep(1)
self.server_socket.connect((self.ip_socker, self.port_socker))
self.server_status = True
if not self.link_server_Thread.is_alive():
self.receive_server_Thread.start()
except:
self.server_status = False
def receive_server(self):
while self.server_status:
try:
receive_content_json = self.server_socket.recv(1024).decode('utf-8') # 追加接收到的数据
receive_content = json.loads(receive_content_json)
print(receive_content)
except:
self.server_status = False
def Send_server(self, genre, source, content):
def send_server(self, genre, target, content):
if self.server_status:
data = {"genre": genre, "source": source, "data": content,
data = {"genre": genre, "target": target, "data": content,
"datetime": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
data_json = json.dumps(data)
self.server_socket.sendto(data_json.encode("utf-8"), (self.ip_socker, self.port_socker))
self.server_socket.send(data_json.encode("utf-8"))

11
python/test/Cilent/__init__.py
View File

@ -3,13 +3,10 @@ from .Session_server import *
class Client_main(LoginFrame, Session_server):
def __init__(self, ip_socker, port_socker):
Session_server.__init__(self, ip_socker, port_socker)
def __init__(self):
Session_server.__init__(self) # 与服务器会话
LoginFrame.__init__(self)
Session_server.Link_server(self)
# 主页按键绑定
def Sign_Account(self, event):
if self.GetParent().server_status:
pass

BIN
python/test/Cilent/__pycache__/Login.cpython-312.pyc
View File

Binary file not shown.

BIN
python/test/Cilent/__pycache__/Session_server.cpython-312.pyc
View File

Binary file not shown.

BIN
python/test/Cilent/__pycache__/__init__.cpython-312.pyc
View File

Binary file not shown.

29
python/test/Server/Session_client.py Normal file
View File

@ -0,0 +1,29 @@
import string
import time
import socket
import random
import json
import threading
class link_client:
def __init__(self ):
self.server_host = "127.0.0.1"
self.server_port = 8000
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.bind((self.server_host, self.server_port))
self.server_socket.listen()
self.client_socket_dict = {}
self.link_client_Thread = threading.Thread(target=self.link_client)
self.link_client_Thread.start()
def link_client(self):
while True:
try:
client_socket, client_address = self.server_socket.accept()
data_json = self.server_socket.recv(1024).decode('utf-8')
data = json.loads(data_json)
print(data)
except:
print("无")

7
python/test/Server/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .Session_client import *
class Server_main(link_client):
def __init__(self):
link_client.__init__(self)

2
python/test/StartClient.py
View File

@ -3,6 +3,6 @@ from Cilent import Client_main
if __name__ == '__main__':
app = wx.App()
frame = Client_main("127.0.0.1", 8000) # 这里填写你的 ip_socker 和 port_socker
frame = Client_main() # 这里填写你的 ip_socker 和 port_socker
frame.Show()
app.MainLoop()

4
python/test/StartServer.py Normal file
View File

@ -0,0 +1,4 @@
from Server import Server_main
if __name__ == '__main__':
server = Server_main()

87
python/test/test.py
View File

@ -1,46 +1,59 @@
import tkinter as tk
from tkinter import messagebox
import string
import threading
import time
import socket
import random
import json
import concurrent.futures
# 创建主窗口
root = tk.Tk()
root.withdraw() # 隐藏主窗口
# 弹出消息框
messagebox.showinfo("标题", "消息内容")
class Session_server:
def __init__(self, ip_socker, port_socker):
self.ip_socker = ip_socker
self.port_socker = port_socker
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.server_status = False # 服务器状态
# 弹出警告框
messagebox.showwarning("警告", "警告内容")
# 创建线程池
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
# 弹出错误框
messagebox.showerror("错误", "错误内容")
# 启动连接服务器的任务
self.executor.submit(self.Link_server)
# 弹出询问框
result = messagebox.askquestion("询问", "是否继续？")
if result == 'yes':
print("用户选择了继续")
def Link_server(self):
while True:
Verification_code = ""
for _ in range(20):
if random.choice([True, False]):
Verification_code += str(random.randint(0, 9))
else:
print("用户选择了取消")
# 弹出确认框
result = messagebox.askokcancel("确认", "是否继续？")
if result:
print("用户选择了继续")
Verification_code += random.choice(string.ascii_letters)
data = {"genre": "test", "source": "Link_server", "data": Verification_code,
"datetime": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
data_json = json.dumps(data)
self.server_socket.sendto(data_json.encode("utf-8"), (self.ip_socker, self.port_socker))
try:
receive_content_json = self.server_socket.recv(1024).decode('utf-8')
receive_content = json.loads(receive_content_json)
if receive_content["data"] == Verification_code:
self.server_status = True # 服务器状态
# 启动接收服务器响应的任务
self.executor.submit(self.Receive_server)
else:
print("用户选择了取消")
self.server_status = False
except:
self.server_status = False
time.sleep(60)
# 弹出是/否框
result = messagebox.askyesno("是/否", "是否继续？")
if result:
print("用户选择了是")
else:
print("用户选择了否")
def Receive_server(self):
while self.server_status:
receive_content_json = self.server_socket.recv(1024).decode('utf-8') # 追加接收到的数据
receive_content = json.loads(receive_content_json)
print(receive_content)
# 弹出重试/取消框
result = messagebox.askretrycancel("重试/取消", "是否继续？")
if result:
print("用户选择了重试")
else:
print("用户选择了取消")
# 显示主窗口
root.mainloop()
def Send_server(self, genre, source, content):
if self.server_status:
data = {"genre": genre, "source": source, "data": content,
"datetime": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
data_json = json.dumps(data)
self.server_socket.sendto(data_json.encode("utf-8"), (self.ip_socker, self.port_socker))