仿QQ实时聊天(socket)

在线互动聊天系统的设计与实现

一.系统需求分析

1.1 引言

1.1.1编写目的

本文档的编写目的是在系统分析设计阶段,对于网络聊天系统的需求进行明确的描述,并在此基础上对系统的总体设计思路进行概述,为后续的详细设计、编码和测试工作提供参考。

1.1.2背景

是重庆邮电大学一名学生在其程序设计实验课程的作品。

1.1.3术语和缩略词

待定

1.1.4参考资料

https://github.com/YuumiAndYasuo/socket_chat

1.2任务概述

本系统的开发任务是设计并实现一个类似QQ的网络聊天系统,系统需要包含用户注册、登录、用户关系管理、单聊、群聊、消息收发等功能。

1.3项目概述

1.3.1 项目来源及背景

随着智能手机和移动互联网的普及,各类聊天软件已成为人们获取信息和沟通的重要工具。本系统希望开发一个类似QQ的聊天系统,提供更好的用户体验。

1.3.2 项目目标

开发一个支持PC终端的网络聊天系统,

支持文本消息类型传输

提供联系人、群组管理功能,支持群聊

1.3.3系统功能概述

本系统主要包含以下功能模块:

1.用户管理:用户账号注册、登录等功能

2.好友管理:添加、删除好友,好友分组等功能

3.单聊、群聊:一对一聊天和群聊天功能

4.消息处理:支持多种消息类型

5.系统支持Windows平台,用python语言编写，服务器端基于Windows10平台。数据库选用MySQL。

1.4用户特点

以年轻用户为主,用户年龄分布以20-35岁为主。

较高的智能手机和移动互联网使用习惯,有更高的聊天工具使用需求

要求聊天工具易用性较高,交互体验友好

1.5功能需求

1.5.1系统功能

用户管理:注册、登录

好友管理:添加、删除

单聊:一对一文字聊天

群聊:群聊天

消息处理:文本消息

1.5.2功能描述

注：此部分按功能模块描述。

1.用户管理

注册:提交用户名、密码完成注册

登录:用户名密码验证后登录系统

查找用户:通过用户id搜索查找用户

2.好友管理

添加好友:向用户发起好友请求后建立好友

删除好友:解除好友关系

3.单聊

发起单聊:选择好友发起文字聊天

4.群聊

创建群聊:创建群聊天

群聊天:群成员实时聊天

1.6性能需求

1.6.1数据精度要求

好友关系数据准确性要求较高,错误率<0.1%

消息内容存储精确度要求较高,所有消息原样存储

1.6.2时间特性

服务器响应时间小于200ms

消息传递延迟小于500ms

1.7运行需求

1.7.1系统界面需求

界面风格简洁干净,符合现代视觉风格

1.7.2软件接口

socket, tkinter, threading, time, pymyusql

1.7.3硬件接口

无

二.系统设计

2.1 系统整体功能结构

2.2 系统功能模块详细设计

注：以下部分按照各个模块功能进行介绍

1. 用户注册登录模块:描述用户注册、登录等功能

新用户在使用此系统时，必须先进行新用户注册，后台将注册信息添加到数据库user_info表中。

注册成功之后再进行登录，用户id与密码匹配成功才可进入主界面

2. 好友管理模块:添加、删除好友、添加群聊、删除群聊等功能

在输入栏中输入正确的用户id或者群id才可以添加，主页面的好友栏可以右击选中，点击删除即可删除好友或者群聊，点击之后删除数据库friend_info 表和group_info表中相应的信息，自动刷新好友或群聊列表

3. 单聊模块:发送消息和接收消息

进入聊天时启动客户端程序，接收消息线程打开，用于接收和发送消息

在聊天界面消息输入框输入想要发送的消息，点击发送，通过socket模块功能将发送的消息发送到指定接收对象，消息区域显示发送过或收到的消息。接收到消息时将消息和发送者打印在消息区域

4. 群聊模块:发送消息和接收消息

需提前启动服务端程序，打开每个群的接收消息及转发消息进程

在聊天界面消息输入框输入想要发送的消息，将消息发送的指定的群聊端口。

服务器转发群聊消息时，在数据库查找群聊所有群成员的端口，将消息发送至除消息发送至以外的所有群成员

2.3 系统界面设计

2.4 数据库设计

实体1:用户(User)

用户ID(用户的唯一标识)

密码

实体2:好友关系(Friend)

用户ID

好友用户ID

实体3:群组(Group)

群ID

用户ID

通过这些实体可以描述出用户与用户、用户与群组之间的关系。

主要关系:

用户与用户之间建立好友关系

用户可加入多个群组,群组中有多位成员

用户可在群组中发送消息,消息可推送给所有成员

用户之间可以互发私聊消息

2.4.1 数据库E-R图

2.4.2 关系模式

1. 用户表(user_info)

User_info(用户ID, 密码,端口号)

2. 好友关系表(friend_info)

friend_info(用户ID, 好友用户ID)

3. 群组表(group_info)

Group_info(群ID, 用户ID,端口号)

2.4.3 物理结构设计

1.用户表(user_info)

CREATE TABLE user_info
(
    user_id varchar(10) PRIMARY KEY,
    pwd VARCHAR(15) NOT NULL,
    port INT NOT NULL
);

2. 好友关系表(friend_id)

CREATE TABLE friend_info
(
    user_id varchar(10),
    friend_id varchar(10),
    PRIMARY KEY (user_id, friend_id),
    FOREIGN KEY (user_id)
    REFERENCES user_info(user_id)
);

3. 群组表(group_id)

CREATE TABLE group_info
(
    Group_id varchar(10)
    member_id varchar(10),
    Port INT NOT NULL,
    FOREIGN KEY (member_ID) REFERENCES user_info(user_ID)
)

三.系统编码与实现

3.1 用户登录功能编码实现

#登录
class LoginPanel:
    def run(self):
        self.root.mainloop()

    def __init__(self):
        # 图形界面区域
        self.root = Tk()
        self.root.title('登录')
        self.root.geometry('450x220+700+400')

        # 输入区域
        self.input = Canvas(self.root, bg='#ffffff')
        self.input.place(x=0, y=0, heigh=350, width=450)

        # 账号输入框标签
        Label(self.root, text="账号:", background='white').place(x=75, y=53)
        Label(self.root, text="密码:", background='white').place(x=75, y=90)
        db=connect_to_mysql()
        cursor = db.cursor()
        sql1 = 'select id from user_info'
        cursor.execute(sql1)
        print('执行成功!')
        id_data = cursor.fetchall()
        print(id_data)

        # 账号输入框  账号输入框
        xVariable = tkinter.StringVar()  # #创建变量，便于取值
        self.accountinput = ttk.Combobox(self.root, textvariable=xVariable)  # 创建下拉菜单
        self.accountinput.pack()  # #将下拉菜单绑定到窗体
        self.accountinput.place(x=130, y=50, heigh=30, width=210)
        self.accountinput["value"] = id_data  # #给下拉菜单设定值
        self.accountinput.current(0)  # 设定下拉菜单的默认值为第1个

        # 密码输入框
        self.passwordinput = Entry(self.input, font=("宋体", 16, "bold"), show='*')
        self.passwordinput.place(x=130, y=85, heigh=30, width=210)

        # 登陆按钮
        self.loginbutton = Button(self.input, text='登录', bg='#4fcffd',command=self.loginbutton_clicked)
        self.loginbutton.place(x=100, y=160, heigh=40, width=240)

        # 注册账号按钮
        self.registerbutton = Button(self.input, text='注册账号', bg='white',command=RegisterPanel)
        self.registerbutton.place(x=10, y=190, heigh=20, width=70)
        sql3 = 'select id from user_info where id=%s' % (self.accountinput.get())
        cursor.execute(sql3)
        self.account_id=cursor.fetchone()
        db.close()  #关闭数据库
        print('数据库连接断开')
        # 登录按钮点击事件

    def loginbutton_clicked(self):
        # 输入过滤
        account = self.accountinput.get().strip().replace(' ', '')
        self.accountinput.delete(0, END)
        self.accountinput.insert(END, account)
        print(account)
        password = self.passwordinput.get().strip().replace(' ', '')
        self.passwordinput.delete(0, END)
        self.passwordinput.insert(END, password)
        print(password)
        print('等待连接数据库。。。')
        db = connect_to_mysql()
        cursor = db.cursor()
        sql1 = "select * from user_info where id=%s" % account  # 数据库中查询与账号对应的元组
        sql2 = 'select id from user_info'
        # 获取数据库中的信息
        cursor.execute(sql1)
        result1 = cursor.fetchone()
        # cursor.execute(sql2)
        # result2=cursor.fetchone()
        print(account)
        password = self.passwordinput.get().replace(' ', '')
        print(password)

        if len(account) < 8 or not account.isdigit():
            messagebox.showinfo('登录失败', '查无此号')
        for c in password:
            if ord(c) > 255:
                messagebox.showinfo('登录失败', '密码错误\n( ⊙ o ⊙ )')
        try:
            print('results', result1)
            if result1[1] == password:
                messagebox.showinfo('登录成功', '登录成功')
                db.close()
                self.root.destroy()
                mainpanel = MainPanel(result1[0])
                mainpanel.run()
                return 1
            messagebox.showinfo('登录失败', '账号密码不匹配')
        except:
            print('登录抛出异常')
            db.rollback()
        return -1

3.2 用户注册功能编码实现

#注册
class RegisterPanel:
    def run(self):
        self.root.mainloop()
    #图形界面
    def __init__(self):
        self.root = Tk()
        self.root.title('用户注册')
        self.root.geometry('450x220+700+450')

        # 输入区域
        self.input = Canvas(self.root, bg='#ffffff')
        self.input.place(x=0, y=0, heigh=220, width=450)

        # 昵称账号密码输入提醒
        Label(self.root, text="账号:", background='white').place(x=77, y=63)
        Label(self.root, text="密码:", background='white').place(x=77, y=97)

        # 账号输入框
        self.accountinput = Entry(self.input, font=("宋体", 16, "bold"))
        self.accountinput.place(x=130, y=60, heigh=30, width=210, )

        # 密码输入框
        self.passwordinput = Entry(self.input, font=("宋体", 16, "bold"), show='*')
        self.passwordinput.place(x=130, y=95, heigh=30, width=210)

        # 注册按钮
        self.registerbutton = Button(self.input, text='立即注册', bg='#4fcffd', command=self.register_button_clicked)
        self.registerbutton.place(x=100, y=160, heigh=40, width=240)

    #注册按钮点击
    def register_button_clicked(self):
        # 输入
        account = self.accountinput.get().strip().replace(' ', '')
        self.accountinput.delete(0, END)
        self.accountinput.insert(END, account)
        print(account)
        password = self.passwordinput.get().strip().replace(' ', '')
        self.passwordinput.delete(0, END)
        self.passwordinput.insert(END, password)
        print(password)
        # 输入过滤
        #实际用的时候好像不是那么理想，待完善
        if len(account) < 8 or len(password) < 8:
            messagebox.showinfo('注册失败', '账号或密码至少8位\no(︶︿︶)o')
            return -1
        if not account.isdigit():
            messagebox.showinfo('注册失败', '账号必须全为数字\n(╯﹏╰）')
            return -2
        for c in password:
            if ord(c) > 255:
                messagebox.showinfo('注册失败', '密码不能包含非法字符\n( ⊙ o ⊙ )')
                return -3
        port=int(random.randint(1,1000000)%65535)
        db=connect_to_mysql()
        cursor=db.cursor()
        sql="INSERT INTO user_info(id, pwd, port) VALUES ('%s', '%s', %d)" % (account, password, port)
        try:
            cursor.execute(sql)
            db.commit()
            print('成功修改数据库内容')
        except:
            db.rollback()
            print("失败")
        db.close()
        messagebox.showinfo('注册成功','恭喜您 注册成功\n~\(≧▽≦)/~')
        self.root.destroy()#销毁注册窗口
        return 1

3.3 用户私聊功能编码实现

注：根据自己的系统功能划分进行编码介绍，粘贴主要代码，对相应的功能进行详细描述。

开启聊天时，自动打开客户端,提供用户私聊功能

#客户端
class client:
    def __init__(self, user_id, contact_id, chatPanel):
        self.chatPanel=chatPanel
        self.user_id = user_id
        self.contact_id=contact_id
        print('conact_id为：'+str(self.contact_id))
        #连接数据库，获取用户端口
        self.db = connect_to_mysql()
        self.cursor = self.db.cursor()
        sql1 = 'select port from user_info where id=%s' % (self.user_id)
        self.cursor.execute(sql1)
        self.user_port = int(self.cursor.fetchone()[0])
        self.s = socket.socket()#绑定用户socket
        self.s.bind(('127.0.0.1',self.user_port))

    #私聊发信息
    def send_message_to_friend(self, msg):
        # 根据好友id从数据库获取端口号
        sql2 = 'select port from user_info where id=%s' % (self.contact_id)
        print(sql2)
        self.cursor.execute(sql2)
        self.friend_port = int(self.cursor.fetchone()[0])
        print(self.friend_port)
        # 建立socket连接
        self.send_socket = socket.socket()
        self.send_socket.connect(('127.0.0.1', self.friend_port))
        try:
            # 发送消息给好友
            message=str(self.user_id)+','+str(msg)#发送者id和消息内容
            self.send_socket.send(bytes(message,encoding='utf-8'))
            print('消息发送成功')
        except:
            messagebox.showinfo('错误','消息发送出错')
    #群聊发信息
    def send_message_to_group(self, msg):
        # 根据群聊从数据库获取端口号
        sql2 = 'select port from group_info where group_id=%s' % (self.contact_id[0])
        self.cursor.execute(sql2)
        self.group_port = int(self.cursor.fetchone()[0])
        #建立socket连接
        self.send_socket = socket.socket()
        self.send_socket.connect(('127.0.0.1', self.group_port))
        try:
            # 发送消息给服务器
            message=str(self.user_id)+','+str(msg)#发送者id和消息内容
            self.send_socket.send(bytes(message, encoding='utf-8'))
            print('消息发送成功')
        except:
            messagebox.showinfo('错误', '消息发送出错')

    # 接收私聊发来的信息
    def recv_private_message(self):
        self.s.listen(5)
        while True:
            conn, addr = self.s.accept()
            # 接收消息
            t = strftime("%Y-%m-%d %H:%M:%S", localtime())
            message = conn.recv(1024).decode()
            if not message:
                continue
            send_user_id = message.split(',')[0]  # 发送者id
            msg = message.split(',')[1]  # 消息内容
            self.chatPanel.chat_scroll_box.insert(END, str(t) + '\n' + str(send_user_id) + '\t:' + str(msg) + '\n')
    #接收群聊信息
    def recv_group_message(self):
        self.s.listen(5)
        while True:
            conn, addr = self.s.accept()
            # 接收消息
            t = strftime("%Y-%m-%d %H:%M:%S", localtime())
            message = conn.recv(1024).decode()
            if not message:
                continue
            send_user_id=message.split(',')[0]#发送者id
            msg=message.split(',')[1]#消息内容
            self.chatPanel.chat_scroll_box.insert(END, str(t) + '\n' + str(send_user_id) + '\t:' + str(msg) + '\n')
    #断开与服务器的连接
    def close_connect(self):
        self.s.close()
        self.db.close()#关闭数据库连接
        print('客户端与服务器断开连接！！！\n')

用户私聊功能实现

#一对一聊天界面
class private_ChatPanel:
    def run(self):
        threading.Thread(target=self.client.recv_private_message).start()  # 接收私聊信息线程开启
        print('接收信息线程启动!\n')
        self.root.mainloop()

    def __init__(self,user_id,friend_id):
        self.root = Tk()
        self.client=client(user_id,friend_id,self)
        self.user_id = user_id
        self.friend_id = friend_id

        # 标题
        self.root.title('登录用户: '+str(self.user_id)+'聊天对象:'+str(self.friend_id))
        self.root.geometry('700x600+400+50')
        self.headtitle = Canvas(self.root, bg='skyblue')
        self.headtitle.place(x=5, y=5, heigh=40, width=690)

        #好友名字标签
        titlenamelable=Label(self.headtitle,text=self.friend_id,bg='skyblue')
        titlenamelable.place(x=50,y=5, heigh=30, width=300)
        #聊天信息区域
        self.chat_area = Canvas(self.root, bg='orange')
        self.chat_area.place(x=5, y=45, heigh=400, width=690)
        #输入区域
        self.input_area = Canvas(self.root, bg='pink')
        self.input_area.place(x=5, y=445, heigh=150, width=690)

        #发送按钮
        send_button=Button(self.input_area, text='发送',command=self.send_button_clicked) #待确定
        send_button.place(x=600, y=110, heigh=30, width=80)
        #关闭按钮
        closebutton=Button(self.input_area, text='关闭', command=self.close_button_clicked)#绑定按键功能为摧毁当前界面
        closebutton.place(x=510, y=110, heigh=30, width=80)

        #聊天信息框
        self.chat_scroll_box=scrolledtext.ScrolledText(self.chat_area, font=("宋体", 16, "normal"))
        self.chat_scroll_box.place(x=5, y=5, heigh=390, width=680)

        # 信息输入框
        self.input_chat_box = scrolledtext.ScrolledText(self.input_area, font=("宋体", 16, "normal"))
        self.input_chat_box.place(x=5, y=5, heigh=100, width=680)

    #关闭按钮点击事件
    def close_button_clicked(self):
        self.root.destroy()
        self.client.close_connect()

    #发送按钮点击事件
    def send_button_clicked(self):
        # 获取输入信息
        t = strftime("%Y-%m-%d %H:%M:%S", localtime())
        msg = self.input_chat_box.get('0.0', END)
        self.input_chat_box.delete('0.0', END)
        # 若不为空将消息发送
        if msg:
            self.client.send_message_to_friend(msg)
        else:
            messagebox.showinfo('提示','消息不能为空')
        print('发送按钮被点击了')
        # 将输入信息贴到聊天记录中
        self.chat_scroll_box.insert(END, str(t) + '\n' + str('我:  ') + str(msg)+'\n')

3.4 用户群聊功能编码实现

服务端程序（需在打开群聊之前开启）

#服务器程序
class service:
    def __init__(self,group_id):
        self.group_id=group_id[0]
        #获取群端口
        db=connect_to_mysql()
        cursor = db.cursor()
        sql1 = 'SELECT distinct port FROM group_info where group_id=%s' % self.group_id
        cursor.execute(sql1)
        self.group_port=cursor.fetchone()[0]#群端口
        sql2='select member_id from group_info where group_id=%s' % self.group_id
        cursor.execute(sql2)
        self.group_member_list=cursor.fetchall()#群聊中的成员列表
        print(self.group_member_list)
        # 创建服务器socket
        self.server_socket = socket.socket()
        # self.server_socket.setblocking(False)#设置socket为非阻塞模式
        # 绑定监听端口
        self.server_socket.bind(('127.0.0.1', int(self.group_port)))
        self.server_socket.listen(50)
        # 开始监听
        threading.Thread(target=self.recv_message).start()#接收信息线程
    # 接收信息
    def recv_message(self):
        while True:
            try:
                print('群聊' + str(self.group_id) + '正在监听')
                conn,addr=self.server_socket.accept()#接收连接
                print(str(addr)+'已连接')
                message = conn.recv(1024).decode()
                #将信息数据拆分
                send_user_id=message.split(',')[0]#发送者id
                msg=message.split(',')[1]#发送的消息内容
                # 给群里的成员每一个都发信息
                for j in range(len(self.group_member_list)):
                    self.send_msg_to_group(self.group_member_list[j][0],send_user_id,msg)
            except:
                continue
    #群发消息
    def send_msg_to_group(self,recv_id,send_user_id,msg):
        db = connect_to_mysql()
        cursor = db.cursor()
        sql1 = 'SELECT port FROM user_info where id=%s' % recv_id#得到接收者的端口
        cursor.execute(sql1)
        recv_port=cursor.fetchone()[0]
        if recv_id !=send_user_id:
            #根据群聊成员端口发送信息
            self.send_socket=socket.socket()
            self.send_socket.connect(('127.0.0.1', recv_port))
            self.send_socket.send(bytes(str(send_user_id)+','+str(msg),encoding='utf-8'))#发送者id和消息内容

群聊功能实现

#群聊界面
class group_ChatPanel:
    def run(self):
        threading.Thread(target=self.client.recv_group_message).start()  # 接收群聊信息线程开启
        print('接收信息线程启动!\n')
        self.root.mainloop()

    def __init__(self,user_id,group_id):
        self.root = Tk()
        self.client=client(user_id,group_id,self)
        self.user_id = user_id
        self.group_id = group_id

        # 标题
        self.root.title('登录用户: '+str(self.user_id)+' 群聊:'+str(self.group_id[0]))
        self.root.geometry('700x600+400+50')
        self.headtitle = Canvas(self.root, bg='skyblue')
        self.headtitle.place(x=5, y=5, heigh=40, width=690)

        #好友名字标签
        titlenamelable=Label(self.headtitle,text=self.group_id,bg='skyblue')
        titlenamelable.place(x=50,y=5, heigh=30, width=300)
        #聊天信息区域
        self.chat_area = Canvas(self.root, bg='orange')
        self.chat_area.place(x=5, y=45, heigh=400, width=690)
        #输入区域
        self.input_area = Canvas(self.root, bg='pink')
        self.input_area.place(x=5, y=445, heigh=150, width=690)

        #发送按钮
        send_button=Button(self.input_area, text='发送',command=self.send_button_clicked) #待确定
        send_button.place(x=600, y=110, heigh=30, width=80)
        #关闭按钮
        closebutton=Button(self.input_area, text='关闭', command=self.close_button_clicked)#绑定按键功能为摧毁当前界面
        closebutton.place(x=510, y=110, heigh=30, width=80)

        #聊天信息框
        self.chat_scroll_box=scrolledtext.ScrolledText(self.chat_area, font=("宋体", 16, "normal"))
        self.chat_scroll_box.place(x=5, y=5, heigh=390, width=680)

        # 信息输入框
        self.input_chat_box = scrolledtext.ScrolledText(self.input_area, font=("宋体", 16, "normal"))
        self.input_chat_box.place(x=5, y=5, heigh=100, width=680)

    #关闭按钮点击事件
    def close_button_clicked(self):
        self.root.destroy()
        self.client.close_connect()

    #发送按钮点击事件
    def send_button_clicked(self):
        # 获取输入信息
        t = strftime("%Y-%m-%d %H:%M:%S", localtime())
        msg = self.input_chat_box.get('0.0', END)
        self.input_chat_box.delete('0.0', END)
        # 将消息发送
        message=str(msg)
        self.client.send_message_to_group(message)
        print('发送按钮被点击了')
        # 将输入信息贴到聊天记录中
        self.chat_scroll_box.insert(END, str(t) + '\n' + str('我: ') + str(msg)+'\n')

四.系统测试

4.1 测试范围

1.注册界面测试

2.登录界面测试

3.好友、群聊管理测试

4.私聊测试

5.群聊测试

4.2 测试环境与系统配置

Windows10, python 3.10

4.3 测试覆盖设计

1. 注册界面测试

• 验证所有输入字段均能正常输入

• 异常输入检查:空字段/错误格式输入

• 提交注册请求,确认返回注册成功页面

2. 登录界面测试

• 正确账号密码登录,确认进入主页面

• 错误账号/密码登录,确认提示错误信息

3. 好友、群聊管理测试

• 添加/删除好友,确认好友列表同步更新

• 创建、加入、退出群聊,确认群聊列表同步更新

4. 私聊测试

• 不同账号间发送文本消息,确认消息正确发送和接收

5. 群聊测试

• 群成员间发送文本消息,确认群成员都能收到消息

4.4 功能测试用例

完整的项目在https://github.com/gaifagafin/-Python-socket-QQ-bushi

项目存在很多瑕疵和不足，可能等以后有时间再去慢慢完善