首页 > 代码库 > python网络编程socket模块实现ftp上传下载
python网络编程socket模块实现ftp上传下载
本实验实现ftp上传文件下载文件功能，并具有校验文件完整性，打印进度条功能,
主要练习socket,struct模块。
ftp用户文件存放在user.json文件中
user.json文件内容
{"lisi": "abcdef", "hyh": "123456"}
ftp客户端脚本ftpclient.py
#!/usr/bin/python
# --*-- coding: utf-8 --*--
import socket
import json
import time
import sys
import struct
from hashlib import md5
ftp_obj = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ftp_obj.connect((‘127.0.0.1‘, 8080))
def login_auth(ftp_obj):
"""用户登录三次认证,输入用户名密码跟服务器账号密码匹配返回True,否则返回False"""
count = 0
while count < 3:
user_name = input("请输入账户: ").strip()
user_passwd = input("请输入密码: ").strip()
if user_name and user_passwd:
ftp_obj.send(user_name.encode(‘utf-8‘))
ftp_obj.send(user_passwd.encode(‘utf-8‘))
else:
count += 1
continue
#time.sleep(10)
login_res_bytes = ftp_obj.recv(1024)
# print(login_res_bytes)
# print(login_res_bytes.decode(‘gbk‘))
login_res = login_res_bytes.decode(‘gbk‘)
if login_res == ‘True‘:
return True
else:
count += 1
continue
else:
return False
def progress_bar(num, total):
"""打印进度条"""
rate = num / total
rate_num = int(rate * 100)
r = ‘\r%s%d%%‘ % (‘#‘ * rate_num, rate_num,)
print(r)
def get_md5(data):
"""校验文件内容"""
m = md5()
m.update(data.encode(‘utf-8‘))
res = m.hexdigest()
return res
def ftp_cmd(ftp_obj):
"""上传下载命令"""
while True:
cmd = input("请输入命令(dir or put filepath or get filepath or exit)>>: ").strip()
if cmd == ‘exit‘:
sys.exit(0)
#print(cmd_list)
if cmd == ‘dir‘:
ftp_obj.send(cmd.encode(‘utf-8‘))
head_struct = ftp_obj.recv(4)
head_len = struct.unpack(‘i‘, head_struct)[0]
head_bytes = ftp_obj.recv(head_len)
head_json = head_bytes.decode(‘utf-8‘)
head_dict = json.loads(head_json)
total_size = head_dict[‘total_size‘]
recv_size = 0
data = b‘‘
while recv_size < total_size:
recv_data = ftp_obj.recv(1024)
data += recv_data
recv_size += len(recv_data)
print(data.decode(‘gbk‘))
else:
cmd_list = cmd.split()
if cmd_list[0] == ‘get‘:
ftp_obj.send(cmd.encode(‘utf-8‘))
head_struct = ftp_obj.recv(4)
head_len = struct.unpack(‘i‘, head_struct)[0]
head_bytes = ftp_obj.recv(head_len)
head_json = head_bytes.decode(‘utf-8‘)
head_dict = json.loads(head_json)
total_size = head_dict[‘total_size‘]
recv_size = 0
data = b‘‘
while recv_size < total_size:
recv_data = ftp_obj.recv(1024)
data += recv_data
recv_size += len(recv_data)
progress_bar(recv_size, total_size) #打印进度条，传入两个参数，第一个是接受的数据字节，第二个是数据头解包的数据总长度
with open(cmd_list[1], ‘w‘, encoding=‘utf-8‘) as f:
f.write(data.decode(‘utf-8‘))
check_md5 = get_md5(data.decode(‘utf-8‘)) #校验传输的内容
if head_dict[‘hashlib‘] == check_md5:
print("下载成功，文件内容完整")
else:
print("下载完成，文件内容不完整")
elif cmd_list[0] == ‘put‘:
ftp_obj.send(cmd.encode(‘utf-8‘))
with open(cmd_list[1], ‘r‘, encoding=‘utf-8‘) as f:
data = f.read()
check_put_md5 = get_md5(data)
head_dict = {‘filename‘: cmd_list[1], ‘hashlib‘: check_put_md5, ‘total_size‘: len(data)}
head_json = json.dumps(head_dict)
head_bytes = head_json.encode(‘utf-8‘)
ftp_obj.send(struct.pack(‘i‘, len(head_bytes)))
ftp_obj.send(head_bytes)
is_upload = ftp_obj.recv(10)
if is_upload.decode(‘gbk‘) == ‘True‘:
#ftp_obj.send(data.encode(‘utf-8‘))
with open(cmd_list[1], ‘r‘, encoding=‘utf-8‘) as f:
dataline = f.readlines()
data_len = 0
for i in dataline:
ftp_obj.send(i.encode(‘utf-8‘))
data_len += len(i)
time.sleep(0.1)
progress_bar(data_len, len(data))
print("上传成功")
is_check = ftp_obj.recv(10)
if is_check.decode(‘gbk‘) == ‘True‘:
print("文件上传完整")
else:
print("文件上传不完整")
else:
print("文件太大，超出磁盘限额")
continue
else:
print("命令错误，重重新输入")
continue
def main():
auth_res = login_auth(ftp_obj)
if auth_res:
ftp_cmd(ftp_obj)
else:
print("用户或密码不正确，退出程序")
sys.exit(1)
if __name__ == ‘__main__‘:
main()
ftp服务端脚本ftpserver.py
#!/usr/bin/python
# --*-- coding:utf-8
import socket
import json
import os
import subprocess
import struct
from hashlib import md5
import time
dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
hyh_size = 5 #‘hyh‘用户M空间限额5M
lisi_size = 10 #‘lisi‘用户空间限额10M
ftp_obj = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ftp_obj.bind((‘127.0.0.1‘, 8080))
ftp_obj.listen(5)
def size(user_home_dir):
"""计算用户家目录大小"""
# print(user_home_dir)
# print(os.listdir(user_home_dir))
file_size_bytes = 0
for i in os.listdir(user_home_dir):
file = user_home_dir + ‘\\‘ + i
file_size_bytes += os.path.getsize(file)
file_size_mbytes = file_size_bytes / (1024 * 1024)
return file_size_mbytes
def get_md5(data):
"""校验文件内容"""
m = md5()
m.update(data.encode(‘utf-8‘))
res = m.hexdigest()
return res
def ftp_load(conn, user_name):
"""接收客户端命令，并执行"""
user_home_dir = dir + ‘\\‘ + user_name
used_size = size(user_home_dir)
while True:
try:
print(‘开始接收发送数据‘)
cmd_res_bytes = conn.recv(1024)
cmd_res = cmd_res_bytes.decode(‘utf-8‘)
if cmd_res == ‘dir‘:
res = subprocess.Popen(cmd_res + ‘ ‘ + user_home_dir, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
err = res.stderr.read()
if err:
cmd_stdout = err
else:
cmd_stdout = res.stdout.read()
#print(cmd_stdout.decode(‘gbk‘))
head_dict = {‘dir‘: user_name, ‘hashlib‘: None, ‘total_size‘: len(cmd_stdout)}
head_json = json.dumps(head_dict)
head_bytes = head_json.encode(‘utf-8‘)
conn.send(struct.pack(‘i‘, len(head_bytes)))
conn.send(head_bytes)
conn.send(cmd_stdout)
else:
cmd_res_list = cmd_res.split() #获取命令和文件列表
user_home_file = user_home_dir + ‘\\‘ + cmd_res_list[1] #获取文件
if cmd_res_list[0] == ‘get‘:
with open(user_home_file, ‘r‘, encoding=‘utf-8‘) as f:
data = f.read()
check_md5 = get_md5(data) #校验文件内容md5
head_dict = {‘filename‘: cmd_res_list[1], ‘hashlib‘: check_md5, ‘total_size‘: len(data)}
head_json = json.dumps(head_dict)
head_bytes = head_json.encode(‘utf-8‘)
conn.send(struct.pack(‘i‘, len(head_bytes)))
conn.send(head_bytes)
with open(user_home_file, ‘r‘, encoding=‘utf-8‘) as f:
dataline = f.readlines()
for i in dataline:
time.sleep(0.1)
conn.send(i.encode(‘utf-8‘))
#conn.send(data.encode(‘utf-8‘))
else:
head_struct = conn.recv(4)
head_len = struct.unpack(‘i‘, head_struct)[0]
head_bytes = conn.recv(head_len)
head_json = head_bytes.decode(‘utf-8‘)
head_dict = json.loads(head_json)
total_size = head_dict[‘total_size‘]
if user_name == ‘hyh‘:
if hyh_size - used_size > total_size / (1024 * 1024):
conn.send(‘True‘.encode(‘utf-8‘))
recv_size = 0
data = b‘‘
while recv_size < total_size:
recv_data = conn.recv(1024)
data += recv_data
recv_size += len(recv_data)
with open(user_home_file, ‘w‘, encoding=‘utf-8‘) as f:
f.write(data.decode(‘utf-8‘))
check_put_md5 = get_md5(data.decode(‘utf-8‘))
if head_dict[‘hashlib‘] == check_put_md5:
conn.send(‘True‘.encode(‘utf-8‘))
continue
else:
conn.send(‘False‘.encode(‘utf-8‘))
continue
else:
conn.send(‘False‘.encode(‘utf-8‘))
continue
elif user_name == ‘lisi‘:
if lisi_size - used_size > total_size / (1024 * 1024):
conn.send(‘True‘.encode(‘utf-8‘))
recv_size = 0
data = b‘‘
while recv_size < total_size:
recv_data = conn.recv(1024)
data += recv_data
recv_size += len(recv_data)
with open(user_home_file, ‘w‘, encoding=‘utf-8‘) as f:
f.write(data.decode(‘utf-8‘))
continue
else:
conn.send(‘False‘.encode(‘utf-8‘))
continue
else:
continue
except Exception:
break
while True:
print(‘waitting to accept...‘)
conn, addr = ftp_obj.accept()
print(‘client: ‘, addr)
while True:
try:
user_name_bytes = conn.recv(30)
user_passwd_bytes = conn.recv(30)
user_name = user_name_bytes.decode(‘gbk‘)
user_passwd = user_passwd_bytes.decode(‘gbk‘)
with open(‘user.json‘, ‘r‘, encoding=‘utf-8‘) as f:
user_dict = json.loads(f.read())
if user_name in user_dict and user_dict[user_name] == user_passwd:
conn.send(‘True‘.encode(‘utf-8‘))
print("ftp连接成功")
ftp_load(conn, user_name) #ftp上传下载实现
else:
conn.send(‘False‘.encode(‘utf-8‘))
except Exception:
conn.send(‘False‘.encode(‘utf-8‘))
conn.close()
ftp_obj.close()本文出自 “linux技术” 博客，请务必保留此出处http://haoyonghui.blog.51cto.com/4278020/1941026
python网络编程socket模块实现ftp上传下载
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。