基于python的多线程聊天室的客户端部分（包括ini配置文件）
本帖最后由 白冥 于 2024-4-29 00:20 编辑事先说明，代码有点长，可能被吞：
client.py
from client_socket import ClientSocket
from client_thread import ClientThread
class client:
def __init__(self):
self.socket=ClientSocket()
self.thread=ClientThread(self.socket)
self.socket.start()
self.thread.start_thread(
send_thread=self.socket.send,
receive_thread=self.socket.receive,
heartbeat_thread=self.socket.heartbeat
)
if not self.thread.condition:
self.socket.stop()
client_socket.py
import socketimport timefrom config import Configfrom message import Messageclass ClientSocket: def __init__(self): self._init_socket() self.id=Config.clientHost self.running=True self.server_disconnected=False def _init_socket(self): self.socket=socket.socket(socket.AF_INET,socket.SOCK_STREAM) self.socket.connect((Config.serverHost,Config.usingPort)) def start(self): self.send(Config.MessageType.inquire,Config.Instruction.please,self.id,Config.serverId) def stop(self): self.running=False self.socket.close() def send(self,type,instruction,sender,addressee,content=''): MESSAGE=Message.dumps(type,instruction,sender,addressee,content) self.socket.sendall(MESSAGE) self.handle_send(MESSAGE) def receive(self): MESSAGE=self.socket.recv(Config.maximumTextLimit).decode() self.handle_receive(MESSAGE) if Message.is_message(MESSAGE) else None def handle_send(self,message): match message['type']: case Config.MessageType.inquire: match message["instruction"]: case Config.Instruction.bye: self.stop() def handle_receive(self,message): DICTIONARY=Message.loads(message) if type(DICTIONARY)==dict: match DICTIONARY['type']: case Config.MessageType.transmit: match DICTIONARY['instruction']: case Config.Instruction.text: print(f'A message from {DICTIONARY['sender']}\n') print('Main body:\n') print(f'{DICTIONARY['content']}\n') case Config.MessageType.detection: match DICTIONARY['instruction']: case Config.Instruction.detect: match DICTIONARY['sender']: case Config.serverId: self.server_disconnected=False case Config.MessageType.inquire: match DICTIONARY['instruction']: case Config.Instruction.bye: print(f'Addressee {DICTIONARY['sender']} stopped communication') case Config.MessageType.respond: match DICTIONARY['instruction']: case Config.Instruction.text: print(f'A message from {DICTIONARY['sender']}\n') print('Main body:\n') print(f'{DICTIONARY['content']}\n') case Config.MessageType.report: match DICTIONARY['sender']: case Config.serverId: match DICTIONARY['instruction']: case Config.Instruction.id: print(f'You has been assigned an ID:{DICTIONARY['content']}') self.sender_id=DICTIONARY['content'] case Config.Instruction.known: print('Reconnecting to the server succeeded.') def heartbeat(self): self.send(Config.MessageType.detection,Config.Instruction.detect,self.id,Config.serverId) self.server_disconnected=True time.sleep(Config.heartbeatRate) if self.server_disconnected: print('We temporarily lost contact with the server.') if not self.reconnect(): self.stop def reconnect(self,attempt=1): if attempt<=Config.maximumAttemptLimit: try: self.stop() self.running=True self._init_socket() return True except Exception as error: print(f'Failed to reconnect to the server') time.sleep(5) self.reconnect(attempt+1) else: return False
client_thread.py
import threadingclass ClientThread: def __init__(self,client_socket): self.condition=client_socket.running def loop(self,callback): while self.condition: callback() def start_thread(self,**threads): try: threads_list:list=[] for thread,callback in threads.items(): thread=threading.Thread(target=ClientThread.loop,args=(self,callback)) thread.start() threads_list.append(thread) for thread in threads_list: thread.join() except Exception as error: print(f'An error occurs:{error}') self.condition=False
config.py
import configparserCONFIG=configparser.ConfigParser()CONFIG.read('config.ini')class Config: serverHost=CONFIG.get('SERVER','host') clientHost=CONFIG.get('CLIENT','host') usingPort=CONFIG.getint('SERVER','port') serverId=CONFIG.get('SERVER','id') heartbeatRate=CONFIG.getint('CLIENT','heartbeat_rate') maximumAttemptLimit=CONFIG.getint('CLIENT','maximun_attempt_limit') maximumTextLimit=CONFIG.getint('TEXT','maximum_text_limit') class Instruction: text=CONFIG.get('INSTRUCTIONS','send_text') file=CONFIG.get('INSTRUCTIONS','send_file') error=CONFIG.get('INSTRUCTIONS','send_error') bye=CONFIG.get('INSTRUCTIONS','end_communication') please=CONFIG.get('INSTRUCTIONS','request_to_join') id=CONFIG.get('INSTRUCTIONS','allocation_id') call=CONFIG.get('INSTRUCTIONS','request_reconnection') known=CONFIG.get('INSTRUCTIONS','successfull_reconnection') detect=CONFIG.get('INSTRUCTIONS','heartbeat_detection') class MessageType: transmit=CONFIG.get('MESSAGE_TYPES','transmit') detection=CONFIG.get('MESSAGE_TYPES','detection') inquire=CONFIG.get('MESSAGE_TYPES','inquire') respond=CONFIG.get('MESSAGE_TYPES','respond') report=CONFIG.get('MESSAGE_TYPES','report')
message.py
import jsonclass Message: @staticmethod def dumps(msg_type,instruction,sender,addressee,content): DICTIONARY={ "type":msg_type, "instruction":instruction, "sender":sender, "addressee":addressee, "content":content, } return json.dumps(DICTIONARY).encode() @staticmethod def is_message(message): TEMPLATE={"type","instruction","sender","addressee","content"} try: DICTIONARY=json.loads(message) if type(DICTIONARY)==dict: return set(DICTIONARY.keys())==TEMPLATE except json.JSONDecodeError: return False @staticmethod def loads(message): if Message.is_message(message): return json.loads(message)
config.ini
host=127.0.0.1
port=6363
id=Server
host=localhost
heartbeat_rate=10
maximum_attempt_limit=3
send_text=/text
send_file=/file
send_error=/error
end_communication=/bye
request_to_join=/please
allocation_id=/id
request_reconnection=/call
successfull_reconnection=/known
heartbeat_detection=/detect
transmit=transmit
detection=detection
inquire=inquire
respond=respond
report=report
maximum_text_limit=4096
看起来应该是python3的代码，不过python3的版本也是挺多的 好专业的感觉呀，多线程聊天室是用于聊天用的代码吗？ 泥潭要转型github了吗 已晕，外行看热闹，内行看门道吧…… 专业起来了，，，话说服务端咋没一起放上来 好专业的样子，只能看看热闹的说XD 喔⊙ω⊙搜了搜，类似这种的咩https://blog.csdn.net/weixin_62428212/article/details/135996783 看起来好复杂的样子诶
咱对于代码一窍不通hh坐等一个傻瓜式教学QAQ
感觉是个很实用的工具，可惜我不懂惹~ 好复杂的代码，像我这种咸鱼就只能围观大佬们交流了{:6_194:} {:6_164:}虽然标题已经说明是什么东西，但我还是看不懂啊… 挺复杂的代码，感觉要同行才看得懂呢。 好好学习，以后说不定泥潭靠你出力～ 看上去好专业的东西，泥潭真是各种大佬都有啊 {:6_188:}有种在泥潭看到了课程设计精简版的美 懂了，下次写什么小玩意要嵌聊天的时候可以参考了
作为简易的就是足够简易才好，这个就很不错 额，我觉得是不是应该写个说明什么的？
没有介绍看不懂是做什么的{:6_167:} Python的代码真简洁，就这么短一点，要是用其他语言来写估计要写一大堆。
页:
[1]
2