基于python实现聊天室程序

作者：跟我搞对象吧 | 2022-12-27 15:16

这篇文章主要为大家详细介绍了基于python实现聊天室程序，该程序由客户端与服务器构成，使用UDP服务，实现了群发、私发、点对点文件互传功能，具有一定的参考价值，感兴趣的小伙伴们可以参考一下

本文实例为大家分享了python实现简单聊天室的具体代码，供大家参考，具体内容如下

刚刚接触python编程，又从接触java开始一直对socket模块感兴趣，所以就做了一个聊天室的小程序。

该程序由客户端与服务器构成，使用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()

服务器端主要进行消息的分类转发处理，用户列表、地址列表的维护。

服务器端代码如下：

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)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

推荐阅读

程序员
iOS9"游戏中心无法识别此游戏."

如何解决《iOS9"游戏中心无法识别此游戏."》经验，为你挑选了3个好方法。 ... [详细]
程序员
如何将我的远程git存储库还原为某个提交？

如何解决《如何将我的远程git存储库还原为某个提交？》经验，为你挑选了1个好方法。 ... [详细]
程序员
将\ in(是集合的成员)表示法添加到qplot中

$将\ in(是集合的成员)表示法添加到qplot中$

如何解决《将\in(是集合的成员)表示法添加到qplot中》经验，为你挑选了1个好方法。 ... [详细]
程序员
null和undefined不一致的比较

如何解决《null和undefined不一致的比较》经验，为你挑选了1个好方法。 ... [详细]
程序员
deleteRowsAtIndexPaths:滑动删除行永远不会释放单元格？

如何解决《deleteRowsAtIndexPaths:滑动删除行永远不会释放单元格？》经验，为你挑选了0个好方法。 ... [详细]
程序员
如何使用Java加载纸莎草uml模型？

如何解决《如何使用Java加载纸莎草uml模型？》经验，为你挑选了1个好方法。 ... [详细]
程序员
如何搜索有人用git更改的字符串？

如何解决《如何搜索有人用git更改的字符串？》经验，为你挑选了0个好方法。 ... [详细]
程序员
BouncyCastle现在更新了pgp密钥,导致校验和不匹配错误

如何解决《BouncyCastle现在更新了pgp密钥,导致校验和不匹配错误》经验，为你挑选了1个好方法。 ... [详细]
程序员
将SEXP从R转换为C++中的字符串向量

如何解决《将SEXP从R转换为C++中的字符串向量》经验，为你挑选了1个好方法。 ... [详细]
程序员
C#6 null传播当object为null时设置的值

如何解决《C#6null传播当object为null时设置的值》经验，为你挑选了2个好方法。 ... [详细]
程序员
iOS Swift:从两个Int创建Float值

如何解决《iOSSwift:从两个Int创建Float值》经验，为你挑选了1个好方法。 ... [详细]
程序员
寻找对应于最小特征值的特征向量

如何解决《寻找对应于最小特征值的特征向量》经验，为你挑选了1个好方法。 ... [详细]
程序员
NPM安装与package.json在提取 - > gunzTarPerm期间退出,没有错误消息

如何解决《NPM安装与package.json在提取->gunzTarPerm期间退出,没有错误消息》经验，为你挑选了1个好方法。 ... [详细]
程序员
在Google BigQuery中选择除一些以外的所有列？

如何解决《在GoogleBigQuery中选择除一些以外的所有列？》经验，为你挑选了2个好方法。 ... [详细]
程序员
完成Docker for Windows教程时找不到Eval命令

如何解决《完成DockerforWindows教程时找不到Eval命令》经验，为你挑选了0个好方法。 ... [详细]
程序员
前n个奇数的总和

如何解决《前n个奇数的总和》经验，为你挑选了1个好方法。 ... [详细]
程序员
Javascript \ x转义

如何解决《Javascript\x转义》经验，为你挑选了1个好方法。 ... [详细]
程序员
Vimscript:在字符串上获取正则表达式的所有匹配项

如何解决《Vimscript:在字符串上获取正则表达式的所有匹配项》经验，为你挑选了1个好方法。 ... [详细]
程序员
如何检查整数是否是一个完美的正方形

如何解决《如何检查整数是否是一个完美的正方形》经验，为你挑选了1个好方法。 ... [详细]
程序员
NodeJS - connect-keycloak中间件空响应错误

如何解决《NodeJS-connect-keycloak中间件空响应错误》经验，为你挑选了0个好方法。 ... [详细]

跟我搞对象吧

这个屌丝很懒，什么也没留下！

关注作者

Tags | 热门标签

RankList | 热门文章