python實(shí)現(xiàn)多人聊天室
更新時(shí)間：2020年03月31日 16:42:29 作者：DOLFAMINGO
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)多人聊天室，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下
本文實(shí)例為大家分享了python實(shí)現(xiàn)多人聊天室的具體代碼，供大家參考，具體內(nèi)容如下
一、目的
以實(shí)現(xiàn)小項(xiàng)目的方式，來(lái)鞏固之前學(xué)過(guò)的Python基本語(yǔ)法以及相關(guān)的知識(shí)。
二、相關(guān)技術(shù)
1.wxpython GUI編程
2.網(wǎng)絡(luò)編程
3.多線程編程
4.數(shù)據(jù)庫(kù)編程
5.簡(jiǎn)單的將數(shù)據(jù)導(dǎo)出到Excel表
三、存在的漏洞以及不足
1.由于數(shù)據(jù)庫(kù)編碼的問(wèn)題，無(wú)法使用中文。
2.在客戶端關(guān)閉后，其相關(guān)的線程仍然存在于服務(wù)器的用戶線程隊(duì)列中，所以服務(wù)器會(huì)錯(cuò)誤地往已關(guān)閉的客戶端傳送信息。
3.客戶端初始登錄并加載歷史記錄時(shí)，會(huì)出現(xiàn)每條歷史消息后面的回車鍵丟失的現(xiàn)象，解決的方法是：在加載相鄰兩條消息之間加個(gè)時(shí)間間隔，但效果不佳。
四、源碼
服務(wù)器Server：
# -*- coding: UTF-8 -*-
from socket import *
import time
import threading
import wx
import MySQLdb
import xlwt
from clientthread import ClientThread
class Server(wx.Frame):
def __init__(self,parent=None,id=-1,title='服務(wù)器',pos=wx.DefaultPosition,size=(500,300)):
'''窗口'''
wx.Frame.__init__(self,parent,id,title,pos,size=(400,470))
pl = wx.Panel(self)
con = wx.BoxSizer(wx.VERTICAL)
subcon = wx.FlexGridSizer(wx.HORIZONTAL)
sta = wx.Button(pl , size=(133, 40),label='啟動(dòng)服務(wù)器')
end = wx.Button(pl, size=(133, 40), label='關(guān)閉服務(wù)器')
hist = wx.Button(pl,size=(133,40),label='導(dǎo)出聊天記錄')
subcon.Add(sta, 1, wx.BOTTOM)
subcon.Add(hist, 1, wx.BOTTOM)
subcon.Add(end, 1, wx.BOTTOM)
con.Add(subcon,1,wx.ALIGN_CENTRE|wx.BOTTOM)
self.Text = wx.TextCtrl(pl, size=(400,250),style = wx.TE_MULTILINE|wx.TE_READONLY)
con.Add(self.Text, 1, wx.ALIGN_CENTRE)
self.ttex = wx.TextCtrl(pl, size=(400,100),style=wx.TE_MULTILINE)
con.Add(self.ttex, 1, wx.ALIGN_CENTRE)
sub2 = wx.FlexGridSizer(wx.HORIZONTAL)
clear = wx.Button(pl, size=(200, 40), label='清空')
send = wx.Button(pl, size=(200, 40), label='發(fā)送')
sub2.Add(clear, 1, wx.TOP | wx.LEFT)
sub2.Add(send, 1, wx.TOP | wx.RIGHT)
con.Add(sub2, 1, wx.ALIGN_CENTRE)
pl.SetSizer(con)
'''窗口'''
'''綁定'''
self.Bind(wx.EVT_BUTTON, self.EditClear, clear)
self.Bind(wx.EVT_BUTTON, self.SendMessage, send)
self.Bind(wx.EVT_BUTTON, self.Start, sta)
self.Bind(wx.EVT_BUTTON, self.Break, end)
self.Bind(wx.EVT_BUTTON, self.WriteToExcel, hist)
'''綁定'''
'''服務(wù)器準(zhǔn)備工作'''
self.UserThreadList = []
self.onServe = False
addr = ('', 21567)
self.ServeSock = socket(AF_INET, SOCK_STREAM)
self.ServeSock.bind(addr)
self.ServeSock.listen(10)
'''服務(wù)器準(zhǔn)備工作'''
'''數(shù)據(jù)庫(kù)準(zhǔn)備工作，用于存儲(chǔ)聊天記錄'''
self.db = MySQLdb.connect('localhost', 'root', '123456', 'user_info')
self.cursor = self.db.cursor()
self.cursor.execute("select * from history order by time")
self.Text.SetValue('')
for data in self.cursor.fetchall(): #加載歷史聊天記錄
self.Text.AppendText('%s said:\n%s\nwhen %s\n\n' % (data[0], data[2], data[1]))
'''數(shù)據(jù)庫(kù)準(zhǔn)備工作，用于存儲(chǔ)聊天記錄'''
#將聊天記錄導(dǎo)出到EXCEl表中
def WriteToExcel(self,event):
wbk = xlwt.Workbook()
sheet = wbk.add_sheet('sheet 1')
self.cursor.execute("select * from history order by time")
sheet.write(0, 0, "User")
sheet.write(0, 1, "Datetime")
sheet.write(0, 5, "Message")
index = 0
for data in self.cursor.fetchall():
index = index + 1
Time = '%s'%data[1] #將datetime轉(zhuǎn)成字符形式，否則直接寫(xiě)入Excel會(huì)變成時(shí)間戳
sheet.write(index,0,data[0])
sheet.write(index,1,Time) #寫(xiě)進(jìn)EXCEL會(huì)變成時(shí)間戳
sheet.write(index,5,data[2])
wbk.save(r'D:\History_Dialog.xls')
#啟動(dòng)服務(wù)器的服務(wù)線程
def Start(self,event):
if not self.onServe:
'''啟動(dòng)服務(wù)線程'''
self.onServe = True
mainThread = threading.Thread(target=self.on_serving, args=())
mainThread.setDaemon(True) # 解決父線程結(jié)束，子線程還繼續(xù)運(yùn)行的問(wèn)題
mainThread.start()
'''啟動(dòng)服務(wù)線程'''
#關(guān)閉服務(wù)器
def Break(self,event):
self.onServe = False
#服務(wù)器主循環(huán)
def on_serving(self):
print '...On serving...'
while self.onServe:
UserSocket, UserAddr = self.ServeSock.accept()
username = UserSocket.recv(1024).decode(encoding='utf-8') #接收用戶名
userthread = ClientThread(UserSocket, username,self)
self.UserThreadList.append(userthread) #將用戶線程加到隊(duì)列中
userthread.start()
self.ServeSock.close()
#綁定發(fā)送按鈕
def SendMessage(self,event):
if self.onServe and cmp(self.ttex.GetValue(),''):
data = self.ttex.GetValue()
self.AddText('Server',data,time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
self.ttex.SetValue('')
# 向所有客戶端（包括自己）發(fā)送信息,同時(shí)更新到數(shù)據(jù)庫(kù)
def AddText(self, source, data,Time):
self.cursor.execute("insert into history values(\"%s\",\"%s\",\"%s\")" % (source,Time,data)) #雙引號(hào)里面有雙引號(hào)，bug：句子不能有雙引號(hào)、以及中文
self.db.commit()
sendData = '%s said:\n%s\nwhen %s\n' % (source,data,Time)
self.Text.AppendText('%s\n'%sendData)
for user in self.UserThreadList: #bug：客戶端關(guān)閉了仍然在隊(duì)列中。如果客戶端關(guān)閉了，那怎么在服務(wù)器判斷是否已經(jīng)關(guān)閉了？客戶端在關(guān)閉之前發(fā)一條信息給服務(wù)器？
user.UserSocket.send(sendData.encode(encoding='utf-8'))
#綁定清空按鈕
def EditClear(self,event):
self.ttex.Clear()
def main():
app = wx.App(False)
Server().Show()
app.MainLoop()
if __name__ == '__main__':
main()
服務(wù)器的客戶線程Clientthread：
# -*- coding: UTF-8 -*-
import threading
import time
class ClientThread(threading.Thread):
def __init__(self,UserSocket, Username,server):
threading.Thread.__init__(self)
self.UserSocket = UserSocket
self.Username = Username
self.server = server
self.Loadhist()
# 加載歷史聊天記錄
def Loadhist(self):
self.server.cursor.execute("select * from history order by time")
for data in self.server.cursor.fetchall():
time.sleep(0.6) #幾條信息同時(shí)發(fā)，會(huì)造成末尾回車鍵的丟失，所以要有時(shí)間間隔
sendData = '%s said:\n%s\nwhen %s\n'%(data[0], data[2], data[1])
self.UserSocket.send(sendData.encode(encoding='utf-8'))
#方法重寫(xiě)，線程的入口
def run(self):
size = 1024
while True:
data = self.UserSocket.recv(size) #未解決：客戶端斷開(kāi)連接后這里會(huì)報(bào)錯(cuò)
self.server.AddText(self.Username,data.decode(encoding='utf-8'),time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
self.UserSocket.close() #這里都執(zhí)行不到
客戶登錄界面Logframe：
# -*- coding: UTF-8 -*-
from socket import *
import wx
import MySQLdb
from client import Client
class LogFrame(wx.Frame):
def __init__(self,parent=None,id=-1,title='登錄窗口',pos=wx.DefaultPosition,size=(500,300)):
'''窗口'''
wx.Frame.__init__(self,parent,id,title,pos,size=(400,280))
self.pl = wx.Panel(self)
con = wx.BoxSizer(wx.VERTICAL)
subcon = wx.FlexGridSizer(2,2,10,10)
username = wx.StaticText(self.pl, label="Username:",style=wx.ALIGN_LEFT)
password = wx.StaticText(self.pl, label="Password:",style=wx.ALIGN_LEFT)
self.tc1 = wx.TextCtrl(self.pl,size=(180,20))
self.tc2 = wx.TextCtrl(self.pl,size=(180,20),style=wx.TE_PASSWORD)
subcon.Add(username,wx.TE_LEFT)
subcon.Add(self.tc1,1,wx.EXPAND)
subcon.Add(password)
subcon.Add(self.tc2,1,wx.EXPAND)
con.Add(subcon,1,wx.ALIGN_CENTER)
subcon2 = wx.FlexGridSizer(1,2,10,10)
register = wx.Button(self.pl,label='Register')
login = wx.Button(self.pl,label='Login')
subcon2.Add(register,1, wx.TOP)
subcon2.Add(login,1, wx.TOP)
con.Add(subcon2,1,wx.ALIGN_CENTRE)
self.pl.SetSizer(con)
self.Bind(wx.EVT_BUTTON,self.Register,register)
self.Bind(wx.EVT_BUTTON,self.Login,login)
'''窗口'''
self.isConnected = False
self.userSocket = None
#連接到服務(wù)器
def ConnectToServer(self):
if not self.isConnected:
ADDR = ('localhost', 21567)
self.userSocket = socket(AF_INET, SOCK_STREAM)
try:
self.userSocket.connect(ADDR)
self.userSocket.send(self.tc1.GetValue().encode(encoding='utf-8'))
self.isConnected = True
return True
except Exception:
return False
else:
return True
#登錄
def Login(self,event):
if not self.ConnectToServer():
err = wx.MessageDialog(None, '服務(wù)器未啟動(dòng)', 'ERROR!', wx.OK)
err.ShowModal()
err.Destroy()
else:
username = self.tc1.GetValue()
password = self.tc2.GetValue()
db = MySQLdb.connect('localhost', 'root', '123456', 'user_info')
cursor = db.cursor()
cursor.execute("select * from user_list where username='%s' and password='%s'"%(username,password))
if not cursor.fetchone():
err = wx.MessageDialog(None,'用戶不存在或密碼錯(cuò)誤','ERROR!',wx.OK)
err.ShowModal()
else:
self.Close()
Client(opSock=self.userSocket, username=username).Show()
db.commit()
db.close()
#注冊(cè)
def Register(self,event):
if not self.ConnectToServer():
err = wx.MessageDialog(None, '服務(wù)器未啟動(dòng)', 'ERROR!', wx.OK)
err.ShowModal()
err.Destroy()
else:
username = self.tc1.GetValue()
password = self.tc2.GetValue()
db = MySQLdb.connect('localhost', 'root', '123456', 'user_info')
cursor = db.cursor()
cursor.execute("select * from user_list where username='%s'"%username)
if not cursor.fetchone():
cursor.execute("insert into user_list(username,password) values('%s','%s')"%(username,password))
else:
err = wx.MessageDialog(None, '用戶已存在', 'ERROR!', wx.OK)
err.ShowModal()
db.commit()
db.close()
def main():
app = wx.App(False)
LogFrame().Show()
app.MainLoop()
if __name__ == '__main__':
main()
客戶端Client：
#/usr/bin/env python
# -*- coding: UTF-8 -*-
import wx
import threading
from time import ctime
class Client(wx.Frame):
def __init__(self,opSock,username,parent=None,id=-1,title='客戶端',pos=wx.DefaultPosition,size=(500,300)):
'''窗口'''
wx.Frame.__init__(self,parent,id,title,pos,size=(400,470))
self.opSock = opSock
self.username = username
pl = wx.Panel(self)
con = wx.BoxSizer(wx.VERTICAL)
subcon = wx.FlexGridSizer(wx.HORIZONTAL)
sta = wx.Button(pl, size=(200, 40),label='連接')
end = wx.Button(pl, size=(200, 40),label='斷開(kāi)')
subcon.Add(sta, 1, wx.TOP|wx.LEFT)
subcon.Add(end, 1, wx.TOP|wx.RIGHT)
con.Add(subcon,1,wx.ALIGN_CENTRE)
self.Text = wx.TextCtrl(pl, size=(400,250),style = wx.TE_MULTILINE|wx.TE_READONLY)
con.Add(self.Text, 1, wx.ALIGN_CENTRE)
self.ttex = wx.TextCtrl(pl, size=(400,100),style=wx.TE_MULTILINE)
con.Add(self.ttex, 1, wx.ALIGN_CENTRE)
sub2 = wx.FlexGridSizer(wx.HORIZONTAL)
clear = wx.Button(pl, size=(200, 40), label='清空')
send = wx.Button(pl, size=(200, 40), label='發(fā)送')
sub2.Add(clear, 1, wx.TOP | wx.LEFT)
sub2.Add(send, 1, wx.TOP | wx.RIGHT)
con.Add(sub2, 1, wx.ALIGN_CENTRE)
pl.SetSizer(con)
'''窗口'''
'''綁定'''
self.Bind(wx.EVT_BUTTON, self.EditClear, clear)
self.Bind(wx.EVT_BUTTON, self.Send, send)
self.Bind(wx.EVT_BUTTON, self.Login, sta)
self.Bind(wx.EVT_BUTTON, self.Logout, end)
'''綁定'''
self.isConnected = False
#登錄
def Login(self,event):
'''客戶端準(zhǔn)備工作'''
self.isConnected = True
t = threading.Thread(target=self.Receive, args=())
t.setDaemon(True)
t.start()
'''客戶端準(zhǔn)備工作'''
#退出
def Logout(self,event):
self.isConnected = False
#綁定發(fā)送按鈕
def Send(self,event):
if self.isConnected and cmp(self.ttex.GetValue(),''):
self.opSock.send(self.ttex.GetValue().encode(encoding='utf-8'))
self.ttex.SetValue('')
#綁定清空按鈕
def EditClear(self,event):
self.ttex.Clear()
#接收客戶端的信息（獨(dú)立一個(gè)線程）
def Receive(self):
while self.isConnected:
data = self.opSock.recv(1024).decode(encoding='utf-8')
self.Text.AppendText('%s\n'%data)
更多關(guān)于python聊天功能的精彩文章請(qǐng)點(diǎn)擊專題： python聊天功能匯總
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
python開(kāi)啟多個(gè)子進(jìn)程并行運(yùn)行的方法
這篇文章主要介紹了python開(kāi)啟多個(gè)子進(jìn)程并行運(yùn)行的方法,涉及Python進(jìn)程操作的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-04-04
python相對(duì)企業(yè)語(yǔ)言優(yōu)勢(shì)在哪
在本篇文章里小編給大家分享的是關(guān)于python相對(duì)企業(yè)語(yǔ)言優(yōu)勢(shì)以及相關(guān)知識(shí)點(diǎn)，需要的朋友們可以參考下。2020-06-06
這篇文章主要介紹了Python 合并拼接字符串的方法，文中講解非常細(xì)致，代碼幫助大家更好的理解和學(xué)習(xí)，感興趣的朋友可以了解下2020-07-07
python的命名規(guī)則知識(shí)點(diǎn)總結(jié)
在本篇文章里小編給大家分享的是關(guān)于python的命名規(guī)則知識(shí)點(diǎn)總結(jié)，有需要的朋友們可以參考下。2019-10-10
python兒童學(xué)游戲編程知識(shí)點(diǎn)總結(jié)
在本文里小編給大家整理了關(guān)于python兒童學(xué)游戲編程知識(shí)點(diǎn)以及內(nèi)容總結(jié)，需要的朋友們參考學(xué)習(xí)下。2019-06-06
python腳本監(jiān)控logstash進(jìn)程并郵件告警實(shí)例
這篇文章主要介紹了python腳本監(jiān)控logstash進(jìn)程并郵件告警實(shí)例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-04-04
比較詳細(xì)Python正則表達(dá)式操作指南(re使用)
Python 1.5之前版本則是通過(guò) regex 模塊提供 Emecs 風(fēng)格的模式。Emacs 風(fēng)格模式可讀性稍差些，而且功能也不強(qiáng)，因此編寫(xiě)新代碼時(shí)盡量不要再使用 regex 模塊，當(dāng)然偶爾你還是可能在老代碼里發(fā)現(xiàn)其蹤影2008-09-09
這篇文章主要為大家介紹了python生成驗(yàn)證碼圖片代碼，生成原理是將一串隨機(jī)產(chǎn)生的數(shù)字或符號(hào)，生成一幅圖片，圖片里加上一些干擾象素，想要實(shí)現(xiàn)驗(yàn)證碼圖片的朋友可以參考一下2016-01-01
這篇文章主要介紹了在Django的模型中執(zhí)行原始SQL查詢的方法,Django是最具人氣的Python web開(kāi)發(fā)框架,需要的朋友可以參考下2015-07-07
深度解析Django REST Framework 批量操作
這篇文章主要介紹了深度解析Django REST Framework批量操作，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-05-05