前言
操作系统：win10
python版本：3.10.10
打包了个exe，方便使用：https://download.csdn.net/download/Ikaros_521/88716716
效果图
v1.0
v2.0
源码
v1.0
import socket
import threading
import logging
# 创建日志记录器
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# 创建一个格式器
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# 添加文件处理器
file_handler = logging.FileHandler('server.log')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 添加终端处理器
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
class TCPServer:
def __init__(self, host="0.0.0.0", port=55555):
self.host = host
self.port = port
self.clients = []
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置 SO_REUSEADDR 选项
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(5)
logger.info(f"Server listening on {
self.host}:{
self.port}")
def accept_clients(self):
while True:
client_socket, client_address = self.server_socket.accept()
logger.info(f"Client {
client_address} connected.")
self.clients.append(client_socket)
threading.Thread(target=self.handle_client, args=(client_socket,)).start()
def handle_client(self, client_socket):
while True:
try:
data = client_socket.recv(1024)
if not data:
break
# 十六进制显示
# hex_data = data.hex()
# logger.info(f"Received data from {client_socket.getpeername()}: {hex_data}")
logger.info(f"Received data from {
client_socket.getpeername()}: [{
data.decode()}]")
except socket.error as e:
logger.error(f"Error: {
e}")
break
client_socket.close()
self.clients.remove(client_socket)
def start(self):
threading.Thread(target=self.accept_clients).start()
threading.Thread(target=self.send_messages).start()
def send_messages(self):
while True:
message = input("Enter hex data to send to all clients (e.g., '331122'): ")
for client in self.clients[:]:
try:
try:
# 发送十六进制字符串
# byte_data = bytes.fromhex(message)
# client.sendall(byte_data)
# 发送ASCII字符串
client.sendall(message.encode())
logger.info(f"Sent data to {
client.getpeername()}: [{
message}]")
except ValueError:
logger.error("Invalid hex input.")
except socket.error as e:
logger.error(f"Error sending message: {
e}")
self.clients.remove(client)
if __name__ == "__main__":
server = TCPServer(host="0.0.0.0", port=55555)
server.start()
v2.0 终端输入ip 端口
import socket
import threading
import logging
# 创建日志记录器
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# 创建一个格式器
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# 添加文件处理器
file_handler = logging.FileHandler('server.log')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 添加终端处理器
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
class TCPServer:
def __init__(self, host="0.0.0.0", port=55555):
self.host = host
self.port = port
self.clients = []
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置 SO_REUSEADDR 选项
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(5)
logger.info(f"Server listening on {
self.host}:{
self.port}")
def accept_clients(self):
while True:
client_socket, client_address = self.server_socket.accept()
logger.info(f"Client {
client_address} connected.")
self.clients.append(client_socket)
threading.Thread(target=self.handle_client, args=(client_socket,)).start()
def handle_client(self, client_socket):
while True:
try:
data = client_socket.recv(1024)
if not data:
break
# 十六进制显示
# hex_data = data.hex()
# logger.info(f"Received data from {client_socket.getpeername()}: {hex_data}")
logger.info(f"Received data from {
client_socket.getpeername()}: [{
data.decode()}]")
except socket.error as e:
logger.error(f"Error: {
e}")
break
client_socket.close()
self.clients.remove(client_socket)
def start(self):
threading.Thread(target=self.accept_clients).start()
threading.Thread(target=self.send_messages).start()
def send_messages(self):
while True:
message = input("Enter hex data to send to all clients (e.g., '331122'): ")
for client in self.clients[:]:
try:
try:
# 发送十六进制字符串
# byte_data = bytes.fromhex(message)
# client.sendall(byte_data)
# 发送ASCII字符串
client.sendall(message.encode())
logger.info(f"Sent data to {
client.getpeername()}: [{
message}]")
except ValueError:
logger.error("Invalid hex input.")
except socket.error as e:
logger.error(f"Error sending message: {
e}")
self.clients.remove(client)
if __name__ == "__main__":
# 提示用户输入 IP 和端口，或使用默认值
host_input = input("Enter IP address (default '0.0.0.0'): ")
port_input = input("Enter port (default 55555): ")
# 如果直接回车，使用默认值
host = host_input if host_input else "0.0.0.0"
port = int(port_input) if port_input else 55555
# 创建并启动服务器
server = TCPServer(host=host, port=port)
server.start()
文章评论