day23-2人聊天

talk_check.py 输入判定

import re

class TalkCheck:
    # 网址正则
    __network = r'(http|ftp|https):\/\/[\w\-_]+(\.[\w\-_]+)+([\w\-\.,@?^=%&:/~\+#]*[\w\-\@?^=%&/~\+#])?'
    # 图片正则
    __image = r'.*(.jpg|.png|.gif)$'

    @classmethod
    def check_word(cls, str_talk):
        if str_talk == '拜拜':
            return 0
        elif re.fullmatch(cls.__network, str_talk) is not None:
            return 1
        elif re.fullmatch(cls.__image, str_talk) is not None:
            return 2
        return

server.py 服务器端（1人）

import socket
from talk_check import TalkCheck
from requests import request

if __name__ == '__main__':
    flag = True
    server = socket.socket()
    server.bind(('10.7.181.82', 8080))
    server.listen(50)
    while flag:
        c, addr = server.accept()
        print(addr)
        while True:
            m = input('server>>>')
            if TalkCheck.check_word(m) == 0: 
                c.send('connection break......'.encode())
                print('connection break......')
                c.close()
                flag = False
                break
            elif TalkCheck.check_word(m) == 2:
                with open(m, 'rb') as f:
                    content = f.read()
                c.send(content)
                c.close()
            else:
                c.send(m.encode())
            try:
                data = c.recv(1024)
            except (ConnectionAbortedError, OSError):
                print('send picture only once, connection break......')
                flag = False
                break
            try:
                data_test = data.decode()
                print(data_test)
                if data_test == 'connection break......':
                    c.close()
                    flag = False
                    break
                elif TalkCheck.check_word(data_test) == 1:
                    response = request('GET', data_test)
                    text = response.text
                    with open('./client.txt', 'w', encoding='utf-8') as f:
                        f.write(text)
            except UnicodeDecodeError:
                print('对方传过来一张图片，正在下载......')
                data_null = bytes()
                while data:
                    data_null += data
                    data = c.recv(1024)
                with open('./client.gif', 'wb') as f:
                    f.write(data_null)

client.py 客户端（1人）

import socket
from talk_check import TalkCheck
from requests import request

if __name__ == '__main__':
    client = socket.socket()
    client.connect(('10.7.181.82', 8080))
    while True:
        try:
            data = client.recv(1024)
        except (ConnectionAbortedError, OSError):
            print('send picture only once, connection break......')
            break
        try:
            data_test = data.decode()
            print(data_test)
            if data_test == 'connection break......':
                client.close()
                break
            elif TalkCheck.check_word(data_test) == 1:
                response = request('GET', data_test)
                text = response.text
                with open('./server.txt', 'w', encoding='utf-8') as f:
                    f.write(text)
        except UnicodeDecodeError:
            print('对方传过来一张图片，正在下载......')
            data_null = bytes()
            while data:
                data_null += data
                data = client.recv(1024)
            with open('./server.gif', 'wb') as f:
                f.write(data_null)
        str1 = input('client>>>')
        if TalkCheck.check_word(str1) == 0:
            client.send('connection break......'.encode())
            print('connection break......')
            client.close()
            break
        elif TalkCheck.check_word(str1) == 2:
            with open(str1, 'rb') as f:
                content = f.read()
            client.send(content)
            client.close()
        else:
            client.send(str1.encode())

注：交流文字和站点资源没问题，但是传递图片会断开。