TCP聊天服务器套接字v1.7
所有版本记录:
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)退出
文章目录
-
-
- server.ui
-
- 全部代码
-
- |server
-
- server.pyw
- Qlogging.py
- data.py 依旧没改动哈
终于来了, PyQt5服务端!
#界面
server.ui
代码语言：javascript复制<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>MainWindow</class>
<widget class="QMainWindow" name="MainWindow">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>788</width>
<height>685</height>
</rect>
</property>
<property name="font">
<font>
<family>Consolas</family>
<pointsize>11</pointsize>
</font>
</property>
<property name="windowTitle">
<string>Socket Server</string>
</property>
<property name="styleSheet">
<string notr="true"/>
</property>
<widget class="QWidget" name="centralwidget">
<layout class="QGridLayout" name="gridLayout">
<item row="4" column="0">
<widget class="QLabel" name="label_6">
<property name="text">
<string>Database(Logging and traceback):</string>
</property>
</widget>
</item>
<item row="5" column="0">
<widget class="QTextEdit" name="textEdit_2">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="0" column="0" rowspan="4">
<widget class="QGroupBox" name="groupBox">
<property name="title">
<string>Server Setup</string>
</property>
<layout class="QFormLayout" name="formLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label_2">
<property name="text">
<string>Maximum load(kb):</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QLineEdit" name="lineEdit">
<property name="text">
<string>1024</string>
</property>
</widget>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_8">
<property name="text">
<string>backlog:</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLineEdit" name="lineEdit_3">
<property name="text">
<string>10</string>
</property>
</widget>
</item>
<item row="2" column="0">
<widget class="QLabel" name="label_7">
<property name="text">
<string>CODEC(Unalterable):</string>
</property>
</widget>
</item>
<item row="2" column="1">
<widget class="QLineEdit" name="lineEdit_2">
<property name="text">
<string>utf8</string>
</property>
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
<item row="5" column="0">
<spacer name="horizontalSpacer">
<property name="orientation">
<enum>Qt::Horizontal</enum>
</property>
<property name="sizeHint" stdset="0">
<size>
<width>40</width>
<height>20</height>
</size>
</property>
</spacer>
</item>
<item row="5" column="1">
<widget class="QPushButton" name="pushButton">
<property name="text">
<string>Run</string>
</property>
</widget>
</item>
<item row="3" column="1">
<widget class="QLineEdit" name="lineEdit_4">
<property name="text">
<string>127.0.0.1</string>
</property>
</widget>
</item>
<item row="3" column="0">
<widget class="QLabel" name="label_9">
<property name="text">
<string>Address:</string>
</property>
</widget>
</item>
<item row="4" column="0">
<widget class="QLabel" name="label_10">
<property name="text">
<string>Port:</string>
</property>
</widget>
</item>
<item row="4" column="1">
<widget class="QLineEdit" name="lineEdit_5">
<property name="text">
<string>429</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
<item row="2" column="1" rowspan="4">
<widget class="Line" name="line">
<property name="orientation">
<enum>Qt::Vertical</enum>
</property>
</widget>
</item>
<item row="0" column="1" colspan="3">
<widget class="QLabel" name="label_5">
<property name="font">
<font>
<family>Comic Sans MS</family>
<pointsize>14</pointsize>
<weight>50</weight>
<bold>false</bold>
</font>
</property>
<property name="styleSheet">
<string notr="true">background-color:qlineargradient(spread:pad, x1:0, y1:0, x2:1, y2:1, stop:0 rgba(0, 255, 241, 255), stop:0.930348 rgba(0, 158, 255, 255));
color:rgb(85, 0, 255)</string>
</property>
<property name="text">
<string>TCP Server v{___version__}</string>
</property>
</widget>
</item>
<item row="1" column="2" rowspan="5" colspan="2">
<widget class="QGroupBox" name="groupBox_2">
<property name="enabled">
<bool>false</bool>
</property>
<property name="title">
<string>Run</string>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label">
<property name="text">
<string>Running memory with CPU</string>
</property>
</widget>
</item>
<item row="1" column="1">
<widget class="QLCDNumber" name="lcdNumber"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_3">
<property name="text">
<string>Online user(s):</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QProgressBar" name="progressBar">
<property name="value">
<number>56</number>
</property>
</widget>
</item>
<item row="3" column="0" colspan="2">
<widget class="Line" name="line_3">
<property name="orientation">
<enum>Qt::Horizontal</enum>
</property>
</widget>
</item>
<item row="4" column="0">
<widget class="QLabel" name="label_4">
<property name="text">
<string>Logged in users:</string>
</property>
</widget>
</item>
<item row="5" column="0" colspan="2">
<widget class="QListWidget" name="listWidget"/>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
<widget class="QStatusBar" name="statusbar"/>
</widget>
<resources/>
<connections>
<connection>
<sender>pushButton</sender>
<signal>clicked()</signal>
<receiver>groupBox_2</receiver>
<slot>show()</slot>
<hints>
<hint type="sourcelabel">
<x>325</x>
<y>141</y>
</hint>
<hint type="destinationlabel">
<x>652</x>
<y>346</y>
</hint>
</hints>
</connection>
</connections>
</ui>至于为什么这么久才更新， PyQt5气煞人了, dll又不见了, Qpainter没调用自己还有问题, 又不让线程修改页面什么的，还有信号还不能converted, 越搜越乱.终于出来了
全部代码
|server
server.pyw
代码语言：javascript复制import os, socket, sys, time, logging
from threading import Thread
from Qlogging import QLogger
import data #同目录data.py
from psutil import cpu_percent
from PyQt5 import QtCore, QtGui, QtWidgets
#from signal import SIGINT, signal
__version__ = 1.7
def threading(Daemon, **kwargs):
thread = Thread(**kwargs)
thread.setDaemon(Daemon)
thread.start()
return thread
def ignore(function):
def i(*args, **kwargs):
try:
function(*args, **kwargs)
except:
return
return i
logger = logging.getLogger(__name__)
logger.setLevel(level=logging.DEBUG)
Qlog = QLogger()
logger.addHandler(Qlog)
filehandle = logging.FileHandler("log.txt")
filehandle.setFormatter(logging.Formatter("[%(asctime)s(%(levelname)s)]: %(message)s"))
logger.addHandler(filehandle)
logger.setLevel(logging.DEBUG)
bytecount = 1024
def to_logging(command):
def logs(*args, **kwargs):
try:
command(*args, **kwargs)
except:
logger.exception(str())
return False
return True
return logs
class Command_Handler(object):
def __init__(self, bind):
"""Bind Client class"""
assert isinstance(bind, Client)
self.client = bind
def _function(self, _list):
data = {"/info": {"-v": self.get_version(),
"-id": self.get_id(),
"-i": self.info(),
"-h": self.help(),
"-name": self.name()},
}
_dict = data
for n in range(len(_list)):
if type(_dict) == dict:
_dict = _dict.get(_list[n], self.unknown(" ".join(_list)))
else:
break
if type(_dict) == dict:
_dict = "Error:n<font color='blue'>This command must take more arguments. Such as %s.</font>" % list(
_dict.keys())
return _dict
@staticmethod
def help():
return """/info [-v] [-id] [-i]
-v : get version of program.
-id : get your id.
-i : get information.
-h : help.
-name : get your name
For example, <font color=red>/info -id</font>"""
@staticmethod
def get_version():
return "version : " str(__version__)
def get_id(self):
return "Your id is {}.".format(id(self.client))
def name(self):
return "Your name is {}.".format(self.client.username)
def info(self):
return f"Socket Server[version {self.get_version()}] By zmh."
def unknown(self, s):
return """Error:
No command named "%s". Please search [/info -h] to help.
%s""" % (s, self.help())
def cut(self, string):
return string.strip().split()
def handler(self, c):
return "<font color='gray'>[command]</font><font color='brown'>%s</font>n%s" % (
c, str(self._function(self.cut(c))))
def iscommand(self, i):
return i.strip().startswith("/")
class Server():
join_message = "<font color='red'>Server></font> <font color='blue'>%s(%s)</font> 连接服务器. 当前在线人数: <font color='red'>%s</font>"
user_message = "<font color='%s'>%s(%s)%s></font> %s"
quit_message = "%s(%s) 下线了, %s"
def __init__(self, gui=None):
self._call_func = bool(gui)
if self._call_func:
self.add_user = gui.add_user
self.remove_user = gui.remove_user
self.user_num_change = gui.usernumChanged
def Setup(self, addr, port, backlog=10, max_count=bytecount**2, encode='utf8'):
self.address = addr, port
self.backlog = backlog
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.bind(self.address)
self.socket.listen(backlog)
self.max_count = max_count
self.connect = []
self.encode = encode
self.user_record = data.user()
return self.run()
def clear_socket(self, clear_ms = 500):
logger.info(f"Clear the closed socket once every {clear_ms} ms.")
while True:
del_list = list(filter(lambda c: hasattr(c, 'Quitted') or (not c.isOpen()), self.connect))
for user in del_list:
self.connect.remove(user)
#if del_list:
# logger.info(f"Clear the closed client socket, number is {len(del_list)}.")
#else:
# logger.info('None of the sockets have been cleaned.')
time.sleep(clear_ms / 1000)
def run(self):
logger.debug(f"pid {os.getpid()}.")
logger.info(f"Server [{':'.join(map(lambda i: str(i), self.address))}] on.")
logger.info("Backlog number: " str(self.backlog))
logger.info('The CODEC is sent as ' self.encode)
threading(Daemon=True, target=self.clear_socket)
return threading(Daemon=True, target=self.accept_client)
def _get_Clients(self) -> list:
def func(c):
return c.__filter__()
return list(filter(func, self.connect))
def _get_sockets(self): # return int
i = len(self._get_Clients())
if self._call_func:
self.user_num_change(i)
return i
def _str_sockets(self):
return f"当前人数 {self._get_sockets()}"
def ServerMessage(self, mes, inc=True):
for user in self._get_Clients():
if user.__filter__():
user._send(mes)
def UserMessage(self, address, _user, mes, inc=True):
if not mes:
return
for user in self.connect:
if user.__filter__():
username = user.username
send_message = Server.user_message % ("brown" if _user == username else "red",
_user,
address,
"(我自己)" if _user == username else "",
mes)
user._send(send_message)
logger.info(f"{address}[{_user}] : {mes}")
def error_handle(self):
for user in filter(lambda c: not c.isOpen(), self.connect):
print(user)
self.connect.remove(user)
def accept_client(self):
while True:
logger.info("The server is listening on the port.")
client, address = self.socket.accept() # 阻塞，等待客户端连接
NewClient = Client(client, address[0], self)
self.connect.append(NewClient)
NewClient.run()
logger.info(f'The address {address[0]} is connected to the server')
def quit(self, username, address):
if self._call_func:
self.remove_user(username)
QuitMessage = Server.quit_message % (username, address, self._str_sockets())
logger.info(QuitMessage)
self.ServerMessage(QuitMessage, False)
def login(self, username, address):
print(username)
if self._call_func:
self.add_user(username)
logger.info(f"{address}[{username}] 登录服务器 , " self._str_sockets())
self.ServerMessage(Server.join_message % (username, address, self._get_sockets()))
class Client(object):
class QuitError(Exception):
def __init__(self, *args):
super().__init__(*args)
def __init__(self, socket, addr, server: Server):
self.socket = socket
self.addr = addr
if not isinstance(server, Server):
raise ValueError
self.server = server
self.encode = self.server.encode
self.max_count = self.server.max_count
self.com = Command_Handler(self)
@self.error
def _recv(self) -> bytes:
return self.socket.recv(self.max_count).decode(encoding=self.encode).strip()
self._recv = lambda: _recv(self)
@self.error
def _send(self, message=None) -> None:
if message:
if type(message) == str:
message.encode(self.encode)
message = b'n' #防止粘包
self.socket.sendall(message)
self._send = lambda m: _send(self, m)
def __del__(self):
self.socket.close()
def isLogin(self) -> bool:
return hasattr(self, "_login") and self._login
def isOpen(self) -> bool:
return not getattr(self.socket, "_closed", True)
def __filter__(self) -> bool:
"""返回是否在线并已可接受消息"""
return self.isLogin() and self.isOpen()
def recv(self) -> str:
data = self._recv()
while not data:
data = self._recv()
# while not (data := self._recv()):
# pass
# 我的PythonIDE是3.8, PyCharm是3.7(anaconda 32x),而赋值表达式是3.8加进来的.
return data
@ignore
def login(self):
self._send(f'<font color="red">欢迎来到服务器[{self.server.address[0]}].您的ip地址为{self.socket.getpeername()[0]}')
self.username = self.recv()[:8]
if self.server.user_record.__in__(self.username):
self._send("<font color='red'>请输入您的密码: (右下[send]键发送)</font>")
i = self.recv()
if self.server.user_record.handler(self.username, i):
self._send(f'<font color="green">欢迎回来, {self.username}.</font>')
else:
self._send('<font color="red">密码错误,请重试.</font>')
self.__del__()
else:
def normal(string):
return (4 <= len(string) <= 10) and not ('n' in string)
while True:
self._send(
"<font color='blue'>[i]提示: 密码需在4 ~ 10位之间, 且不能换行.</font>n<font color='red'>请输入您的密码: (右下[send]键发送)</font>")
p1 = self.recv()
if normal(p1):
break
while True:
self._send("<font color='red'>再次输入您的密码: (右下[send]键发送)</font>")
p2 = self.recv()
if p1 == p2:
break
else:
self._send("<font color='red'>密码与前次不符!</font>")
self.server.user_record.handler(self.username, p1)
self._send(f'初来乍到, {self.username}')
self._login = True
self.server.login(self.username, self.addr)
def quit(self) -> None:
if hasattr(self, 'Quitted'):
return
self.Quitted = True
if self.isOpen() is True:
self.socket.close()
self.server.quit(self.username, self.addr)
@ignore
def forever_receive(self):
self.login()
while self.__filter__():
string = self.recv()
if string == Client.QuitError:
return
if self.com.iscommand(string):
self._send(self.com.handler(string))
else:
self.server.UserMessage(self.addr, self.username, string)
def error(self, func):
def function(*args, **kwargs):
try:
res = func(*args, **kwargs)
return res
except ConnectionAbortedError as e:
self.quit()
except Exception:
logger.exception("error")
return Client.QuitError
return function
def run(self):
self.thread = threading(True, target=self.forever_receive)
def get_host_ip() -> str:
"""get current IP address"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(('8.8.8.8', 80))
ip = s.getsockname()[0]
finally:
s.close()
return ip
class Ui_MainWindow(object):
def __init__(self):
self.setupUi()
self.MainWindow.show()
def setupUi(self):
MainWindow = QtWidgets.QMainWindow()
self.MainWindow = MainWindow
MainWindow.setObjectName("MainWindow")
MainWindow.resize(788, 685)
font = QtGui.QFont()
font.setFamily("Consolas")
font.setPointSize(11)
MainWindow.setFont(font)
MainWindow.setStyleSheet("")
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.gridLayout = QtWidgets.QGridLayout(self.centralwidget)
self.gridLayout.setObjectName("gridLayout")
self.label_6 = QtWidgets.QLabel(self.centralwidget)
self.label_6.setObjectName("label_6")
self.gridLayout.addWidget(self.label_6, 4, 0, 1, 1)
self.textEdit_2 = QtWidgets.QTextEdit(self.centralwidget)
self.textEdit_2.setObjectName("textEdit_2")
self.gridLayout.addWidget(self.textEdit_2, 5, 0, 1, 1)
self.groupBox = QtWidgets.QGroupBox(self.centralwidget)
self.groupBox.setObjectName("groupBox")
self.formLayout_2 = QtWidgets.QFormLayout(self.groupBox)
self.formLayout_2.setObjectName("formLayout_2")
self.label_2 = QtWidgets.QLabel(self.groupBox)
self.label_2.setObjectName("label_2")
self.formLayout_2.setWidget(0, QtWidgets.QFormLayout.LabelRole, self.label_2)
self.lineEdit = QtWidgets.QLineEdit(self.groupBox)
self.lineEdit.setObjectName("lineEdit")
self.formLayout_2.setWidget(0, QtWidgets.QFormLayout.FieldRole, self.lineEdit)
self.label_8 = QtWidgets.QLabel(self.groupBox)
self.label_8.setObjectName("label_8")
self.formLayout_2.setWidget(1, QtWidgets.QFormLayout.LabelRole, self.label_8)
self.lineEdit_3 = QtWidgets.QLineEdit(self.groupBox)
self.lineEdit_3.setObjectName("lineEdit_3")
self.formLayout_2.setWidget(1, QtWidgets.QFormLayout.FieldRole, self.lineEdit_3)
self.label_7 = QtWidgets.QLabel(self.groupBox)
self.label_7.setObjectName("label_7")
self.formLayout_2.setWidget(2, QtWidgets.QFormLayout.LabelRole, self.label_7)
self.lineEdit_2 = QtWidgets.QLineEdit(self.groupBox)
self.lineEdit_2.setReadOnly(True)
self.lineEdit_2.setObjectName("lineEdit_2")
self.formLayout_2.setWidget(2, QtWidgets.QFormLayout.FieldRole, self.lineEdit_2)
spacerItem = QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
self.formLayout_2.setItem(5, QtWidgets.QFormLayout.LabelRole, spacerItem)
self.pushButton = QtWidgets.QPushButton(self.groupBox)
self.pushButton.setObjectName("pushButton")
self.formLayout_2.setWidget(5, QtWidgets.QFormLayout.FieldRole, self.pushButton)
self.lineEdit_4 = QtWidgets.QLineEdit(self.groupBox)
self.lineEdit_4.setObjectName("lineEdit_4")
self.formLayout_2.setWidget(3, QtWidgets.QFormLayout.FieldRole, self.lineEdit_4)
self.label_9 = QtWidgets.QLabel(self.groupBox)
self.label_9.setObjectName("label_9")
self.formLayout_2.setWidget(3, QtWidgets.QFormLayout.LabelRole, self.label_9)
self.label_10 = QtWidgets.QLabel(self.groupBox)
self.label_10.setObjectName("label_10")
self.formLayout_2.setWidget(4, QtWidgets.QFormLayout.LabelRole, self.label_10)
self.lineEdit_5 = QtWidgets.QLineEdit(self.groupBox)
self.lineEdit_5.setObjectName("lineEdit_5")
self.formLayout_2.setWidget(4, QtWidgets.QFormLayout.FieldRole, self.lineEdit_5)
self.gridLayout.addWidget(self.groupBox, 0, 0, 4, 1)
self.line = QtWidgets.QFrame(self.centralwidget)
self.line.setFrameShape(QtWidgets.QFrame.VLine)
self.line.setFrameShadow(QtWidgets.QFrame.Sunken)
self.line.setObjectName("line")
self.gridLayout.addWidget(self.line, 2, 1, 4, 1)
self.label_5 = QtWidgets.QLabel(self.centralwidget)
font = QtGui.QFont()
font.setFamily("Comic Sans MS")
font.setPointSize(14)
font.setBold(False)
font.setWeight(50)
self.label_5.setFont(font)
self.label_5.setStyleSheet("background-color:qlineargradient(spread:pad, x1:0, y1:0, x2:1, y2:1, stop:0 rgba(0, 255, 241, 255), stop:0.930348 rgba(0, 158, 255, 255));n"
"color:rgb(85, 0, 255)")
self.label_5.setObjectName("label_5")
self.gridLayout.addWidget(self.label_5, 0, 1, 1, 3)
self.groupBox_2 = QtWidgets.QGroupBox(self.centralwidget)
self.groupBox_2.setEnabled(False)
self.groupBox_2.setObjectName("groupBox_2")
self.gridLayout_2 = QtWidgets.QGridLayout(self.groupBox_2)
self.gridLayout_2.setObjectName("gridLayout_2")
self.lcdNumber = QtWidgets.QLCDNumber(self.groupBox_2)
self.lcdNumber.setObjectName("lcdNumber")
self.gridLayout_2.addWidget(self.lcdNumber, 1, 1, 1, 1)
self.line_3 = QtWidgets.QFrame(self.groupBox_2)
self.line_3.setFrameShape(QtWidgets.QFrame.HLine)
self.line_3.setFrameShadow(QtWidgets.QFrame.Sunken)
self.line_3.setObjectName("line_3")
self.gridLayout_2.addWidget(self.line_3, 3, 0, 1, 2)
self.listView_2 = QtWidgets.QListWidget(self.groupBox_2)
self.listView_2.setObjectName("listView_2")
self.gridLayout_2.addWidget(self.listView_2, 5, 0, 1, 2)
self.label_4 = QtWidgets.QLabel(self.groupBox_2)
self.label_4.setObjectName("label_4")
self.gridLayout_2.addWidget(self.label_4, 4, 0, 1, 1)
self.label_3 = QtWidgets.QLabel(self.groupBox_2)
self.label_3.setObjectName("label_3")
self.gridLayout_2.addWidget(self.label_3, 1, 0, 1, 1)
self.label = QtWidgets.QLabel(self.groupBox_2)
self.label.setObjectName("label")
self.gridLayout_2.addWidget(self.label, 0, 0, 1, 1)
self.progressBar = QtWidgets.QProgressBar(self.groupBox_2)
self.progressBar.setProperty("value", 0)
self.progressBar.setObjectName("progressBar")
self.gridLayout_2.addWidget(self.progressBar, 0, 1, 1, 1)
self.gridLayout.addWidget(self.groupBox_2, 1, 2, 5, 2)
self.textEdit_2.setReadOnly(True)
MainWindow.setCentralWidget(self.centralwidget)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.groupBox_2.setEnabled(False)
self.retranslateUi(MainWindow)
self.pushButton.clicked.connect(self.run)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
self.lcdNumber.display(0)
self.cpu = cpuThread()
self.cpu.signal.connect(self.update_progessBar)
Qlog.connect(self.update_textEdit_2)
def update_progessBar(self,v):
self.progressBar.setValue(int(v))
@to_logging
def handle(self):
self.max_recv = int(self.lineEdit.text())
self.backlog = int(self.lineEdit_3.text())
self.addr = self.lineEdit_4.text()
self.port = int(self.lineEdit_5.text())
server.Setup(self.addr, self.port, self.backlog, self.max_recv)
def run(self, _):
if self.handle():
self.groupBox.setEnabled(False)
self.groupBox_2.setEnabled(True)
self.timer = QtCore.QTimer()
self.cpu.start()
def update_textEdit_2(self, data):
if data:
for x in str(data).split("n"):
self.textEdit_2.append(data.strip())
self.textEdit_2.moveCursor(QtGui.QTextCursor.End)
def usernumChanged(self, int):
if int != int(self.lcdNumber.value()):
self.lcdNumber.display(int)
def add_user(self, name:str):
self.listView_2.addItem(name)
def remove_user(self, name:str):
for i in range(self.listView_2.count()):
if name == self.listView_2.item(i).text():
self.listView_2.takeItem(i)
return True
return False
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "Socket Server"))
self.label_6.setText(_translate("MainWindow", "Database(Logging and traceback):"))
self.groupBox.setTitle(_translate("MainWindow", "Server Setup"))
self.label_2.setText(_translate("MainWindow", "Maximum load(kb):"))
self.lineEdit.setText(_translate("MainWindow", "1024"))
self.label_8.setText(_translate("MainWindow", "backlog:"))
self.lineEdit_3.setText(_translate("MainWindow", "10"))
self.label_7.setText(_translate("MainWindow", "CODEC(Unalterable):"))
self.lineEdit_2.setText(_translate("MainWindow", "utf8"))
self.pushButton.setText(_translate("MainWindow", "Run"))
self.lineEdit_4.setText(_translate("MainWindow", "127.0.0.1"))
self.label_9.setText(_translate("MainWindow", "Address:"))
self.label_10.setText(_translate("MainWindow", "Port:"))
self.lineEdit_5.setText(_translate("MainWindow", "429"))
self.label_5.setText(_translate("MainWindow", f"TCP Server v{__version__}"))
self.groupBox_2.setTitle(_translate("MainWindow", "Run"))
self.label_4.setText(_translate("MainWindow", "Logged in users:"))
self.label_3.setText(_translate("MainWindow", "Online user(s):"))
self.label.setText(_translate("MainWindow", "Running memory with CPU"))
class cpuThread(QtCore.QThread):
signal = QtCore.pyqtSignal(int)
def run(self) -> None:
while True:
self.signal.emit(int(cpu_percent(interval=1)))
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
gui = Ui_MainWindow()
server = Server()
sys.exit(app.exec_())Qlogging.py
代码语言：javascript复制import logging
class QLogger(logging.Handler):
def __init__(self, *args, **kwargs):
logging.Handler.__init__(self, *args, **kwargs)
self.setFormatter(logging.Formatter("[<font color='darkgreen'>%(asctime)s</font>(<font color='blue'>%(levelname)s</font>)]: <font color='brown'>%(message)s</font>"))
def emit(self, record):
record = self.format(record)
if hasattr(self, "output"):
self.output(record)
def connect(self, func):
if callable(func):
self.output = funcdata.py 依旧没改动哈
代码语言：javascript复制from json import load, dump
from os import path, mkdir
from hashlib import md5
from time import time
def encode(data: str):
m = md5()
m.update(data.encode('utf8'))
return m.hexdigest()
file = '.clientsdata.json'
folder = '.clients'
if not path.exists(folder):
mkdir(folder)
class user():
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)
def __register__(self, username, password, time: (int, float)) -> None:
self.data[username] = (encode(password), int(time))
self.__write__()
def __login__(self, username, password) -> bool:
return self.data[username][0] == encode(username)
def handler(self, username, password) -> bool:
if self.__in__(username):
return self.__login__(username, password)
else:
self.__register__(username, password, time())
return True
def get_time(self, username):
return self.data[username][1]
