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 13:52:36 +08:00
parent 52fc7ca65d
commit c8e0af64ac
10 changed files with 106 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -54,7 +54,6 @@ class LoginFrame(wx.Frame):
self.SetSizer(self.sizer)
def update_status_bar(self, event):
if self.server_status:
self.status_bar.SetStatusText("服务器\t已连接", 0)
@ -138,7 +137,6 @@ class LoginFrame(wx.Frame):
self.SetSizer(LoginPanel_main_box)
class RegisterPanel(wx.Panel):
def __init__(self, parent):
super().__init__(parent)

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

@ -26,18 +26,23 @@ class Session_server:
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")
if not self.receive_server_Thread.is_alive():
self.receive_server_Thread.start()
except Exception as a:
self.server_status = False
print("连接错误:" + str(a))
def login_page_receive(self, receive_content):
pass
def receive_server(self):
while self.server_status:
try:
receive_content_json = self.server_socket.recv(10240).decode('utf-8') # 追加接收到的数据
receive_content_json = self.server_socket.recv(1024).decode('utf-8')
receive_content = json.loads(receive_content_json)
if receive_content["genre"] in ['注册', '登录']:
self.login_page_receive(receive_content)
print(receive_content)
except Exception as a:
print("接收错误:" + str(a))
self.server_status = False
@ -52,4 +57,3 @@ class Session_server:
except Exception as a:
print("发送错误:" + str(a))
self.server_status = False

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

@ -22,9 +22,22 @@ class Client_main(LoginFrame, Session_server):
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()
account = self.register_panel.RegisterPanel_account_text.GetValue().strip()
password = self.register_panel.RegisterPanel_password_text.GetValue().strip()
content = {'account': account, 'password': password}
target = "服务器"
genre = "注册"
self.send_server(genre=genre, target=target, content=content)
def login_page_receive(self, receive_content):
if receive_content["genre"] == '登录':
if receive_content["data"] == 0:
wx.MessageBox('登录成功', '登录', wx.OK | wx.ICON_INFORMATION)
elif receive_content["data"] == -1:
wx.MessageBox('重复登录', '登录', wx.OK | wx.ICON_INFORMATION)
elif receive_content["data"] == 1:
wx.MessageBox('密码错误', '登录', wx.OK | wx.ICON_INFORMATION)
elif receive_content["data"] == 2:
wx.MessageBox('未找到该账号', '登录', wx.OK | wx.ICON_INFORMATION)
elif receive_content["genre"] == '注册':
wx.MessageBox(f'注册成功\n网名 : {receive_content["data"]['NetName']} \n账号 : {receive_content["data"]['Id']} \n密码 : {receive_content["data"]['Password']}', '注册', wx.OK | wx.ICON_INFORMATION)

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/Database_formula.py
View File

@ -17,6 +17,9 @@ class database:
self.link_database_thread = threading.Thread(target=self.link_database)
self.link_database_thread.start()
# 创建一个互斥锁对象
self.lock = threading.Lock()
def link_database(self):
while not self.database_state:
time.sleep(1)
@ -47,13 +50,29 @@ class database:
self.database_state = False
print(a)
def sign_account(self, Id: int, password):
def sign_account(self, NetName, Password):
if self.database_state:
# 获取锁
self.lock.acquire()
try:
self.database_cursor.execute("insert into Account(NetName, Password)"
f"values ('{NetName}','{Password}')")
self.database_conn.commit()
self.database_cursor.execute(f"SELECT SCOPE_IDENTITY() AS LastInsertedId;")
Id = self.database_cursor.fetchone()[0]
# 释放锁
self.lock.release()
return {"Id": int(Id), "NetName": NetName, "Password": Password}
except pymssql as a:
self.lock.release()
self.database_state = False
return False
def alter_state_database(self, Id: int, sate):
if self.database_state:
try:
self.database_cursor.execute(f"select Id,Password from Account where Id = {Id}")
return 0
self.database_cursor.execute(f"update Account set State = N'{sate}' where Id = {Id}")
self.database_conn.commit()
except pymssql as a:
self.database_state = False
print(a)
return 1

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

@ -1,6 +1,5 @@
import time
import socket
import random
import json
import threading
@ -27,31 +26,43 @@ class link_client:
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))
except:
client_socket.close()
def pick_data(self, data):
pass
def recv_client(self, client_socket):
while True:
state = True
while state:
try:
data_json = client_socket.recv(1024).decode('utf-8')
data = json.loads(data_json)
print(data_json["genre"])
if data_json["genre"] == '登录':
if data["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']])
if data['data']['account'] not in self.client_socket_dict:
if content == 0:
self.client_socket_dict[data['data']['account']] = client_socket
self.alter_state(data['data']['account'], "在线")
self.send_client(client_socket, data["genre"], '客户端', content)
else:
self.send_client(client_socket, data["genre"], '客户端', -1)
elif data["genre"] == '注册':
content = self.pick_data(data)
self.send_client(client_socket, data["genre"], '客户端', content)
else:
self.pick_data(data)
except:
try:
client_socket.close()
finally:
state = False
self.alter_state(client_socket, "离线")
def alter_state(self, client_socket, state):
pass

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

@ -11,6 +11,14 @@ class Server_main(link_client, database):
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']}
return account_state
elif data['genre'] == '注册':
pass
account_data = self.sign_account(data['data']['account'], data['data']['password'])
return account_data
def alter_state(self, client_socket, state):
for key, value in self.client_socket_dict.items():
if value == client_socket:
del self.client_socket_dict[key]
break
self.alter_state_database(client_socket, state)

68
python/test/test.py
View File

@ -1,52 +1,26 @@
import wx
class MultiPanelFrame(wx.Frame):
def __init__(self):
super().__init__(None, title="Multi Panel Example", size=(400, 300))
self.panel1 = wx.Panel(self)
self.panel2 = wx.Panel(self)
self.panel3 = wx.Panel(self)
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
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)
self.panel1.Show()
self.panel2.Hide()
self.panel3.Hide()
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
frame = wx.Frame(None, -1, '多行文本弹窗示例', size=(300, 200))
frame.Center()
# 创建一个Panel
panel = wx.Panel(frame, -1)
# 创建一个按钮
button = wx.Button(panel, -1, '点击弹窗', pos=(100, 50))
# 定义按钮的点击事件处理函数
def on_button_click(event):
message = "这是一个多行文本的弹窗。\n第二行文本。\n第三行文本。"
dlg = wx.MessageDialog(frame, message, '多行文本提示', wx.OK | wx.ICON_INFORMATION)
dlg.ShowModal()
dlg.Destroy()
# 绑定按钮的点击事件
button.Bind(wx.EVT_BUTTON, on_button_click)
frame.Show()
app.MainLoop()