TCP聊天+传输文件服务器服务器套接字v2.6 - 登录注册界面更新 - loading界面应用_zmh-program

irpas 02-07 564

TCP聊天+传输文件服务器服务器套接字v2.6 更改的地方: 主要是客户端界面更改

注册，登录界面（都知道啊， v1.8的改进后输入用户名，到了主界面的时候才能输入密码，但现在是直接输入用户名和密码，然后转为json，等待回复)转到注册 (我用了commandLink)记住密码 (checkBox, 如果记住报存到config.json, 下次直接转入登录，设置为默认)两大堡垒 (客户端与服务端两次堡垒，防止更改客户端数据直接发送的[不符合规范的json数据包], 所以在服务端在识别[用户是否存在] + [账户密码是否正确]的前提下又增加了[检测用户名密码长度]的规则)界面淡入淡出

文章目录客户端更改服务端更改server.pywdata.py 客户端与服务端两次堡垒，防止更改客户端数据直接发送的[不符合规范的json数据包], 所以在服务端在识别用户[是否存在] + [账户密码是否正确]的前提下又增加了[检测用户名密码长度]的规则资源gitcode

所有版本记录: 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, "" 服务端更改 server.pyw 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) data.py 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 客户端与服务端两次堡垒，防止更改客户端数据直接发送的[不符合规范的json数据包], 所以在服务端在识别用户[是否存在] + [账户密码是否正确]的前提下又增加了[检测用户名密码长度]的规则资源

resource

gitcode

zmh-program / Python套接字Socket服务器