异常处理 udp通信

客户端
import socket
import os
import json
import struct
client = socket.socket()
client.connect((‘127.0.0.1‘,8080))
while True:
    MOVIE_DIR = ‘G:\Python10期脱产视频\day01\视频‘
    movie_list = os.listdir(MOVIE_DIR)
    for i, movie in enumerate(movie_list,1):  # 获取电影列表,循环展示
        print(i,movie)
    choice = input(‘>>>:‘).strip()  #用户选择
    if choice.isdigit():   # 判断是否是数字
        choice = int(choice) - 1
        if choice in range(len(movie_list)):   # 判断用户选择的是否在列表中
            path = movie_list[choice]   # 获取用户上传的文件路径
            print(path)
            file_path = os.path.join(MOVIE_DIR,path)   #获取文件路径
            file_size = os.path.getsize(file_path)  #获取文件大小
            print(file_path)
            print(file_size)
            res = {‘name‘:‘大保健‘,‘file_size‘:file_size}  #定义一个字典
            json_d = json.dumps(res)   # 序列化字典
            json_bytes = json_d.encode(‘utf-8‘)
            print(json_bytes)
            header = struct.pack(‘i‘,len(json_bytes))  #制作一个报头
            client.send(header)  #发送报头
            client.send(json_bytes)  # 发送字典
            with open(file_path,‘rb‘)as f:   # 发文件数据
                for line in f:
                    client.send(line)
        else:
            print(‘not in range‘)
    else:
        print(‘must be a number‘)

服务端
import socket
import struct
import json
import os
server = socket.socket()
server.bind((‘127.0.0.1‘,8080))
server.listen(5)
while True:
    sock,addr = server.accept()
    while True:
        try:
            header_len = sock.recv(4)   # 解析字典报头
            header_len = struct.unpack(‘i‘,header_len)[0]
            header_dic = sock.recv(header_len)  # 接收字典数据
            real_dic = json.loads(header_dic.decode(‘utf-8‘))
            print(real_dic)
            print(real_dic.get(‘file_size‘))
            total_size = real_dic.get(‘file_size‘)  # 获取字典长度
            print(total_size)
            recv_size = 0  #循环写入文件
            print(real_dic.get(‘file_size‘))
            with open(real_dic.get(‘name‘),‘wb‘)as f:
                while recv_size < total_size:
                    data = sock.recv(1024)
                    f.write(data)
                    recv_size += len(data)
                print(‘上传成功‘)
        except ConnectionResetError as e:
            print(e)
            break
    sock.close()

try:
    res = {‘name‘:‘‘}
    print(res[‘pwd‘])
    print(res.get(‘pwd‘,‘None‘))
except KeyError:   #  键值错误
    print(‘KeyError‘)
except NameError:   # 变量名错误
    print(‘NameError‘)
except IndexError:    # 索引错误
    print(‘IndexError‘)
except Exception:     # 所有异常都可以捕获
    print(‘Exception‘)
except BaseException:   # 所有异常都可以捕获
    print(‘BaseException‘)

try:
    l = [1,2,3]
    l[111]
except Exception:  # 万能异常  所有的异常类型都被捕获
    print(‘老子天下无敌‘)
else:
    print(‘被检测的代码没有任何的异常发生 才会走else‘)
finally:
    print(‘无论被检测的代码有没有异常发生 都会在代码运行完毕之后执行我‘)

#主动抛异常
if ‘egon‘ == ‘DSB‘:
    pass
else:
    raise TypeError(‘尽说大实话‘)
# 关键字raise就是主动抛出异常

l = [1,2,3]
assert len(l) < 0  # 断言  预言
# 猜某个数据的状态 猜对了 不影响代码执行 正常走
# 猜错了  直接报错

# 自定义异常
class MyError(BaseException):
    def __init__(self, msg):
        super().__init__()
        self.msg = msg
    def __str__(self):
        return ‘<dfsdf%ssdfsdaf>‘ % self.msg

raise MyError(‘我自己定义的异常‘)  # 主动抛出异常其实就是将异常类的对象打印出来,会走__str__方法

udp 使用

服务端
import socket
server = socket.socket(type=socket.SOCK_DGRAM)  #UDP协议
server.bind((‘127.0.0.1‘,8080))
while True:
    data,addr = server.recvfrom(1024)
    print(‘数据:‘,data)
    print(‘地址:‘,addr)
    server.sendto(data.upper(),addr)

服务端
import socket
client = socket.socket(type=socket.SOCK_DGRAM)
server_address = (‘127.0.0.1‘,8080)
while True:
    client.sendto(b‘hello‘,server_address)
    data,addr = client.recvfrom(1024)
    print(‘服务端发来的数据‘,data)
    print(‘服务端发来的地址‘,addr)

简易版qq

客户端
import socket
client = socket.socket(type=socket.SOCK_DGRAM)
server_address = (‘127.0.0.1‘,8080)
while True:
    res = input(‘>>>:‘).strip()
    client.sendto(res.encode(‘utf-8‘),server_address)
    data,server_address = client.recvfrom(1024)
    print(data.decode(‘utf-8‘))

服务端
import socket
server = socket.socket(type=socket.SOCK_DGRAM)
server.bind((‘127.0.0.1‘,8080))
while True:
    data,addr = server.recvfrom(1024)
    print(data.decode(‘utf-8‘))
    res = input(‘>>>:‘).strip()
    server.sendto(res.encode(‘utf-8‘),addr)

tcp 解决粘包
客户端
import socket
client = socket.socket()
client.connect((‘127.0.0.1‘,8080))
while True:
    client.send(b‘hello‘)
    data = client.recv(1024)
    print(data.decode(‘utf-8‘))

服务端
import socketserver
class MyServer(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            data = self.request.recv(1024)
            print(self.client_address)  # 客户端地址
            print(data.decode(‘utf-8‘))  #
            self.request.send(data.upper())
if __name__ == ‘__main__‘:    # 只要有客户连接,会自定义交给类中handler方法去处理
    server = socketserver.ThreadingTCPServer((‘127.0.0.1‘,8080),MyServer)
    server.serve_forever()   #启动该服务器