目录
简介
运行效果
代码
客户端
服务器
自定义库
通讯协议及相关配置定义库(DY.py)
数据库调用库（SJK.py）
套接字库（TJZ.py）
套接字通讯库（服务器端）(TJZ_FZ.py)
简介
一个用python写简易的聊天室程序，拥有登录、注册、找回密码、聊天功能，采用TPC通讯，无管理员功能，修改密码功能。代码正常运行需要安装Mysql数据库，请在代码前在Mysql数据库中创建一个用户，用户名为：root 密码为：123456。代码含详细注释，在此就不多赘述
运行效果
代码
客户端
调用自定义库DY.py文章来源：https://www.toymoban.com/news/detail-485707.html
import socket
from DY import *
import tkinter as tk
from tkinter.scrolledtext import ScrolledText
import threading
import sys
#创建窗口
class talkroom_system(object):
def __init__(self):
#窗口模块
self.tk=tk
#将相应函数放入字典（窗口函数调用）
self.msg_denglu={}
self.room_all={START_login:self.room_denglu,START_zhuce:self.room_zhuce,
START_zh:self.room_zhaohui,START_zh_end:self.room_zhaohui_end,
START_talk:self.room_laitian,START_zh_ok:self.room_get}
#参数
self.i=0
self.n=0
#将不同函数与相对应的命令一一对应
self.next_step={START_login:{START_login_yes:self.room_change,START_login_no:self.denglu_fail},
START_zhuce:{START_zhuce_yes:self.zhuce_result,START_zhuce_no:self.zhuce_result},
START_zh:{START_n_yes:self.room_change,START_n_no:self.zh_name_no}}
#数据存储
self.user_self=[]
self.user_qaa=[]
#连接服务器
def connecting(self):
connect_room=self.tk.Tk()
connect_room.title(TITLE)
connect_room.geometry("500x120+500+270")
lb_connect=self.tk.Label(connect_room,text='连接失败.........',font=('黑体',20)).place(x=50,y=40,width=400,height=30)
#尝试连接
while self.i==0:
try:
self.TJZ=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
self.TJZ.connect((MASTER_IP,9999))
self.i=1
connect_room.destroy()
except:
print('连接中...')
self.n+=1
if self.n==10:
connect_room.mainloop()
sys.exit()
#启动登录窗口
self.room_denglu()
#登录窗口
def room_denglu(self):
#窗口建立
denglu_room=self.tk.Tk()
denglu_room.title(TITLE)
denglu_room.geometry("500x210+500+270")
#文本设置
lb_name=self.tk.Label(denglu_room,text='用户名').place(x=50,y=40,width=50,height=30)
lb_mm=self.tk.Label(denglu_room,text='密码').place(x=50,y=90,width=50,height=30)
#输入框
tt_name=self.tk.Entry(denglu_room)
tt_name.place(x=110,y=40,width=300,height=30)
tt_mm=self.tk.Entry(denglu_room)
tt_mm.place(x=110,y=90,width=300,height=30)
data_denglu=[START_login,tt_name,tt_mm]
#按钮设置
bn_denglu=self.tk.Button(denglu_room,text='登录',command=lambda: self.msg_get(data_denglu,denglu_room))
bn_denglu.place(x=125,y=150,width=100,height=30)
bn_zhuce=self.tk.Button(denglu_room,text='注册',command=lambda:self.room_change(denglu_room,START_zhuce)).place(x=275,y=150,width=100,height=30)
bn_zhaohui=self.tk.Button(denglu_room,text='找回密码',command=lambda:self.room_change(denglu_room,START_zh)).place(x=370,y=125,width=50,height=20)
#保持窗口
denglu_room.mainloop()
#注册窗口
def room_zhuce(self):
#创建窗口
zhuce_room=self.tk.Tk()
zhuce_room.title(TITLE)
zhuce_room.geometry("500x350+500+270")
#文本显示
lb_name=self.tk.Label(zhuce_room,text='用户名').place(x=50,y=40,width=50,height=30)
lb_mm=self.tk.Label(zhuce_room,text='密码').place(x=50,y=90,width=50,height=30)
lb_mm=self.tk.Label(zhuce_room,text='再次输入密码').place(x=50,y=140,width=50,height=30)
lb_mm=self.tk.Label(zhuce_room,text='问题').place(x=50,y=190,width=50,height=30)
lb_mm=self.tk.Label(zhuce_room,text='答案').place(x=50,y=240,width=50,height=30)
#输入框
tt_name=self.tk.Entry(zhuce_room)
tt_name.place(x=110,y=40,width=300,height=30)
tt_mm_1=self.tk.Entry(zhuce_room)
tt_mm_1.place(x=110,y=90,width=300,height=30)
tt_mm_2=self.tk.Entry(zhuce_room)
tt_mm_2.place(x=110,y=140,width=300,height=30)
tt_question=self.tk.Entry(zhuce_room)
tt_question.place(x=110,y=190,width=300,height=30)
tt_answer=self.tk.Entry(zhuce_room)
tt_answer.place(x=110,y=240,width=300,height=30)
#按钮设置
data_zhuce=[START_zhuce,tt_name,tt_mm_1,tt_mm_2,tt_question,tt_answer]
bn_yes=self.tk.Button(zhuce_room,text='确认',command=lambda:self.msg_get(data_zhuce,zhuce_room))
bn_yes.place(x=320,y=290,width=100,height=30)
#保持窗口
zhuce_room.mainloop()
#找回窗口1
def room_zhaohui(self):
#建立窗口
zhaohui_room=self.tk.Tk()
zhaohui_room.title(TITLE)
zhaohui_room.geometry("500x120+500+270")
#文本显示
lb_name=self.tk.Label(zhaohui_room,text='用户名').place(x=50,y=40,width=50,height=30)
#输入框
tt_name=self.tk.Entry(zhaohui_room)
tt_name.place(x=110,y=40,width=300,height=30)
#按钮设置
data=[START_zh,tt_name]
bn_yes=self.tk.Button(zhaohui_room,text='确认',command=lambda:self.msg_get(data,zhaohui_room))
bn_yes.place(x=350,y=80,width=90,height=28)
#保持窗口
zhaohui_room.mainloop()
#找回窗口2
def room_zhaohui_end(self):
#建立窗口
zh_end_room=self.tk.Tk()
zh_end_room.title(TITLE)
zh_end_room.geometry("500x120+500+270")
#文本显示
qst='问题:'+str(self.user_qaa[3])
lb_qst=self.tk.Label(zh_end_room,text=qst,justify='left').place(x=-62,y=20,width=300,height=20)
lb_ans=self.tk.Label(zh_end_room,text='答案').place(x=50,y=40,width=50,height=30)
#输入框
tt_ans=self.tk.Entry(zh_end_room)
tt_ans.place(x=110,y=40,width=300,height=30)
#按钮设置
data=[START_zh_end,tt_ans]
bn_yes=self.tk.Button(zh_end_room,text='确认',command=lambda:self.msg_get(data,zh_end_room))
bn_yes.place(x=350,y=80,width=90,height=28)
#保持窗口
zh_end_room.mainloop()
#找回成功窗口
def room_get(self):
#建立窗口
zh_end_room=self.tk.Tk()
zh_end_room.title(TITLE)
zh_end_room.geometry("500x120+500+270")
#文本显示
qst='用户名:'+self.user_qaa[0]
ans='密码：'+self.user_qaa[1]
lb_qst=self.tk.Label(zh_end_room,text=qst,font=('黑体',20))
lb_qst.place(x=0,y=20,width=300,height=30)
lb_ans=self.tk.Label(zh_end_room,text=ans,font=('黑体',20))
lb_ans.place(x=0,y=50,width=300,height=30)
#按钮设置
bn_yes=self.tk.Button(zh_end_room,text='返回',command=lambda:self.room_change(zh_end_room,START_login))
bn_yes.place(x=350,y=80,width=90,height=28)
#保持窗口
zh_end_room.mainloop()
#聊天室
def room_laitian(self):
#窗口初始化
liaotian_room=self.tk.Tk()
liaotian_room.title(TITLE)
liaotian_room.geometry("750x450+400+180")
#滚动文本聊天框
tt_sc=ScrolledText(liaotian_room,font=('黑体',12))
tt_sc.place(x=0,y=0,width=600,height=350)
tt_sc.tag_config('tag1', foreground='blue')
tt_sc.insert(self.tk.END, '欢迎进入群聊，大家开始聊天吧!', 'tag1')
shuru=self.tk.StringVar()
shuru.set('')
#输入框
tt_sr=self.tk.Entry(liaotian_room,textvariable=shuru,font=('黑体',12))
tt_sr.place(x=0,y=350,width=600,height=100)
#列表框
tt_user=self.tk.Listbox(liaotian_room)
tt_user.place(x=600,y=0,width=150,height=350)
#按钮设置
bt_sr=self.tk.Button(liaotian_room,text='发送',font=('黑体',20),command=lambda:self.msg_send(tt_sr,shuru))
bt_sr.place(x=600,y=350,width=150,height=100)
self.dxc_start(tt_sc,tt_user)
#保持窗口
liaotian_room.mainloop()
#获取信息
def msg_get(self,data,room):
#分割信息
check=0
msg_get_ls=[]
for i in data:
try:
msg_get_ls.append(i.get())
except:
msg_get_ls.append(i)
#判断个别进程选项
if msg_get_ls[0]==START_zhuce:
if msg_get_ls[2]==msg_get_ls[3]:
del msg_get_ls[2]
else:
lb_fail=self.tk.Label(room,text='两次输入密码不同！！！',justify='left',).place(x=80,y=270,width=200,height=20)
check=1
elif msg_get_ls[0]==START_zh_end:
if msg_get_ls[1]==self.user_qaa[3]:
self.room_change(room,START_zh_ok)
check=1
else:
lb_fail=self.tk.Label(room,text='答案错误！！！',justify='left',).place(x=50,y=70,width=200,height=20)
check=1
#进入所匹配函数
if check==0:
self.msg_use(msg_get_ls,room)
#窗口跳转
def room_change(self,room_now,room):
room_now.destroy()
room_next=self.room_all[room]
try:
room_next()
except:
return
#使用信息
def msg_use(self,data_end,room):
#判断传递信息类型，按协议处理数据
if data_end[0]==START_login:
self.user_self.append(data_end[1])
msg_send=data_end[0]+SHOW_differ+data_end[1]+SHOW_differ+data_end[2]
elif data_end[0]==START_zhuce:
msg_send=data_end[0]+SHOW_differ+data_end[1]+SHOW_differ+data_end[2]+SHOW_differ+data_end[3]+SHOW_differ+data_end[4]
elif data_end[0]==START_zh:
msg_send=data_end[0]+SHOW_differ+data_end[1]
#编码数据信息并发送至服务器
self.TJZ.send(msg_send.encode(SHOW_change))
recv_data=self.TJZ.recv(256).decode(SHOW_change)
self.msg_recv(data_end[0],recv_data,room)
#返回数据处理
def msg_recv(self,request,recv_datas,room):
#分割数据信息
datas=recv_datas.split(SHOW_differ)
recv_data=datas[0]
next_step=self.next_step[request][recv_data]
#判断信息类型并处理信息
#登录成功
if recv_data==START_login_yes:
next_step(room,START_talk)
msg_end=START_leave
self.TJZ.send(msg_end.encode(SHOW_change))
#登录失败
elif recv_data==START_login_no:
next_step(room)
del self.user_self[0]
#注册成功
elif recv_data==START_zhuce_yes:
next_step(recv_data,room)
self.room_change(room,START_login)
#注册失败
elif recv_data==START_zhuce_no:
next_step(recv_data,room)
#找回用户名存在
elif recv_data==START_n_yes:
self.user_qaa.append(datas[1])
self.user_qaa.append(datas[2])
self.user_qaa.append(datas[3])
self.user_qaa.append(datas[4])
next_step(room,START_zh_end)
#找回用户名不存在
elif recv_data==START_n_no:
next_step(room)
#登录失败
def denglu_fail(self,room):
lb_fail=self.tk.Label(room,text='用户名或密码错误！！！',justify='left',).place(x=80,y=120,width=200,height=20)
#注册结果
def zhuce_result(self,result,room):
if result==START_zhuce_yes:
lb_fail=self.tk.Label(room,text='注册成功！！！',justify='left',).place(x=80,y=270,width=200,height=20)
elif result==START_zhuce_no:
lb_fail=self.tk.Label(room,text='用户名已存在！！！',justify='left',).place(x=80,y=270,width=200,height=20)
#找回失败1
def zh_name_no(self,room):
lb_fail=self.tk.Label(room,text='用户名不存在！！！',justify='left',).place(x=65,y=70,width=200,height=20)
#发送聊天消息
def msg_send(self,tt_sr,shuru):
msg_get=START_talk+SHOW_differ+self.user_self[0]+SHOW_differ+tt_sr.get()
self.TJZ.send(msg_get.encode(SHOW_change))
shuru.set('')
#处理接受的与聊天相关的消息
def msg_last(self,tt_sc,tt_user):
#解码信息
msg_new=self.TJZ.recv(256).decode(SHOW_change)
#分割信息
msg_chuli=msg_new.split(SHOW_differ)
#判断聊天信息类型
#用户列表
if msg_chuli[0]==START_talk_users:
tt_user.delete(0,self.tk.END)
for i in range(1,len(msg_chuli)):
tt_user.insert(self.tk.END,msg_chuli[i])
#聊天信息
elif msg_chuli[0]==START_talk_msg:
user_name=msg_chuli[1]
user_msg=msg_chuli[2]
if user_name==self.user_self[0]:
user_name=USER_self
msg_last=user_name+':'+user_msg
tt_sc.insert(self.tk.INSERT,'\n')
tt_sc.insert(self.tk.END,msg_last)
#新用户进入聊天室
elif msg_chuli[0]==START_user_in:
msg_last=msg_chuli[1]+'进入了聊天室....'
tt_sc.insert(self.tk.INSERT,'\n')
tt_sc.insert(self.tk.END,msg_last)
#客户端离线
elif msg_chuli[0]==START_leave_yes:
sys.exit()
#多线程准备
def chixu(self,tt_sc,tt_user):
try:
while True:
self.msg_last(tt_sc,tt_user)
except:
sys.exit()
#多线程
def dxc_start(self,tt_sc,tt_user):
self.T = threading.Thread(target=lambda:self.chixu(tt_sc,tt_user)) # 多线程
self.T.start() # 启动
#主程序
if __name__=='__main__':
talkroom_system().connecting()
服务器
调用自定义库TJZ.py、TJZ_FZ.py、SJK.py、DY.py文章来源地址https://www.toymoban.com/news/detail-485707.html
from TJZ import ServerSocket
from TJZ_FZ import Socket_dabao
from threading import Thread
from DY import *
from SJK import mysql_use
#服务器核心模块
class Server(object):
def __init__(self):
#套接字
self.TJZ=ServerSocket()
#数据库
self.SJK=mysql_use()
self.SJK.create_kb()
self.message_all=self.SJK.message_get()
#在线人数
self.user_online={}
self.number=1
#处理响应不同功能
self.SJ_differ={}
self.chuli_sjlx(START_zhuce,self.request_zhuce)
self.chuli_sjlx(START_login,self.request_denglu)
self.chuli_sjlx(START_talk,self.request_talk)
self.chuli_sjlx(START_zh,self.request_zh_name)
#将函数装入字典
def chuli_sjlx(self,request,gongneng):
self.SJ_differ[request]=gongneng
#等待客户端连接
def starting(self):
while True:
#获取客户端连接
soc,addr =self.TJZ.accept()
#封装对象
KHD_soc=Socket_dabao(soc)
#传递信息(多线程进行)
Thread(target=lambda:self.ZXC_msg(KHD_soc)).start()
#接受信息并解码
def ZXC_msg(self,KHD_soc):
while True:
#获取数据
recv_data=KHD_soc.recv_data()
if recv_data==START_leave or recv_data=='':
#关闭客户端套接字
self.user_offline(KHD_soc)
KHD_soc.close()
break
#解码数据
KHD_msg_jm=self.shuju_jiema(recv_data)
gongneng=self.SJ_differ.get(KHD_msg_jm['request'])
#调用相应函数
if gongneng:
gongneng(KHD_msg_jm,KHD_soc)
#离线处理
def user_offline(self,KHD_soc):
#发送允许下线回应
KHD_soc.send_data(START_leave_yes)
#如果存在其他用户，告知新的在线用户列表
try:
for name,message_person in self.user_online.items():
if message_person['soc']==KHD_soc:
#制作新的用户列表
self.user_online.pop(name)
users_all=''
for name,message_person in self.user_online.items():
user_lingshi1=message_person['name_user']
user_lingshi2=users_all
users_all=user_lingshi2+SHOW_differ+user_lingshi1
msg_sendto=START_talk_users+users_all
#发送至所以在线客户端
for name,message_person in self.user_online.items():
KHD_soc.send_data(START_login_yes)
message_person['soc'].send_data(msg_sendto)
#如果不存在其他在线用户
except:
return
#处理数据
def shuju_jiema(self,revc_data):
#分割信息
recv_list=revc_data.split(SHOW_differ)
#将分割好的数据存入字典
data_chuli={}
data_chuli['request']=recv_list[0]
if data_chuli['request']==START_zhuce:
data_chuli['name_user']=recv_list[1]
data_chuli['password']=recv_list[2]
data_chuli['question']=recv_list[3]
data_chuli['answer']=recv_list[4]
elif data_chuli['request']==START_login:
data_chuli['name_user']=recv_list[1]
data_chuli['password']=recv_list[2]
elif data_chuli['request']==START_talk:
data_chuli['name_user']=recv_list[1]
data_chuli['message']=recv_list[2]
elif data_chuli['request']==START_zh:
data_chuli['name_user']=recv_list[1]
#返回数据字典
return data_chuli
#注册响应
def request_zhuce(self,data_chuli,KHD_soc):
#注册参数
check=1
#注册用户数据
lingshi=[data_chuli['name_user'],data_chuli['password'],data_chuli['question'],data_chuli['answer']]
data=tuple(lingshi)
#判断用户名是否重复
for i in self.message_all:
if lingshi[0]==i[0] :
check=2
#注册失败
KHD_soc.send_data(START_zhuce_no)
#注册成功
if check==1:
#更新数据库用户信息
self.SJK.message_add(data)
KHD_soc.send_data(START_zhuce_yes)
#获取新的数据库用户信息
self.message_all=self.SJK.message_get()
#登录响应
def request_denglu(self,data_chuli,KHD_soc):
#登录参数
check=1
#登录用户信息
list_lingshi=[data_chuli['name_user'],data_chuli['password']]
#判断用户信息是否正确
for i in self.message_all:
#登陆成功
if list_lingshi[0]==i[0] and list_lingshi[1]==i[1]:
check=2
#加入在线用户字典
self.user_online[list_lingshi[0]]={'soc':KHD_soc,'name_user':list_lingshi[0]}
#制作用户列表数据
users_all=''
for name,message_person in self.user_online.items():
user_lingshi1=message_person['name_user']
user_lingshi2=users_all
users_all=user_lingshi2+SHOW_differ+user_lingshi1
msg_sendto=START_talk_users+users_all
#登录成功请求恢复
KHD_soc.send_data(START_login_yes)
self.number+=1
user_in=START_user_in+SHOW_differ+list_lingshi[0]
if self.number>len(self.user_online):
#更新客户端用户列表
for name,message_person in self.user_online.items():
message_person['soc'].send_data(msg_sendto)
#告知在线客户端新用户登录
for name,message_person in self.user_online.items():
if message_person['name_user']!=list_lingshi[0]:
message_person['soc'].send_data(user_in)
#登陆失败
if check==1:
KHD_soc.send_data(START_login_no)
#找回响应
def request_zh_name(self,data_chuli,KHD_soc):
#找回参数
check=1
list_lingshi=[data_chuli['name_user']]
#判断找回用户是否存在
for i in self.message_all:
#存在该用户，返回数据信息
if list_lingshi[0]==i[0] :
check=2
qaa=START_n_yes+SHOW_differ+i[0]+SHOW_differ+i[1]+SHOW_differ+i[2]+SHOW_differ+i[3]
KHD_soc.send_data(qaa)
#不存在该用户
if check==1:
KHD_soc.send_data(START_n_no)
#聊天响应
def request_talk(self,data_chuli,KHD_soc):
#根据协议制作聊天信息
list_lingshi=[data_chuli['name_user'],data_chuli['message']]
msg=START_talk_msg +SHOW_differ+list_lingshi[0]+SHOW_differ+list_lingshi[1]
#转发至所有在线客户端
for name,message_person in self.user_online.items():
message_person['soc'].send_data(msg)
#测试程序
if __name__=='__main__':
Server().starting()自定义库
通讯协议及相关配置定义库(DY.py)
#数据协议相关配置
START_zhuce='zhuce'#注册请求
START_login='denglu'#登录请求
START_talk='liaotian'#聊天请求
START_zh='zhaohui'#找回请求
START_leave='end_user'#离线请求
START_user_in='user_in'#用户加入聊天室
START_zhuce_yes='zhuce_yes'#注册请求响应1
START_zhuce_no='zhuce_no'#注册请求响应2
START_login_yes='denglu_yes'#登录结果响应1
START_login_no='denglu_no'#登录结果响应2
START_talk_msg='msg'#聊天响应1
START_talk_users='users'#聊天响应2
START_n_yes='name_yes'#找回响应1
START_n_no='name_no'#找回响应2
START_zh_end='zh_end'#找回响应3
START_zh_ok='zh_yes'#找回响应4
START_leave_yes='xiaxian'#离线响应
SHOW_differ='+-+'#自定义协议分割符
SHOW_change='utf-8'#编码解码方式
#标题
TITLE='简易聊天室'
#用户自称
USER_self='我'
#服务器数据配置
MASTER_IP='127.0.0.1'
MASTER_PORT=9999
数据库调用库（SJK.py）
import pymysql
class mysql_use(object):
def __init__(self):
self.sjk=pymysql.connect(host='127.0.0.1',user='root',password='123456')
self.use_sjk=self.sjk.cursor()
self.jk='CREATE DATABASE IF NOT EXISTS FWQ'
self.use_k='use FWQ'
self.jb= """CREATE TABLE IF NOT EXISTS user_message(
name_user varchar(50) NOT NULL,
password varchar(50) NOT NULL,
question varchar(50) NOT NULL,
answer varchar(50) NOT NULL,
PRIMARY KEY (name_user)
) ENGINE=MyISAM AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
"""
#建库建表
def create_kb(self):
# 创建库和表
self.use_sjk.execute(self.jk)
try:
self.use_sjk.execute(self.use_k)
self.use_sjk.exec
except:
return
#添加用户信息
def message_add(self,user_data):
self.use_sjk.execute(self.use_k)
#输入数据
self.use_sjk.execute("insert into user_message(name_user,password,question,answer) values (%s,%s,%s,%s)",user_data)
#上传数据
self.sjk.commit()
#获取用户数据
def message_get(self):
self.use_sjk.execute(self.use_k)
self.use_sjk.execute("SELECT * FROM user_message")
user_data=self.use_sjk.fetchall()
data=[]
for i in user_data:
lingshi=[]
lingshi.append(i[0])
lingshi.append(i[1])
lingshi.append(i[2])
lingshi.append(i[3])
data.append(lingshi)
return data
#关闭数据库
def sjk_close(self):
self.use_sjk.close()
self.sjk.close()
套接字库（TJZ.py）
import socket
from DY import *
#自定义套接字，初始化服务器套接字的相关参数
class ServerSocket(socket.socket):
def __init__(self):
#设置为TCP类型
super(ServerSocket,self).__init__(socket.AF_INET,socket.SOCK_STREAM)
#绑定地址和端口
self.bind((MASTER_IP,MASTER_PORT))
#设置为监听模式
self.listen(100)
套接字通讯库（服务器端）(TJZ_FZ.py)
#封装套接字
class Socket_dabao(object):
def __init__(self,soc):
self.soc=soc
#接收数据并解码
def recv_data(self):
try:
return self.soc.recv(256).decode('utf-8')
except:
return ''
#发送编码数据
def send_data(self,message):
return self.soc.send(message.encode('utf-8'))
#关闭链接
def close(self):
self.soc.close()到了这里，关于简易聊天室（Python版）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
