標籤：

用Python開發你的第一款聊天軟體

08-12
用Python開發你的第一款聊天軟體

來自專欄編程實驗室55 人贊了文章

在本實驗中，我們將實現一個簡單的圖形界面聊天系統。我們可以通過圖形客戶端登錄聊天室，並與其他成員進行聊天。

本教程由實驗樓120發布在實驗樓，完整教程、代碼及在線練習地址：Python 實現文字聊天室（更多項目請查看Python學習路徑）

一、實驗介紹

1、知識點

  • asyncore 、asynchat模塊使用
  • wxPython 圖形開發

2、實驗環境

  • python3.5

二、原理解析

由於 Python 是一門帶 GIL 的語言，所以在 Python 中使用多線程處理IO操作過多的任務並不是很好的選擇。同時聊天伺服器將同多個 socket 進行通信，所以我們可以基於 asyncore 模塊實現聊天伺服器。aysncore 模塊是一個非同步的 socket 處理器，通過使用該模塊將大大簡化非同步編程的難度。asynchat 模塊在 asyncore 模塊的基礎上做了進一步封裝，簡化了基於文本協議的通信任務的開發難度。

既然要開發聊天程序，那必然需要設計聊天時使用的協議。為了簡單起見，我們將要開發的聊天伺服器只支持文本協議，通過command message的方式調用相關的操作。比如如果客戶端發送以下文本，將執行相應的操作

# 登錄操作login
# 在聊天室中發表 hello 內容say hello
# 查看聊天室在線用戶look
# 退出登錄logout

以上協議流中，login, say, look, logout 就是相關協議代碼。

然後使用下面的命令在 /home/shiyanlou/Code 目錄下創建我們需要的 server.py 和 client.py 文件：

$ touch ~/Code/server.py$ touch ~/Code/client.py

三、伺服器類

這裡我們首先需要一個聊天伺服器類，通過繼承 asyncore 的 dispatcher 類來實現，我們編寫 server.py 文件：

import asynchatimport asyncore# 定義埠PORT = 6666# 定義結束異常類class EndSession(Exception): passclass ChatServer(asyncore.dispatcher): """ 聊天伺服器 """ def __init__(self, port): asyncore.dispatcher.__init__(self) # 創建socket self.create_socket() # 設置 socket 為可重用 self.set_reuse_addr() # 監聽埠 self.bind((, port)) self.listen(5) self.users = {} self.main_room = ChatRoom(self) def handle_accept(self): conn, addr = self.accept() ChatSession(self, conn)

這裡需要補充說明的是，對於 asyncore 和 asynchat 模塊來講，在 python3.6 中，使用 asyncio 模塊代替，但是實驗環境中我們使用的是 python 3.5 ，也由於 wxPython 對於Linux 下 CPython 的支持，所以我們依然使用 python 3.5。

1、會話類

有了伺服器類還需要能維護每個用戶的連接會話，這裡繼承 asynchat 的 async_chat 類來實現，在 server.py 文件中定義，代碼如下：

class ChatSession(asynchat.async_chat): """ 負責和客戶端通信 """ def __init__(self, server, sock): asynchat.async_chat.__init__(self, sock) self.server = server self.set_terminator(b
) self.data = [] self.name = None self.enter(LoginRoom(server)) def enter(self, room): # 從當前房間移除自身，然後添加到指定房間 try: cur = self.room except AttributeError: pass else: cur.remove(self) self.room = room room.add(self) def collect_incoming_data(self, data): # 接收客戶端的數據 self.data.append(data.decode("utf-8")) def found_terminator(self): # 當客戶端的一條數據結束時的處理 line = .join(self.data) self.data = [] try: self.room.handle(self, line.encode("utf-8")) # 退出聊天室的處理 except EndSession: self.handle_close() def handle_close(self): # 當 session 關閉時，將進入 LogoutRoom asynchat.async_chat.handle_close(self) self.enter(LogoutRoom(self.server))

2、協議命令解釋器

在之前的分析中，我們設計了聊天伺服器的協議，我們需要實現協議命令的相應方法，具體來說就是處理用戶登錄，退出，發消息，查詢在線用戶的代碼。在 server.py 文件中定義，

class CommandHandler: """ 命令處理類 """ def unknown(self, session, cmd): # 響應未知命令 # 通過 aynchat.async_chat.push 方法發送消息 session.push((Unknown command {}
.format(cmd)).encode("utf-8")) def handle(self, session, line): line = line.decode() # 命令處理 if not line.strip(): return parts = line.split( , 1) cmd = parts[0] try: line = parts[1].strip() except IndexError: line = # 通過協議代碼執行相應的方法 method = getattr(self, do_ + cmd, None) try: method(session, line) except TypeError: self.unknown(session, cmd)

3、房間

接下來就需要實現聊天室的房間了，這裡我們定義了三種房間，分別是用戶剛登錄時的房間、聊天的房間和退出登錄的房間，這三種房間都繼承自 CommandHandler，在 server.py 文件中定義，代碼如下：

class Room(CommandHandler): """ 包含多個用戶的環境，負責基本的命令處理和廣播 """ def __init__(self, server): self.server = server self.sessions = [] def add(self, session): # 一個用戶進入房間 self.sessions.append(session) def remove(self, session): # 一個用戶離開房間 self.sessions.remove(session) def broadcast(self, line): # 向所有的用戶發送指定消息 # 使用 asynchat.asyn_chat.push 方法發送數據 for session in self.sessions: session.push(line) def do_logout(self, session, line): # 退出房間 raise EndSessionclass LoginRoom(Room): """ 處理登錄用戶 """ def add(self, session): # 用戶連接成功的回應 Room.add(self, session) # 使用 asynchat.asyn_chat.push 方法發送數據 session.push(bConnect Success) def do_login(self, session, line): # 用戶登錄邏輯 name = line.strip() # 獲取用戶名稱 if not name: session.push(bUserName Empty) # 檢查是否有同名用戶 elif name in self.server.users: session.push(bUserName Exist) # 用戶名檢查成功後，進入主聊天室 else: session.name = name session.enter(self.server.main_room)class LogoutRoom(Room): """ 處理退出用戶 """ def add(self, session): # 從伺服器中移除 try: del self.server.users[session.name] except KeyError: passclass ChatRoom(Room): """ 聊天用的房間 """ def add(self, session): # 廣播新用戶進入 session.push(bLogin Success) self.broadcast((session.name + has entered the room.
).encode("utf-8")) self.server.users[session.name] = session Room.add(self, session) def remove(self, session): # 廣播用戶離開 Room.remove(self, session) self.broadcast((session.name + has left the room.
).encode("utf-8")) def do_say(self, session, line): # 客戶端發送消息 self.broadcast((session.name + : + line +
).encode("utf-8")) def do_look(self, session, line): # 查看在線用戶 session.push(bOnline Users:
) for other in self.sessions: session.push((other.name +
).encode("utf-8"))if __name__ == __main__: s = ChatServer(PORT) try: print("chat serve run at 0.0.0.0:{0}".format(PORT)) asyncore.loop() except KeyboardInterrupt: print("chat server exit")

四、登陸窗口

完成了伺服器端後，就需要實現客戶端了。客戶端將基於 wxPython 模塊實現。 wxPython 模塊是 wxWidgets GUI 工具的 Python 綁定。所以通過 wxPython 模塊我們就可以實現 GUI 編程了。同時我們的聊天協議基於文本，所以我們和伺服器之間的通信將基於 telnetlib 模塊實現。

登錄窗口通過繼承 wx.Frame 類來實現，編寫 client.py 文件，代碼如下：

import wximport telnetlibfrom time import sleepimport _thread as threadclass LoginFrame(wx.Frame): """ 登錄窗口 """ def __init__(self, parent, id, title, size): # 初始化，添加控制項並綁定事件 wx.Frame.__init__(self, parent, id, title) self.SetSize(size) self.Center() self.serverAddressLabel = wx.StaticText(self, label="Server Address", pos=(10, 50), size=(120, 25)) self.userNameLabel = wx.StaticText(self, label="UserName", pos=(40, 100), size=(120, 25)) self.serverAddress = wx.TextCtrl(self, pos=(120, 47), size=(150, 25)) self.userName = wx.TextCtrl(self, pos=(120, 97), size=(150, 25)) self.loginButton = wx.Button(self, label=Login, pos=(80, 145), size=(130, 30)) # 綁定登錄方法 self.loginButton.Bind(wx.EVT_BUTTON, self.login) self.Show() def login(self, event): # 登錄處理 try: serverAddress = self.serverAddress.GetLineText(0).split(:) con.open(serverAddress[0], port=int(serverAddress[1]), timeout=10) response = con.read_some() if response != bConnect Success: self.showDialog(Error, Connect Fail!, (200, 100)) return con.write((login + str(self.userName.GetLineText(0)) +
).encode("utf-8")) response = con.read_some() if response == bUserName Empty: self.showDialog(Error, UserName Empty!, (200, 100)) elif response == bUserName Exist: self.showDialog(Error, UserName Exist!, (200, 100)) else: self.Close() ChatFrame(None, 2, title=ShiYanLou Chat Client, size=(500, 400)) except Exception: self.showDialog(Error, Connect Fail!, (95, 20)) def showDialog(self, title, content, size): # 顯示錯誤信息對話框 dialog = wx.Dialog(self, title=title, size=size) dialog.Center() wx.StaticText(dialog, label=content) dialog.ShowModal()

1、聊天窗口

聊天窗口中最主要的就是向伺服器發消息並接受伺服器的消息，這裡通過子線程來接收消息，在 client.py 文件中定義，代碼如下：

class ChatFrame(wx.Frame): """ 聊天窗口 """ def __init__(self, parent, id, title, size): # 初始化，添加控制項並綁定事件 wx.Frame.__init__(self, parent, id, title) self.SetSize(size) self.Center() self.chatFrame = wx.TextCtrl(self, pos=(5, 5), size=(490, 310), stylex=wx.TE_MULTILINE | wx.TE_READONLY) self.message = wx.TextCtrl(self, pos=(5, 320), size=(300, 25)) self.sendButton = wx.Button(self, label="Send", pos=(310, 320), size=(58, 25)) self.usersButton = wx.Button(self, label="Users", pos=(373, 320), size=(58, 25)) self.closeButton = wx.Button(self, label="Close", pos=(436, 320), size=(58, 25)) # 發送按鈕綁定發送消息方法 self.sendButton.Bind(wx.EVT_BUTTON, self.send) # Users按鈕綁定獲取在線用戶數量方法 self.usersButton.Bind(wx.EVT_BUTTON, self.lookUsers) # 關閉按鈕綁定關閉方法 self.closeButton.Bind(wx.EVT_BUTTON, self.close) thread.start_new_thread(self.receive, ()) self.Show() def send(self, event): # 發送消息 message = str(self.message.GetLineText(0)).strip() if message != : con.write((say + message +
).encode("utf-8")) self.message.Clear() def lookUsers(self, event): # 查看當前在線用戶 con.write(blook
) def close(self, event): # 關閉窗口 con.write(blogout
) con.close() self.Close() def receive(self): # 接受伺服器的消息 while True: sleep(0.6) result = con.read_very_eager() if result != : self.chatFrame.AppendText(result)if __name__ == __main__: app = ICANN Verification Required() con = telnetlib.Telnet() LoginFrame(None, -1, title="Login", size=(320, 250)) app.MainLoop()

五、執行

  • 首先，我們執行 server.py ，如下圖所示：

  • 這時，我們再打開一個終端，執行 client.py 文件，如下圖：

  • 輸入對應的信息之後，點擊 Login ，再次重複上一步驟，使用另一用戶名 shiyanlou002登陸，如下圖:

  • 在最終的示例中，我們可以分別通過 shiyanlou001 和 shiyanlou002 的客戶端發送消息，此時，所有的在線用戶都可以收到對應的消息。

六、小結

最後就可以運行程序進行聊天了，注意需要先啟動伺服器再啟動客戶端。這個項目中使用了 asyncore 的 dispatcher 來實現伺服器，asynchat 的 asyn_chat 來維護用戶的連接會話，用 wxPython 來實現圖形界面，用 telnetlib 來連接伺服器，在子線程中接收伺服器發來的消息，由此一個簡單的聊天室程序就完成了。

詳細教程在Python 實現文字聊天室

更多編程練手項目：全部課程

推薦閱讀：

※喜歡玩遊戲的孩子適合學習編程嗎？
※python 正則表達式
※Ruby 又要添加綠色線程了 Thread::Green
※INCA二次開發-INCACOM

TAG:Python | Python開發 | 編程 |