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-13 01:40:25 +08:00
parent 2bfbb8862b
commit 52fc7ca65d
11 changed files with 317 additions and 197 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,8 +2,6 @@ import wx
import threading
import time
from .Session_server import *
class LoginFrame(wx.Frame):
def __init__(self):
@ -29,8 +27,7 @@ class LoginFrame(wx.Frame):
# 创建状态栏
self.status_bar = self.CreateStatusBar(2)
self.server_status = False # 服务器状态
status_thread = threading.Thread(target=self.update_status_bar(None))
status_thread.daemon = True
status_thread = threading.Thread(target=self.update_status_bar(None), daemon=True)
status_thread.start()
# 启动定时器，每秒钟触发一次
@ -44,8 +41,19 @@ class LoginFrame(wx.Frame):
self.Bind(wx.EVT_MENU_OPEN, self.click_menu_title)
# 创建注册面板
self.panel = LoginPanel(self)
self.panel.Show()
self.sizer = wx.BoxSizer(wx.VERTICAL)
self.login_panel = self.LoginPanel(self)
self.register_panel = self.RegisterPanel(self)
self.sizer.Add(self.login_panel, 1, wx.EXPAND)
self.sizer.Add(self.register_panel, 1, wx.EXPAND)
self.login_panel.Hide()
self.register_panel.Hide()
self.login_panel.Show()
self.SetSizer(self.sizer)
def update_status_bar(self, event):
if self.server_status:
@ -59,137 +67,130 @@ class LoginFrame(wx.Frame):
menu = event.GetMenu()
menu_title = menu.GetTitle()
if menu_title == "登录":
self.panel.Destroy()
self.panel = LoginPanel(self)
self.panel.Show()
self.login_panel.Hide()
self.register_panel.Hide()
self.login_panel.Show()
self.Layout()
# 点击菜单检测
def click_menu(self, event):
menu_id = event.GetId()
if menu_id == self.register_id:
self.panel.Destroy()
self.panel = RegisterPanel(self)
self.panel.Show()
self.login_panel.Hide()
self.register_panel.Hide()
self.register_panel.Show()
self.Layout()
elif menu_id == self.set_server_id:
self.panel.Destroy()
self.panel = LoginPanel(self)
self.panel.Show()
self.login_panel.Hide()
self.register_panel.Hide()
self.Layout()
class LoginPanel(wx.Panel):
def __init__(self, parent):
super().__init__(parent)
# 主盒子
LoginPanel_main_box = wx.BoxSizer(wx.VERTICAL)
class LoginPanel(wx.Panel, Session_server):
def __init__(self, parent):
super().__init__(parent)
Session_server.__init__(self)
# 主盒子
main_box = wx.BoxSizer(wx.VERTICAL)
# 顶部盒子,目前只放了软件名
LoginPanel_top_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子
LoginPanel_show_title = wx.StaticText(self, label='登录')
LoginPanel_show_title.SetFont(wx.Font(30, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
LoginPanel_top_box.Add(LoginPanel_show_title, 1, wx.ALIGN_CENTER_HORIZONTAL | wx.ALL, 10)
LoginPanel_main_box.Add(LoginPanel_top_box, 0, wx.EXPAND)
# 顶部盒子,目前只放了软件名
top_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子
show_title = wx.StaticText(self, label='登录')
show_title.SetFont(wx.Font(30, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
top_box.Add(show_title, 1, wx.ALIGN_CENTER_HORIZONTAL | wx.ALL, 10)
main_box.Add(top_box, 0, wx.EXPAND)
# 中部盒子,放账号和密码
LoginPanel_content_box = wx.BoxSizer(wx.VERTICAL)
# 账号
LoginPanel_account_box = wx.BoxSizer(wx.HORIZONTAL)
LoginPanel_account_label = wx.StaticText(self, label=' 账号:')
LoginPanel_account_label.SetFont(
wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
LoginPanel_account_box.Add(LoginPanel_account_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.LoginPanel_account_text = wx.TextCtrl(self, size=(300, 30))
self.LoginPanel_account_text.SetHint('请输入账号')
LoginPanel_account_box.Add(self.LoginPanel_account_text, 1, wx.ALIGN_CENTER_VERTICAL)
LoginPanel_content_box.Add(LoginPanel_account_box)
# 添加空白控件来增加间距
LoginPanel_content_box.Add(wx.StaticText(self, label=''), 0, wx.ALL, 5)
# 密码
LoginPanel_password_box = wx.BoxSizer(wx.HORIZONTAL)
LoginPanel_password_label = wx.StaticText(self, label=' 密码:')
LoginPanel_password_label.SetFont(
wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
LoginPanel_password_box.Add(LoginPanel_password_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.LoginPanel_password_text = wx.TextCtrl(self, size=(300, 30))
self.LoginPanel_password_text.SetHint('请输入密码')
LoginPanel_password_box.Add(self.LoginPanel_password_text, 1, wx.ALIGN_CENTER_VERTICAL)
LoginPanel_content_box.Add(LoginPanel_password_box)
LoginPanel_main_box.Add(LoginPanel_content_box, 0, wx.EXPAND)
# 中部盒子,放账号和密码
content_box = wx.BoxSizer(wx.VERTICAL)
# 账号
account_box = wx.BoxSizer(wx.HORIZONTAL)
account_label = wx.StaticText(self, label=' 账号:')
account_label.SetFont(wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
account_box.Add(account_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.account_text = wx.TextCtrl(self, size=(300, 30))
self.account_text.SetHint('请输入账号')
account_box.Add(self.account_text, 1, wx.ALIGN_CENTER_VERTICAL)
content_box.Add(account_box)
# 添加空白控件来增加间距
content_box.Add(wx.StaticText(self, label=''), 0, wx.ALL, 5)
# 密码
password_box = wx.BoxSizer(wx.HORIZONTAL)
password_label = wx.StaticText(self, label=' 密码:')
password_label.SetFont(wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
password_box.Add(password_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.password_text = wx.TextCtrl(self, size=(300, 30))
self.password_text.SetHint('请输入密码')
password_box.Add(self.password_text, 1, wx.ALIGN_CENTER_VERTICAL)
content_box.Add(password_box)
main_box.Add(content_box, 0, wx.EXPAND)
# 底部盒子
LoginPanel_bottom_box = wx.BoxSizer(wx.VERTICAL)
# 登录按钮
LoginPanel_login_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子布局
self.LoginPanel_login_label = wx.Button(self, size=(200, 50), label='登录')
LoginPanel_login_box.AddStretchSpacer() # 添加一个可伸缩的空间，将登录按钮推到垂直中间
LoginPanel_login_box.Add(self.LoginPanel_login_label, 0, wx.ALIGN_CENTER_HORIZONTAL) # 将登录按钮添加到垂直盒子中
LoginPanel_login_box.AddStretchSpacer() # 再次添加一个可伸缩的空间，将登录按钮推到垂直中间
LoginPanel_bottom_box.Add(LoginPanel_login_box, 1,
wx.ALIGN_CENTER_VERTICAL | wx.ALIGN_CENTER_HORIZONTAL) # 将垂直盒子添加到底部盒子中，设置垂直和水平居中对齐
LoginPanel_main_box.Add(LoginPanel_bottom_box, 1, wx.EXPAND) # 将底部盒子添加到主盒子中，使其填满剩余空间
# 底部盒子
bottom_box = wx.BoxSizer(wx.VERTICAL)
# 登录按钮
login_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子布局
self.login_label = wx.Button(self, size=(200, 50), label='登录')
login_box.AddStretchSpacer() # 添加一个可伸缩的空间，将登录按钮推到垂直中间
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)
self.SetSizer(LoginPanel_main_box)
class RegisterPanel(wx.Panel):
def __init__(self, parent):
super().__init__(parent)
class RegisterPanel(wx.Panel):
def __init__(self, parent):
super().__init__(parent)
# 主盒子
main_box = wx.BoxSizer(wx.VERTICAL)
# 主盒子
RegisterPanel_main_box = wx.BoxSizer(wx.VERTICAL)
# 顶部盒子,目前只放了软件名
top_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子
show_title = wx.StaticText(self, label='注册')
show_title.SetFont(wx.Font(30, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
top_box.Add(show_title, 1, wx.ALIGN_CENTER_HORIZONTAL | wx.ALL, 10)
main_box.Add(top_box, 0, wx.EXPAND)
# 顶部盒子,目前只放了软件名
RegisterPanel_top_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子
RegisterPanel_show_title = wx.StaticText(self, label='注册')
RegisterPanel_show_title.SetFont(
wx.Font(30, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
RegisterPanel_top_box.Add(RegisterPanel_show_title, 1, wx.ALIGN_CENTER_HORIZONTAL | wx.ALL, 10)
RegisterPanel_main_box.Add(RegisterPanel_top_box, 0, wx.EXPAND)
# 中部盒子,放账号和密码
content_box = wx.BoxSizer(wx.VERTICAL)
# 账号
account_box = wx.BoxSizer(wx.HORIZONTAL)
account_label = wx.StaticText(self, label=' 用户名:')
account_label.SetFont(wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
account_box.Add(account_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.account_text = wx.TextCtrl(self, size=(300, 30))
self.account_text.SetHint('请输入用户名')
account_box.Add(self.account_text, 1, wx.ALIGN_CENTER_VERTICAL)
content_box.Add(account_box)
# 添加空白控件来增加间距
content_box.Add(wx.StaticText(self, label=''), 0, wx.ALL, 5)
# 密码
password_box = wx.BoxSizer(wx.HORIZONTAL)
password_label = wx.StaticText(self, label=' 密码: ')
password_label.SetFont(wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
password_box.Add(password_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.password_text = wx.TextCtrl(self, size=(300, 30))
self.password_text.SetHint('请输入密码')
password_box.Add(self.password_text, 1, wx.ALIGN_CENTER_VERTICAL)
content_box.Add(password_box)
main_box.Add(content_box, 0, wx.EXPAND)
# 中部盒子,放账号和密码
RegisterPanel_content_box = wx.BoxSizer(wx.VERTICAL)
# 账号
RegisterPanel_account_box = wx.BoxSizer(wx.HORIZONTAL)
RegisterPanel_account_label = wx.StaticText(self, label=' 用户名:')
RegisterPanel_account_label.SetFont(
wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
RegisterPanel_account_box.Add(RegisterPanel_account_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.RegisterPanel_account_text = wx.TextCtrl(self, size=(300, 30))
self.RegisterPanel_account_text.SetHint('请输入用户名')
RegisterPanel_account_box.Add(self.RegisterPanel_account_text, 1, wx.ALIGN_CENTER_VERTICAL)
RegisterPanel_content_box.Add(RegisterPanel_account_box)
# 添加空白控件来增加间距
RegisterPanel_content_box.Add(wx.StaticText(self, label=''), 0, wx.ALL, 5)
# 密码
RegisterPanel_password_box = wx.BoxSizer(wx.HORIZONTAL)
RegisterPanel_password_label = wx.StaticText(self, label=' 密码: ')
RegisterPanel_password_label.SetFont(
wx.Font(15, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL))
RegisterPanel_password_box.Add(RegisterPanel_password_label, 0, wx.ALIGN_CENTER_VERTICAL)
self.RegisterPanel_password_text = wx.TextCtrl(self, size=(300, 30))
self.RegisterPanel_password_text.SetHint('请输入密码')
RegisterPanel_password_box.Add(self.RegisterPanel_password_text, 1, wx.ALIGN_CENTER_VERTICAL)
RegisterPanel_content_box.Add(RegisterPanel_password_box)
RegisterPanel_main_box.Add(RegisterPanel_content_box, 0, wx.EXPAND)
# 底部盒子
bottom_box = wx.BoxSizer(wx.VERTICAL)
# 登录按钮
register_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子布局
login_label = wx.Button(self, size=(200, 50), label='确认')
register_box.AddStretchSpacer() # 添加一个可伸缩的空间，将登录按钮推到垂直中间
register_box.Add(login_label, 0, wx.ALIGN_CENTER_HORIZONTAL) # 将登录按钮添加到垂直盒子中
register_box.AddStretchSpacer() # 再次添加一个可伸缩的空间，将登录按钮推到垂直中间
bottom_box.Add(register_box, 1,
wx.ALIGN_CENTER_VERTICAL | wx.ALIGN_CENTER_HORIZONTAL) # 将垂直盒子添加到底部盒子中，设置垂直和水平居中对齐
main_box.Add(bottom_box, 1, wx.EXPAND) # 将底部盒子添加到主盒子中，使其填满剩余空间
# 底部盒子
RegisterPanel_bottom_box = wx.BoxSizer(wx.VERTICAL)
# 登录按钮
RegisterPanel_register_box = wx.BoxSizer(wx.VERTICAL) # 使用垂直盒子布局
self.RegisterPanel_login_label = wx.Button(self, size=(200, 50), label='确认')
RegisterPanel_register_box.AddStretchSpacer() # 添加一个可伸缩的空间，将登录按钮推到垂直中间
RegisterPanel_register_box.Add(self.RegisterPanel_login_label, 0,
wx.ALIGN_CENTER_HORIZONTAL) # 将登录按钮添加到垂直盒子中
RegisterPanel_register_box.AddStretchSpacer() # 再次添加一个可伸缩的空间，将登录按钮推到垂直中间
RegisterPanel_bottom_box.Add(RegisterPanel_register_box, 1,
wx.ALIGN_CENTER_VERTICAL | wx.ALIGN_CENTER_HORIZONTAL) # 将垂直盒子添加到底部盒子中，设置垂直和水平居中对齐
RegisterPanel_main_box.Add(RegisterPanel_bottom_box, 1, wx.EXPAND) # 将底部盒子添加到主盒子中，使其填满剩余空间
self.SetSizer(main_box)
self.SetSizer(RegisterPanel_main_box)

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

@ -6,9 +6,9 @@ import threading
class Session_server:
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.socker_ip = "127.0.0.1"
self.socker_port = 8000
self.server_socket = None
self.server_status = False # 服务器状态
self.link_server_Thread = threading.Thread(target=self.link_server)
@ -17,27 +17,39 @@ class Session_server:
self.receive_server_Thread = threading.Thread(target=self.receive_server)
def link_server(self):
while not self.server_status:
try:
while True:
if not self.server_status:
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
if self.server_socket is not None:
self.server_socket = None
try:
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.connect((self.socker_ip, self.socker_port))
self.server_status = True
if not self.link_server_Thread.is_alive():
print("Server")
self.receive_server_Thread.start()
except Exception as a:
self.server_status = False
print("连接错误:"+str(a))
def receive_server(self):
while self.server_status:
try:
receive_content_json = self.server_socket.recv(1024).decode('utf-8') # 追加接收到的数据
receive_content_json = self.server_socket.recv(10240).decode('utf-8') # 追加接收到的数据
receive_content = json.loads(receive_content_json)
except:
except Exception as a:
print("接收错误:"+str(a))
self.server_status = False
def send_server(self, genre, target, content):
if self.server_status:
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.send(data_json.encode("utf-8"))
try:
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.send(data_json.encode("utf-8"))
except Exception as a:
print("发送错误:"+str(a))
self.server_status = False

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

@ -7,6 +7,24 @@ class Client_main(LoginFrame, Session_server):
Session_server.__init__(self) # 与服务器会话
LoginFrame.__init__(self)
# 主页按钮绑定
self.Bind(wx.EVT_BUTTON, self.send_login_button, self.login_panel.LoginPanel_login_label)
self.Bind(wx.EVT_BUTTON, self.send_register_button, self.register_panel.RegisterPanel_login_label)
def send_login_button(self, event):
if self.server_status:
account = self.login_panel.LoginPanel_account_text.GetValue().strip()
password = self.login_panel.LoginPanel_password_text.GetValue().strip()
content = {'account': account, 'password': password}
target = "服务器"
genre = "登录"
self.send_server(genre=genre, target=target, content=content)
def send_register_button(self, event):
if self.server_status:
account = self.login_panel.RegisterPanel_account_text.GetValue().strip()
password = self.login_panel.RegisterPanel_password_text.GetValue().strip()
content = {'account': account, 'password': password}
target = "服务器"
genre = "注册"
self.send_server(genre=genre, target=target, content=content)

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.

59
python/test/Server/Database_formula.py Normal file
View File

@ -0,0 +1,59 @@
import pymssql
import threading
import time
class database:
def __init__(self):
self.database_server = 'localhost'
self.database_user = 'Chat'
self.database_password = '123456'
self.database_port = 8972
self.database_name = 'Chat'
self.database_state = False
self.database_conn = None
self.database_cursor = None
self.link_database_thread = threading.Thread(target=self.link_database)
self.link_database_thread.start()
def link_database(self):
while not self.database_state:
time.sleep(1)
try:
self.database_conn = pymssql.connect(server=self.database_server,
user=self.database_user,
password=self.database_password,
port=self.database_port,
database=self.database_name)
self.database_cursor = self.database_conn.cursor()
self.database_state = True
except pymssql.Error as a:
print(a)
def check_account_state(self, Id: int, password):
if self.database_state:
try:
self.database_cursor.execute(f"select Id,Password from Account where Id = {Id}")
request_data = self.database_cursor.fetchall()
if len(request_data) != 0:
if request_data[0][1] == password:
return 0
else:
return 1
else:
return 2
except pymssql as a:
self.database_state = False
print(a)
def sign_account(self, Id: int, password):
if self.database_state:
try:
self.database_cursor.execute(f"select Id,Password from Account where Id = {Id}")
return 0
except pymssql as a:
self.database_state = False
print(a)
return 1

40
python/test/Server/Session_client.py
View File

@ -1,4 +1,3 @@
import string
import time
import socket
import random
@ -7,7 +6,7 @@ import threading
class link_client:
def __init__(self ):
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)
@ -22,8 +21,37 @@ class link_client:
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)
recv_client_Thread = threading.Thread(target=self.recv_client(client_socket), daemon=True)
recv_client_Thread.start()
except:
print("无")
pass
def send_client(self, client_socket, genre, target, content):
if client_socket in self.client_socket_dict:
try:
data = {"genre": genre, "target": target, "data": content,
"datetime": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
data_json = json.dumps(data)
client_socket.send(data_json.encode("utf-8"))
except Exception as a:
print("发送错误:" + str(a))
def pick_data(self, data):
pass
def recv_client(self, client_socket):
while True:
try:
data_json = client_socket.recv(1024).decode('utf-8')
data = json.loads(data_json)
print(data_json["genre"])
if data_json["genre"] == '登录':
content = self.pick_data(data)
print(content['state'])
if content['state'] == 0:
self.client_socket_dict[content['user']] = client_socket
print(self.client_socket_dict[content['user']])
else:
self.pick_data(data)
except:
client_socket.close()

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

@ -1,7 +1,16 @@
from .Session_client import *
from .Database_formula import *
class Server_main(link_client):
class Server_main(link_client, database):
def __init__(self):
link_client.__init__(self)
database.__init__(self)
def pick_data(self, data):
account_state = None
if data['genre'] == '登录':
account_state = self.check_account_state(data['data']['account'], data['data']['password'])
return {"state": account_state, "user": data['data']['account']}
elif data['genre'] == '注册':
pass

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() # 这里填写你的 ip_socker 和 port_socker
frame = Client_main()
frame.Show()
app.MainLoop()

97
python/test/test.py
View File

@ -1,59 +1,52 @@
import string
import threading
import time
import socket
import random
import json
import concurrent.futures
import wx
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 # 服务器状态
class MultiPanelFrame(wx.Frame):
def __init__(self):
super().__init__(None, title="Multi Panel Example", size=(400, 300))
# 创建线程池
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
self.panel1 = wx.Panel(self)
self.panel2 = wx.Panel(self)
self.panel3 = wx.Panel(self)
# 启动连接服务器的任务
self.executor.submit(self.Link_server)
self.panel1.SetBackgroundColour(wx.Colour(255, 0, 0)) # Red
self.panel2.SetBackgroundColour(wx.Colour(0, 255, 0)) # Green
self.panel3.SetBackgroundColour(wx.Colour(0, 0, 255)) # Blue
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:
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:
self.server_status = False
except:
self.server_status = False
time.sleep(60)
self.sizer = wx.BoxSizer(wx.VERTICAL)
self.sizer.Add(self.panel1, 1, wx.EXPAND)
self.sizer.Add(self.panel2, 1, wx.EXPAND)
self.sizer.Add(self.panel3, 1, wx.EXPAND)
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)
self.panel1.Show()
self.panel2.Hide()
self.panel3.Hide()
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))
self.SetSizer(self.sizer)
self.Bind(wx.EVT_CHAR_HOOK, self.on_key_press)
def on_key_press(self, event):
key = event.GetKeyCode()
if key == ord('1'):
self.panel1.Show()
self.panel2.Hide()
self.panel3.Hide()
self.Layout()
elif key == ord('2'):
self.panel1.Hide()
self.panel2.Show()
self.panel3.Hide()
self.Layout()
elif key == ord('3'):
self.panel1.Hide()
self.panel2.Hide()
self.panel3.Show()
self.Layout()
if __name__ == "__main__":
app = wx.App()
frame = MultiPanelFrame()
frame.Show()
app.MainLoop()