一.Socketserver
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
#服务端 import socketserver class KnightServer(socketserver.BaseRequestHandler): def handle(self): # self.request 相当于conn链接通道 while 1: from_client_msg = self.request.recv(1024) print(from_client_msg.decode('utf-8')) server_msg = input('服务端说>>>') self.request.send(server_msg.encode('utf-8')) if __name__ == '__main__': ip_port = ('127.0.0.1',8001) server = socketserver.ThreadingTCPServer(ip_port,KnightServer) server.serve_forever() #客户端 import socket import time client = socket.socket() client.connect(('127.0.0.1',8001)) while 1: client_data = input('别开车>>>') client.send(client_data.encode('utf-8')) from_server_msg = client.recv(1024) print(from_server_msg.decode('utf-8')) client.close()
#ftp文件上传服务端
import socket
import time
import json
server = socket.socket()
server.bind(('127.0.0.1',8001))
server.listen()
conn,addr = server.accept()
def upload(conn,client_file_info):
#
upload_file_path = r'D:jj' + '\' + client_file_info['file_name']
recv_all_datalen = 0
with open(upload_file_path,'wb') as f:
while recv_all_datalen < client_file_info['file_size']:
every_recv_data = conn.recv(1024)
every_recv_datalen = len(every_recv_data)
recv_all_datalen += every_recv_datalen
f.write(every_recv_data)
print('客户端上传文件')
client_userinfo = conn.recv(1024).decode('utf-8')
#定义一个需要返回给服务端的功能列表
server_function_list = {'1':'上传','2':'下载'}
#将功能字典转换为json字符串,然后进行编码,转成bytes类型
json_func_list = json.dumps(server_function_list,ensure_ascii=False)
#服务端登录认证
with open('userinformation','r',encoding='utf-8') as f:
for i in f:
if i.strip() == client_userinfo:
#登录成功以后,将服务端能够提供的功能,发送给客户端
conn.send(json_func_list.encode('utf-8'))
#接收客户端的文件描述信息,其中有一项是功能序号
client_fileinfo_json = conn.recv(1024).decode('utf-8')
client_file_info = json.loads(client_fileinfo_json)
print(client_file_info)
if client_file_info['func_choice'] == '1':
upload(conn,client_file_info)
break
else:
#登录错误的信息码
conn.send(b'201')
#ftp 文件上传客户端
import socket
import time
import json
server = socket.socket()
server.bind(('127.0.0.1',8001))
server.listen()
conn,addr = server.accept()
def upload(conn,client_file_info):
#
upload_file_path = r'D:jj' + '\' + client_file_info['file_name']
recv_all_datalen = 0
with open(upload_file_path,'wb') as f:
while recv_all_datalen < client_file_info['file_size']:
every_recv_data = conn.recv(1024)
every_recv_datalen = len(every_recv_data)
recv_all_datalen += every_recv_datalen
f.write(every_recv_data)
print('客户端上传文件')
client_userinfo = conn.recv(1024).decode('utf-8')
#定义一个需要返回给服务端的功能列表
server_function_list = {'1':'上传','2':'下载'}
#将功能字典转换为json字符串,然后进行编码,转成bytes类型
json_func_list = json.dumps(server_function_list,ensure_ascii=False)
#服务端登录认证
with open('userinformation','r',encoding='utf-8') as f:
for i in f:
if i.strip() == client_userinfo:
#登录成功以后,将服务端能够提供的功能,发送给客户端
conn.send(json_func_list.encode('utf-8'))
#接收客户端的文件描述信息,其中有一项是功能序号
client_fileinfo_json = conn.recv(1024).decode('utf-8')
client_file_info = json.loads(client_fileinfo_json)
print(client_file_info)
if client_file_info['func_choice'] == '1':
upload(conn,client_file_info)
break
else:
#登录错误的信息码
conn.send(b'201')
# 打印进度条
import time
for i in range(10):
str_p = i * '*'
print('
'+ str_p,end='')
time.sleep(0.5)