該程式由客戶端與伺服器構成，使用UDP服務，伺服器端繫結本地IP和埠，客戶端由系統隨機選擇埠。

實現了群發、私發、點對點檔案互傳功能。

客戶端自建了一個類繼承了Cmd模組，使用自定義的命令command進行操作，呼叫相應的do_command方法。

使用json模組進行訊息的封裝序列化，在接收方進行解析。

客戶端程式碼如下：

import socket
import threading
import json
import os
from cmd import Cmd
 
 
class Client(Cmd):
    """
    客戶端
    """
    prompt = '>>>'
    intro = '[Welcome] 簡易聊天室客戶端(Cli版)\n' + '[Welcome] 輸入help來獲取幫助\n'
    buffersize = 1024
 
    def __init__(self, host):
        """
        構造
        """
        super().__init__()
        self.__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # self.__id = None
        self.__nickname = None
        self.__host = host
        self.thread_recv = None
        self.threadisalive = False
        # 是否在接收檔案
        self.recvfile = False
        # 是否在傳送檔案
        self.sendfile = False
        self.filesize = None
        self.sendfilesize = None
 
        # 接收檔案包計數
        self.filecount = None
        # 接收檔名
        self.filename = None
        # 傳送檔名
        self.sendfilename = None
 
        # 傳送者
        self.filefrom = None
        # 接收者
        self.fileto = None
 
        # 接收檔案流
        self.file_recv = None
        # 傳送檔案流
        self.file_send = None
 
        # 接收檔案地址
        self.filefrom_addr = None
        # 傳送檔案地址
        self.fileto_addr = None
 
    def __receive_message_thread(self):
        """
        接受訊息執行緒
        """
        while self.threadisalive:
            # noinspection PyBroadException
            try:
                buffer, addr = self.__socket.recvfrom(1024)
                '''
                檔案流由傳送端直接傳送，不經過伺服器，故當傳送端發來的訊息時，將收到的資料存入檔案
                '''
                if (addr != self.__host) & (addr == self.filefrom_addr) & self.recvfile:
                    self.file_recv.write(buffer)
                    self.filecount += 1
                    if self.filecount * 1024 >= self.filesize:
                        self.file_recv.close()
                        print(self.filename, 'is received.')
                        self.recvfile = False
                    continue
 
                js = json.loads(buffer.decode())
 
                # 若接收的資料為訊息資訊，則顯示
                if js['type'] == 'message':
                    print(js['message'])
 
                # 若接收的資料為檔案傳送請求，則儲存檔案資訊，並顯示
                elif js['type'] == 'filequest':
                    if self.recvfile:
                        self.__socket.sendto(json.dumps({
                            'type': 'fileres',
                            'fileres': 'no',
                            'nickname': self.__nickname,
                            'who': js['nickname'],
                            'errormessage': 'is transfroming files.',
                        }).encode(), self.__host)
                        continue
                    filename = js['filename']
                    who = js['nickname']
                    filesize = js['filesize']
                    self.recvfile = True
                    self.filesize = filesize
                    self.filename = filename
                    self.filecount = 0
                    self.filefrom = who
                    self.filefrom_addr = (js['send_ip'], js['send_port'])
 
                    print('[system]:', who, ' send a file(',
                          filename, ') to you. receive? ')
 
                # 接受的資料為請求回覆，若同意接收則儲存伺服器發來的接收方的地址，並開啟發送執行緒
                elif js['type'] == 'fileres':
                    if js['fileres'] == 'yes':
                        print(js['recv_ip'], js['recv_port'])
                        self.fileto_addr = (js['recv_ip'], js['recv_port'])
                        thread = threading.Thread(
                            target=self.__send_file_thread)
                        thread.start()
                    else:
                        print(js['nickname'], js['errormessage'])
                        self.sendfile = False
 
            except Exception as e:
                print(e)
                print('[Client] 無法從伺服器獲取資料')
 
    def __send_broadcast_message_thread(self, message):
        """
        傳送廣播訊息執行緒
        :param message: 訊息內容
        """
        self.__socket.sendto(json.dumps({
            'type': 'broadcast',
            'nickname': self.__nickname,
            'message': message,
        }).encode(), self.__host)
 
    def __send_file_thread(self):
        """
        傳送檔案執行緒
        :param message: 訊息內容
        """
        filecount = 0
        print('[system]', 'sending the file...')
        while filecount * 1024 <= self.sendfilesize:
            self.__socket.sendto(
                self.file_send.read(1024), self.fileto_addr)
            filecount += 1
        self.file_send.close()
        self.sendfile = False
        print('[system]', 'the file is sended.')
 
    def __send_whisper_message_thread(self, who, message):
        """
        傳送私發訊息執行緒
        :param message: 訊息內容
        """
        self.__socket.sendto(json.dumps({
            'type': 'sendto',
            'who': who,
            'nickname': self.__nickname,
            'message': message
        }).encode(), self.__host)
 
    def send_exit(self):
        self.__socket.sendto(json.dumps({
            'type': 'offline',
            'nickname': self.__nickname,
        }).encode(), self.__host)
 
 
    def start(self):
        """
        啟動客戶端
        """
        self.cmdloop()
 
    def do_login(self, args):
        """
        登入聊天室
        :param args: 引數
        """
        nickname = args.split(' ')[0]
 
        # 將暱稱傳送給伺服器，獲取使用者id
        self.__socket.sendto(json.dumps({
            'type': 'login',
            'nickname': nickname,
        }).encode(), self.__host)
        # 嘗試接受資料
 
        buffer = self.__socket.recvfrom(1300)[0].decode()
        obj = json.loads(buffer)
        if obj['login'] == 'success':
            self.__nickname = nickname
            print('[Client] 成功登入到聊天室')
            self.threadisalive = True
            # 開啟子執行緒用於接受資料
            self.thread_recv = threading.Thread(
                target=self.__receive_message_thread)
            self.thread_recv.setDaemon(True)
            self.thread_recv.start()
        else:
            print('[Client] 無法登入到聊天室', obj['errormessage'])
 
    def do_send(self, args):
        """
        傳送訊息
        :param args: 引數
        """
        if self.__nickname is None:
            print('請先登入！login nickname')
            return
        message = args
        # 開啟子執行緒用於傳送資料
        thread = threading.Thread(
            target=self.__send_broadcast_message_thread, args=(message, ))
        thread.setDaemon(True)
        thread.start()
 
    def do_sendto(self, args):
        """
        傳送私發訊息
        :param args: 引數
        """
        if self.__nickname is None:
            print('請先登入！login nickname')
            return
        who = args.split(' ')[0]
        message = args.split(' ')[1]
        # # 顯示自己傳送的訊息
        # print('[' + str(self.__nickname) + '(' + str(self.__id) + ')' + ']', message)
        # 開啟子執行緒用於傳送資料
        thread = threading.Thread(
            target=self.__send_whisper_message_thread, args=(who, message))
        thread.setDaemon(True)
        thread.start()
 
    def do_catusers(self, arg):
        if self.__nickname is None:
            print('請先登入！login nickname')
            return
        catmessage = json.dumps({'type': 'catusers'})
        self.__socket.sendto(catmessage.encode(), self.__host)
 
    def do_catip(self, args):
        if self.__nickname is None:
            print('請先登入！login nickname')
            return
        who = args
        catipmessage = json.dumps({'type': 'catip', 'who': who})
        self.__socket.sendto(catipmessage.encode(), self.__host)
 
    def do_help(self, arg):
        """
        幫助
        :param arg: 引數
        """
        command = arg.split(' ')[0]
        if command == '':
            print('[Help] login nickname - 登入到聊天室，nickname是你選擇的暱稱')
            print('[Help] send message - 傳送訊息，message是你輸入的訊息')
            print('[Help] sendto who message - 私發訊息，who是使用者名稱，message是你輸入的訊息')
            print('[Help] catusers - 檢視所有使用者')
            print('[Help] catip who - 檢視使用者IP，who為使用者名稱')
            print('[Help] sendfile who filedir - 向某使用者傳送檔案，who為使用者名稱，filedir為檔案路徑')
            print('[Help] getfile filename who yes/no - 接收檔案，filename 為檔名,who為傳送者，yes/no為是否接收')
        elif command == 'login':
            print('[Help] login nickname - 登入到聊天室，nickname是你選擇的暱稱')
        elif command == 'send':
            print('[Help] send message - 傳送訊息，message是你輸入的訊息')
        elif command == 'sendto':
            print('[Help] sendto who message - 傳送私發訊息，message是你輸入的訊息')
        else:
            print('[Help] 沒有查詢到你想要了解的指令')
 
    def do_exit(self, arg):  # 以do_*開頭為命令
        print("Exit")
        self.send_exit()
        try:
            self.threadisalive = False
            self.thread_recv.join()
        except Exception as e:
            print(e)
        # self.__socket.close()
 
    def do_sendfile(self, args):
        who = args.split(' ')[0]
        filepath = args.split(' ')[1]
        filename = filepath.split('\\')[-1]
        # 判斷是否在傳送檔案
        if self.sendfile:
            print('you are sending files, please try later.')
            return
        if not os.path.exists(filepath):
            print('the file is not exist.')
            return
        filesize = os.path.getsize(filepath)
        # print(who, filename, filesize)
 
        self.sendfile = True
        self.fileto = who
        self.sendfilename = filename
        self.sendfilesize = filesize
        self.file_send = open(filepath, 'rb')
 
        self.__socket.sendto(json.dumps({
            'type': 'filequest',
            'nickname': self.__nickname,
            'filename': self.sendfilename,
            'filesize': self.sendfilesize,
            'who': self.fileto,
            'send_ip': '',
            'send_port': '',
        }).encode(), self.__host)
 
        print('request send...')
 
        # fileres = self.__socket.recvfrom(1024)[0].decode()
        # js = json.loads(fileres)
 
    def do_getfile(self, args):
        filename = args.split(' ')[0]
        who = args.split(' ')[1]
        ch = args.split(' ')[2]
        # print(self.filename is not None, filename, self.filename, who, self.filefrom)
        if (self.filename is not None) & (filename == self.filename) & (who == self.filefrom):
            if ch == 'yes':
                self.file_recv = open(self.filename, 'wb')
                self.__socket.sendto(json.dumps({
                    'type': 'fileres',
                    'fileres': 'yes',
                    'nickname': self.__nickname,
                    'who': who,
                    'recv_ip': '',
                    'recv_port': '',
                }).encode(), self.__host)
                print('you agree to reveive the file(', filename, ') from', who)
 
            else:
                self.__socket.sendto(json.dumps({
                    'type': 'fileres',
                    'fileres': 'no',
                    'nickname': self.__nickname,
                    'errormessage': 'deny the file.',
                    'who': who,
                    'recv_ip': '',
                    'recv_port': '',
                }).encode(), self.__host)
                print('you deny to reveive the file(', filename, ') from', who)
                self.recvfile = False
        else:
            print('the name or sender of the file is wrong.')
 
 
c = Client(('127.0.0.1', 12346))
c.start()
 

server：

import socket
import json
 
obj = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
obj.bind(('127.0.0.1', 12346))
 
conn_list = []
user_list = []
 
while True:
    try:
        receive_data, client_address = obj.recvfrom(1024)
        js = json.loads(receive_data.decode())
        # 登入訊息
        if js['type'] == 'login':
 
            nickname = str(js['nickname'])
            if nickname in user_list:
                obj.sendto(json.dumps({'login': 'fail',
                                       'errormessage': 'the nickname is exists'}).encode(),
                           client_address)
            else:
                # 向其他使用者傳送通知
                for i in range(len(conn_list)):
                    obj.sendto(json.dumps(
                        {
                            'type': 'message',
                            'message': '[system]' + nickname + '已登入.'
                        }).encode(), conn_list[i])
                user_list.append(nickname)
                conn_list.append(client_address)
                print(nickname, client_address, '登入成功！')
                obj.sendto(json.dumps({'login': 'success',
                                       'nickname': nickname}).encode(), client_address)
 
        # 群發訊息
        elif js['type'] == 'broadcast':
            message = js['message']
            nickname = js['nickname']
            for i in range(len(conn_list)):
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': nickname + ':' + message
                    }).encode(), conn_list[i])
 
        # 私發訊息
        elif js['type'] == 'sendto':
            who = js['who']
            nickname = js['nickname']
            message = js['message']
            # 檢查使用者是否存在
            if who not in user_list:
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': who + ' not exist or not online.please try later.'
                    }).encode(),
                    client_address)
            else:
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': nickname + ' whisper to you: ' + message
                    }).encode(),
                    conn_list[user_list.index(who)])
 
        # 檢視使用者列表
        elif js['type'] == 'catusers':
            users = json.dumps(user_list)
            obj.sendto(json.dumps(
                {
                    'type': 'message',
                    'message': users,
                }).encode(),
                client_address)
 
        # 檢視使用者IP
        elif js['type'] == 'catip':
            who = js['who']
            if who not in user_list:
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': who + ' not exist or not online.please try later.'
                    }).encode(),
                    client_address)
            else:
                addr = json.dumps(conn_list[user_list.index(who)])
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': addr,
                    }).encode(),
                    client_address)
 
        # 離線訊息
        elif js['type'] == 'offline':
            user_list.remove(js['nickname'])
            conn_list.remove(client_address)
            obj.sendto(
                (js['nickname'] + 'offline.').encode(),
                client_address)
            # 向其他使用者傳送通知
            for i in range(len(conn_list)):
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': '[system]' + nickname + '下線了.'
                    }).encode(), conn_list[i])
 
        # 傳送檔案請求
        elif js['type'] == 'filequest':
            who = js['who']
            if who not in user_list:
                obj.sendto(json.dumps(
                    {
                        'type': 'message',
                        'message': who + ' not exist or not online.please try later.'
                    }).encode(),
                    client_address)
            else:
                js['send_ip'] = client_address[0]
                js['send_port'] = client_address[1]
                obj.sendto(json.dumps(js).encode(),
                           conn_list[user_list.index(who)])
                print(js['nickname'], 'request to send file to', who)
 
        # 傳送檔案請求回覆
        elif js['type'] == 'fileres':
            who = js['who']
            if js['fileres'] == 'yes':
                js['recv_ip'] = client_address[0]
                js['recv_port'] = client_address[1]
                print(js['nickname'], 'agree to receive file from', js['who'])
            else:
                print(js['nickname'], 'deny to receive file from', js['who'])
            obj.sendto(json.dumps(js).encode(),
                       conn_list[user_list.index(who)])
 
    except Exception as e:
        print(e)