TCP聊天+传输文件服务器服务器套接字v2.6
更改的地方:
主要是客户端界面更改
v1.8
的改进后输入用户名, 到了主界面的时候才能输入密码, 但现在是直接输入用户名和密码, 然后转为json, 等待回复)commandLink
)checkBox
, 如果记住报存到config.json
, 下次直接转入登录, 设置为默认)不符合规范的json数据包
, 所以在服务端在识别用户是否存在
+ 账户密码是否正确
的前提下 又增加了检测用户名密码长度
的规则)所有版本记录:
v1.0
: TCP聊天服务器套接字|PyQt5+socket(TCP端口映射+端口放行)+logging+Thread(含日志,html)+anaconda打包32位exe(3.4万字)|python高阶v1.1
: python TCP套接字服务器v1.1-新增服务端命令功能及修改bug(socket+PyQt5)v1.2
: python TCP服务器v1.2 - 服务端新增用户登录注册(json, md5加密)v1.3
: python TCP服务器v1.3 - 服务器抗压测试及关闭套接字处理v1.4
: python TCP服务器v1.4 - 客户端连接服务器异常(异常情况分类)处理v1.5
: PyQt5可编辑下拉框(comboBox):editable - python TCP服务器v1.5 - 客户端连接界面增加自定义参数(设置超时, 连接地址可选)v1.6
: Python TCP服务器v1.6 - multiprocessing多进程及Ctrl-c(SIGINT)退出v1.7
: Python TCP服务器v1.7 - PyQt5 server服务端来临v1.8
: python TCP服务器v1.8 - PyQt5登录界面美化+淡入淡出v1.9
: socketTCP协程文件+信息传递 - TCP聊天文件服务器v1.9 - 划时代的版本更新(4.6万字)v2.0
: TCP聊天文件服务器v2.0 - 重大bug修复+PyQt5文件传输可视化v2.1
: TCP聊天文件服务器v2.1 - 服务端线程管理(threading.enumerate)v2.2
: TCP聊天文件服务器v2.2 - 服务端客户端套接字解决分包/粘包问题 - SocketQueue继承以及减少冗余v2.3
: gzip的使用 - TCP聊天文件服务器v2.3 - 文件传输建立缓存制度和.gz的解压缩/压缩解决运行内存过大v2.4
: 网络传输测速 - TCP聊天+传输文件服务器服务器套接字v2.4 - socket协程文件传送测速v2.5
: TCP聊天+传输文件服务器服务器套接字v2.5 - socket测速规范已经gzip的弃用
loading界面参见 我的csdn博客 - pyqt5 等待界面 (QMovie 加载 gif)
class Socket(SocketClient):
def __init__(self, Function=lambda i: None, code='utf-8'):
super(Socket, self).__init__(codec=code)
self.handler = message_handle(self.send)
def parse_argument(self, arg: str) -> str:
return self.handler.handle(arg.strip())
def receive_text(self):
return super(Socket, self).recv()
def recv(self):
result = super(Socket, self).recv()
if isinstance(result, str) and result:
self._traceback(f'{result} <font size=1>{convert(len(result))}</font>')
return self.isOpen()
def forever_receive(self) -> None:
self.handler.send_text(self.header)
while True:
if not self.recv():
return
def Check_info(self, *args, **kwargs):
threading(True, target = lambda: self.__Check_info(*args, **kwargs))
def __Check_info(self, info, emit, return_func, err_func):
if not self.is_connect():
res, trace = self.connect()
emit(bool(res))
if not res:
return err_func(trace)
self.handler.send_text(json.dumps(info))
emit(True)
data = self.receive_text()
emit(True)
try:
data = json.loads(data)
except (ValueError, KeyError):
return err_func("解析异常!")
else:
emit(True)
return_func(data)
def run(self): # 线程
threading(True, target=self.forever_receive)
def quitEvent(self):
self.__is_connect = False
if main.is_setup:
main.ConnectionError_signal.emit()
LOGIN_INFO_FILE = "config.json"
LOGIN_INFO = {"type": 0,
"username": "",
"password": ""}
if os.path.isfile(LOGIN_INFO_FILE):
with open(LOGIN_INFO_FILE, "r") as f:
LOGIN_INFO = json.load(f)
def json_dump():
with open(LOGIN_INFO_FILE, "w") as f:
json.dump(LOGIN_INFO, f, indent=4)
def Animation(parent, type=b"windowOpacity", from_value=0, to_value=1, ms=1000, connect=None):
anim = QPropertyAnimation(parent, type)
anim.setDuration(ms)
anim.setStartValue(from_value)
anim.setEndValue(to_value)
if connect:
anim.finished.connect(connect)
anim.start()
return anim
class LoadingProgress(QtWidgets.QDialog):
update_signal = QtCore.pyqtSignal(bool)
def __init__(self, parent=None):
super(LoadingProgress, self).__init__(parent)
self.value = 0
self.update_signal.connect(self.update_progress)
vbox = QtWidgets.QVBoxLayout(self)
self.steps = [f"连接服务器中({TIMEOUT}s)...",
"发送数据中...",
"接收数据中...",
"解析数据中..."]
self.movie_label = QtWidgets.QLabel()
self.movie = QtGui.QMovie("images/loading.gif")
self.movie_label.setMovie(self.movie)
self.movie.start()
self.progress_label = QtWidgets.QLabel()
self.label_update()
vbox.addWidget(self.movie_label)
vbox.addWidget(self.progress_label)
self.setLayout(vbox)
# self.exec_()
def label_update(self):
self.progress_label.setText(self.steps[self.value])
def update_progress(self, boolean: bool) -> None:
self.value += 1
if boolean and self.value < len(self.steps):
self.label_update()
else:
self.close()
class User_Setup(QtWidgets.QDialog):
end_anim, start_anim = None, None
log_progress_signal = QtCore.pyqtSignal(str)
reg_progress_signal = QtCore.pyqtSignal(str)
handle_signal = QtCore.pyqtSignal(dict)
err_signal = QtCore.pyqtSignal(str)
loading_dialog = None
successful = False
def closeEvent(self, event: QtGui.QCloseEvent) -> None:
if not self.end_anim:
self.end_anim = Animation(self, ms=2000, from_value=1, to_value=0, connect=self.close)
event.ignore()
else:
if self.successful:
main.SetupUi()
def __init__(self, parent=None):
super().__init__(parent)
self.start_anim = Animation(self, ms=2000)
self.setObjectName("Dialog")
self.resize(362, 394)
font = QtGui.QFont()
font.setFamily("Consolas")
font.setPointSize(12)
self.setFont(font)
icon = QtGui.QIcon()
icon.addPixmap(QtGui.QPixmap("images/user.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.setWindowIcon(icon)
self.setAutoFillBackground(True)
self.gridLayout = QtWidgets.QGridLayout(self)
self.gridLayout.setObjectName("gridLayout")
self.label = QtWidgets.QLabel(self)
self.label.setFocusPolicy(QtCore.Qt.StrongFocus)
self.label.setStyleSheet("")
self.label.setText("")
self.label.setPixmap(QtGui.QPixmap("images/zmh.png"))
self.label.setObjectName("label")
self.gridLayout.addWidget(self.label, 0, 0, 1, 1)
self.tabWidget = QtWidgets.QTabWidget(self)
self.tabWidget.setAccessibleName("")
self.tabWidget.setAutoFillBackground(True)
self.tabWidget.setObjectName("tabWidget")
self.log = QtWidgets.QWidget()
self.log.setObjectName("login")
self.gridLayout_2 = QtWidgets.QGridLayout(self.log)
self.gridLayout_2.setObjectName("gridLayout_2")
self.lineEdit_5 = QtWidgets.QLineEdit(self.log)
self.lineEdit_5.setStyleSheet(" QLineEdit\n"
" {border:0px;\n"
" border-radius:0;\n"
" font: 12pt \"Consolas\";\n"
" margin:15px;\n"
" border-bottom: 2px solid #B3B3B3;}\n"
" QLineEdit:hover{\n"
" border-bottom:3px solid #66A3FF;\n"
" }\n"
" QLineEdit:focus{\n"
" border-bottom:3px solid #E680BD\n"
" }")
self.lineEdit_5.setText("")
self.lineEdit_5.setEchoMode(QtWidgets.QLineEdit.Password)
self.lineEdit_5.setClearButtonEnabled(True)
self.lineEdit_5.setObjectName("lineEdit_5")
self.gridLayout_2.addWidget(self.lineEdit_5, 2, 0, 1, 2)
self.lineEdit_4 = QtWidgets.QLineEdit(self.log)
self.lineEdit_4.setStyleSheet(" QLineEdit\n"
" {border:0px;\n"
" border-radius:0;\n"
" font: 12pt \"Consolas\";\n"
" margin:15px;\n"
" border-bottom: 2px solid #B3B3B3;}\n"
" QLineEdit:hover{\n"
" border-bottom:3px solid #66A3FF;\n"
" }\n"
" QLineEdit:focus{\n"
" border-bottom:3px solid #E680BD\n"
" }")
self.lineEdit_4.setClearButtonEnabled(True)
self.lineEdit_4.setObjectName("lineEdit_4")
self.gridLayout_2.addWidget(self.lineEdit_4, 1, 0, 1, 2)
self.pushButton_2 = QtWidgets.QPushButton(self.log)
icon1 = QtGui.QIcon()
icon1.addPixmap(QtGui.QPixmap("images/login.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.pushButton_2.setIcon(icon1)
self.pushButton_2.setObjectName("pushButton_2")
self.gridLayout_2.addWidget(self.pushButton_2, 6, 1, 1, 1)
self.checkBox_2 = QtWidgets.QCheckBox(self.log)
self.checkBox_2.setChecked(True)
self.checkBox_2.setObjectName("checkBox_2")
self.gridLayout_2.addWidget(self.checkBox_2, 6, 0, 1, 1)
self.label_3 = QtWidgets.QLabel(self.log)
font = QtGui.QFont()
font.setFamily("Comic Sans MS")
font.setPointSize(11)
font.setBold(False)
font.setItalic(False)
font.setWeight(50)
self.label_3.setFont(font)
self.label_3.setStyleSheet("color: rgb(255, 1, 39);\n"
"font: 11pt \"宋体\";")
self.label_3.setObjectName("label_3")
self.gridLayout_2.addWidget(self.label_3, 4, 0, 1, 2)
self.commandLinkButton = QtWidgets.QCommandLinkButton(self.log)
font = QtGui.QFont()
font.setFamily("Segoe UI")
font.setPointSize(12)
self.commandLinkButton.setFont(font)
self.commandLinkButton.setObjectName("commandLinkButton")
self.gridLayout_2.addWidget(self.commandLinkButton, 3, 0, 1, 2)
self.tabWidget.addTab(self.log, "")
self.reg = QtWidgets.QWidget()
self.reg.setObjectName("reg")
self.gridLayout_3 = QtWidgets.QGridLayout(self.reg)
self.gridLayout_3.setObjectName("gridLayout_3")
self.pushButton = QtWidgets.QPushButton(self.reg)
icon2 = QtGui.QIcon()
icon2.addPixmap(QtGui.QPixmap("images/register.png"), QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.pushButton.setIcon(icon2)
self.pushButton.setObjectName("pushButton")
self.gridLayout_3.addWidget(self.pushButton, 5, 1, 1, 1)
self.checkBox = QtWidgets.QCheckBox(self.reg)
self.checkBox.setChecked(True)
self.checkBox.setTristate(False)
self.checkBox.setObjectName("checkBox")
self.gridLayout_3.addWidget(self.checkBox, 5, 0, 1, 1)
self.lineEdit_3 = QtWidgets.QLineEdit(self.reg)
self.lineEdit_3.setContextMenuPolicy(QtCore.Qt.DefaultContextMenu)
self.lineEdit_3.setAccessibleName("")
self.lineEdit_3.setStyleSheet(" QLineEdit\n"
" {border:0px;\n"
" border-radius:0;\n"
" font: 12pt \"Consolas\";\n"
" margin:15px;\n"
" border-bottom: 2px solid #B3B3B3;}\n"
" QLineEdit:hover{\n"
" border-bottom:3px solid #66A3FF;\n"
" }\n"
" QLineEdit:focus{\n"
" border-bottom:3px solid #E680BD\n"
" }")
self.lineEdit_3.setEchoMode(QtWidgets.QLineEdit.Password)
self.lineEdit_3.setDragEnabled(False)
self.lineEdit_3.setClearButtonEnabled(True)
self.lineEdit_3.setObjectName("lineEdit_3")
self.gridLayout_3.addWidget(self.lineEdit_3, 2, 0, 1, 2)
self.lineEdit_2 = QtWidgets.QLineEdit(self.reg)
font = QtGui.QFont()
font.setFamily("Consolas")
font.setPointSize(12)
font.setBold(False)
font.setItalic(False)
font.setWeight(50)
self.lineEdit_2.setFont(font)
self.lineEdit_2.setStyleSheet(" QLineEdit\n"
" {border:0px;\n"
" border-radius:0;\n"
" font: 12pt \"Consolas\";\n"
" margin:15px;\n"
" border-bottom: 2px solid #B3B3B3;}\n"
" QLineEdit:hover{\n"
" border-bottom:3px solid #66A3FF;\n"
" }\n"
" QLineEdit:focus{\n"
" border-bottom:3px solid #E680BD\n"
" }")
self.lineEdit_2.setInputMask("")
self.lineEdit_2.setEchoMode(QtWidgets.QLineEdit.Password)
self.lineEdit_2.setDragEnabled(False)
self.lineEdit_2.setCursorMoveStyle(QtCore.Qt.LogicalMoveStyle)
self.lineEdit_2.setClearButtonEnabled(True)
self.lineEdit_2.setObjectName("lineEdit_2")
self.gridLayout_3.addWidget(self.lineEdit_2, 1, 0, 1, 2)
self.lineEdit = QtWidgets.QLineEdit(self.reg)
font = QtGui.QFont()
font.setFamily("Consolas")
font.setPointSize(12)
font.setBold(False)
font.setItalic(False)
font.setWeight(50)
self.lineEdit.setFont(font)
self.lineEdit.setStyleSheet(" QLineEdit\n"
" {border:0px;\n"
" border-radius:0;\n"
" font: 12pt \"Consolas\";\n"
" margin:15px;\n"
" border-bottom: 2px solid #B3B3B3;}\n"
" QLineEdit:hover{\n"
" border-bottom:3px solid #66A3FF;\n"
" }\n"
" QLineEdit:focus{\n"
" border-bottom:3px solid #E680BD\n"
" }")
self.lineEdit.setEchoMode(QtWidgets.QLineEdit.Normal)
self.lineEdit.setClearButtonEnabled(True)
self.lineEdit.setObjectName("lineEdit")
self.gridLayout_3.addWidget(self.lineEdit, 0, 0, 1, 2)
self.label_6 = QtWidgets.QLabel(self.reg)
self.label_6.setStyleSheet("color: rgb(255, 1, 39);\n"
"font: 11pt \"宋体\";")
self.label_6.setObjectName("label_6")
self.gridLayout_3.addWidget(self.label_6, 3, 0, 1, 2)
self.tabWidget.addTab(self.reg, "")
self.gridLayout.addWidget(self.tabWidget, 2, 0, 1, 3)
self.label_2 = QtWidgets.QLabel(self)
font = QtGui.QFont()
font.setFamily("Comic Sans MS")
font.setPointSize(14)
self.label_2.setFont(font)
self.label_2.setObjectName("label_2")
self.gridLayout.addWidget(self.label_2, 0, 1, 1, 1)
spacerItem = QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.gridLayout.addItem(spacerItem, 0, 2, 1, 1)
self.comboBox = QtWidgets.QComboBox(self)
font = QtGui.QFont()
font.setFamily("Prestige Elite Std")
font.setPointSize(13)
font.setBold(False)
font.setItalic(False)
font.setWeight(50)
self.comboBox.setFont(font)
self.comboBox.setStyleSheet("")
self.comboBox.setEditable(True)
self.comboBox.setObjectName("comboBox")
self.gridLayout.addWidget(self.comboBox, 1, 0, 1, 2)
self.tabWidget.setCurrentIndex(LOGIN_INFO["type"])
QtCore.QMetaObject.connectSlotsByName(self)
self.commandLinkButton.clicked.connect(lambda: self.tabWidget.setCurrentIndex(1))
_translate = QtCore.QCoreApplication.translate
self.setWindowTitle(_translate("Dialog", "登录 - 注册"))
self.lineEdit_5.setPlaceholderText(_translate("Dialog", "密码(4-10字符)"))
self.lineEdit_4.setPlaceholderText(_translate("Dialog", "用户名(2-12字符)"))
self.pushButton_2.setText(_translate("Dialog", "登录"))
self.checkBox_2.setText(_translate("Dialog", "记住密码"))
self.label_3.setText(_translate("Dialog", ""))
self.commandLinkButton.setText(_translate("Dialog", "没有账号? 注册一个"))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.log), _translate("Dialog", "登录"))
self.pushButton.setText(_translate("Dialog", "注册"))
self.checkBox.setText(_translate("Dialog", "记住密码"))
self.lineEdit_3.setPlaceholderText(_translate("Dialog", "再次输入密码(4-10字符)"))
self.lineEdit_2.setPlaceholderText(_translate("Dialog", "密码(4-10字符)"))
self.lineEdit.setPlaceholderText(_translate("Dialog", "用户名(2-12字符)"))
self.label_6.setText(_translate("Dialog", ""))
self.tabWidget.setTabText(self.tabWidget.indexOf(self.reg), _translate("Dialog", "注册"))
self.label_2.setText(_translate("Dialog", "Socketserver"))
self.comboBox.addItems(ip_list)
self.log_progress_signal.connect(self.label_3.setText)
self.reg_progress_signal.connect(self.label_6.setText)
self.handle_signal.connect(self.handle)
self.err_signal.connect(self.err_handle)
if LOGIN_INFO["type"] == 0:
# login
self.lineEdit_4.setText(LOGIN_INFO["username"])
self.lineEdit_5.setText(LOGIN_INFO["password"])
elif LOGIN_INFO["type"] == 1:
self.lineEdit.setText(LOGIN_INFO["username"])
self.lineEdit_2.setText(LOGIN_INFO["password"])
self.lineEdit_3.setText(LOGIN_INFO["password"])
self.pushButton.clicked.connect(self.register)
self.pushButton_2.clicked.connect(self.login)
self.show()
def Enable(self, boolean):
self.pushButton.setEnabled(boolean)
self.pushButton_2.setEnabled(boolean)
def get_ip(self) -> str:
return self.comboBox.currentText()
def clear(self):
self.label_3.setText("")
self.label_6.setText("")
def login(self, *args):
self.clear()
self.Enable(False)
self._login()
def err_handle(self, string):
self.label_3.setText(string)
self.label_6.setText(string)
self.Enable(True)
def handle(self, dictionary: (dict, str)):
if isinstance(dictionary, dict):
result = dictionary.get("result", False)
reason = dictionary.get("reason", False)
if not result:
self.err_handle(reason)
else:
self.successful = True
self.close()
self.Enable(True)
def login_trace(self, res: bool, reason: str):
self.label_3.setText(reason)
self.Enable(True)
def loading(self) -> callable:
self.loading_dialog = LoadingProgress(self)
return self.loading_dialog.update_signal.emit
def exec_loading_dialog(self):
if isinstance(self.loading_dialog, LoadingProgress):
self.loading_dialog.exec_() # 直接使用造成阻塞, 为此单独调用.
def _login(self):
username = self.lineEdit_4.text().strip()
password = self.lineEdit_5.text().strip()
if not username:
return self.login_trace(False, "未填写用户名!")
if not password:
return self.login_trace(False, "未填写密码!")
if not 2 <= len(username) <= 12:
return self.login_trace(False, "用户名需在2~12位之间!")
if not 4 <= len(password) <= 10:
return self.login_trace(False, "密码需在4~10位之间!")
try:
addr, port = self.get_ip().split(":")
assert isinstance(addr, str)
port = int(port)
except (ValueError, AssertionError):
return self.login_trace(False, "ipv4地址不正确! 结构:[host:port]")
s.change_address(addr, port)
s.Check_info({"type": 0, "username": username, "password": password}, self.loading(),
self.handle_signal.emit, self.err_signal.emit) # self.log_progress_signal
self.exec_loading_dialog()
global LOGIN_INFO
LOGIN_INFO["username"] = username
LOGIN_INFO["password"] = password
if self.checkBox_2.isChecked():
json_dump()
return True, ""
def register(self, *args):
self.Enable(False)
self.clear()
self._register()
def register_trace(self, res: bool, reason: str):
self.label_6.setText(reason)
self.Enable(True)
def _register(self):
username = self.lineEdit.text().strip()
password = self.lineEdit_2.text().strip()
password_check = self.lineEdit_3.text().strip()
if not password_check == password:
return self.register_trace(False, "两次输入密码不同!")
if not username:
return self.register_trace(False, "未填写用户名!")
if not password:
return self.register_trace(False, "未填写密码!")
if not 2 <= len(username) <= 12:
return self.register_trace(False, "用户名需在2~12位之间!")
if not 4 <= len(password) <= 10:
return self.register_trace(False, "密码需在4~10位之间!")
try:
addr, port = self.get_ip().split(":")
assert isinstance(addr, str)
port = int(port)
except (ValueError, AssertionError):
return self.register_trace(False, "ipv4地址不正确! 结构:[host:port]")
s.change_address(addr, port)
s.Check_info({"type": 1, "username": username, "password": password}, self.loading(),
self.handle_signal.emit, self.err_signal.emit) # self.reg_progress_signal
self.exec_loading_dialog()
global LOGIN_INFO
LOGIN_INFO["username"] = username
LOGIN_INFO["password"] = password
if self.checkBox.isChecked():
json_dump()
return True, ""
class Client(...):
...
@ignore
def forever_receive(self):
while self.isOpen():
result, reason, uname = self.server.user_record.handler(**self.json_data())
self.send(json.dumps({"result": result, "reason": reason}))
if result:
self.username = uname
break
self._login = True
self.server.login(self.username, self.addr)
while self.isOpen():
string = self.recv()
if string is None:
continue
elif self.com.iscommand(string):
self.send(self.com.handler(string))
else:
self.server.UserMessage(self.addr, self.username, string)
file = r'.\clients\data.json'
folder = r'.\clients'
if not path.exists(folder):
mkdir(folder)
class data:
def __init__(self):
if path.exists(file):
with open(file, 'r') as f:
self.data = load(f)
else:
self.data = {}
def __get__(self, username, default=None) -> tuple:
return self.data.get(username, default)
def __in__(self, username) -> bool:
return username in self.data.keys()
def __write__(self) -> None:
with open(file, 'w') as f:
dump(self.data, f, indent=4)
def __register__(self, username, password, time: (int, float) = time()) -> None:
self.data[username] = (encode(password), int(time))
self.__write__()
def __login__(self, username, password) -> bool:
return self.data[username][0] == encode(password)
def get_time(self, username):
return self.data[username][1]
def handler(self, type: int, username: str, password: str):
username = username.strip()
if not username:
return False, "未填写用户名!", ""
password = password.strip()
if not password:
return False, "未填写密码!", ""
if not 2 <= len(username) <= 12:
return False, "用户名需在2~12位之间!", ""
if not 4 <= len(password) <= 10:
return False, "密码需在4~10位之间!", ""
if type == 0: # login
if not self.__in__(username):
return False, "用户不存在!", ""
if not self.__login__(username, password):
return False, "用户名 / 密码错误!", ""
return True, "欢迎回来, " + username, username
elif type == 1: # register
if self.__in__(username):
return False, "已存在用户!", ""
self.__register__(username, password)
return True, "初来乍到, " + username, username