python 多用户在线的FTP程序
要求：
1、用户加密认证
2、允许同时多用户登录
3、每个用户有自己的家目录 ，且只能访问自己的家目录
4、对用户进行磁盘配额，每个用户的可用空间不同
5、允许用户在ftp server上随意切换目录
6、允许用户查看当前目录下文件
7、允许上传和下载文件，保证文件一致性
8、文件传输过程中显示进度条
9、附加功能：支持文件的断点续传
README:
设计说明
1、client连接server端需要验证账号密码，密码使用MD5加密传输。
2、用户信息保存在本地文件中，密码MD5加密存储。磁盘配额大小也保存在其中。
3、用户连接上来后，可以执行命令如下
目录变更：cd /cd dirname / cd . /cd ..
文件浏览：ls
文件删除：rm filename
目录增删：mkdir dirname /rmdir dirname
查看当前目录：pwd
查看当前目录大小： du
上传文件：put filename
下载文件：get filename
移动和重命名： mv filename/dirname filename/dirname
上传断点续传: newput filename
下载断点续传: newget filename
4、涉及到目录的操作，用户登录后，程序会给用户一个“锚位”----以用户名字命名的家目录，使用户无论怎么操作，都只能在这个目录底下。而在发给用户的目录信息时，隐去上层目录信息。
5、用户在创建时，磁盘配额大小默认是100M，在上传文件时，程序会计算当前目录大小加文件大小是否会超过配额上限。未超过，上传；超过，返回磁盘大小不够的信息。磁盘配额可通过用户管理程序修改。
6、文件上传和下载后都会进行MD5值比对，验证文件是否一致。
7、服务端和客户端都有显示进度条功能，启用该功能会降低文件传输速度，这是好看的代价。
8、文件断点续传，支持文件上传和下载断点续传。断点续传上传功能还会检测用户控件是否足够。（断点续传命令使用前面new+put/get命名，包含put/get所有功能，由于逻辑增多，代码复杂，特地保留原put/get，以备后用）。
暂且说到这，接下来是正式程序
wKiom1nNBZ3BkioMAAB6yDPSdzI380.jpg-wh_50
试运行截图
wKioL1nNBOjwhBEiAAGwpqCk5VA361.jpg-wh_50
代码如下：
1、服务端
server.conf
####用户端配置文件####
[DEFAULT]
logfile = ../log/server.log
usermgr_log = ../log/usermgr.log
upload_dir= ../user_files
db_dir = ../db
####日志文件位置####
[log]
logfile = ../log/server.log
usermgr_log = ../log/usermgr.log
####上传文件存放位置####
[upload]
upload_dir= ../user_files
####用户信息存放位置####
[db]
db_dir = ../dbmain.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketserver,os
from usermanagement import useropr
from server import MyTCPHandler
info = '''
1、启动服务器
2、进入用户管理
按q退出
'''
if __name__ == '__main__':
while True:
print(info)
choice = input('>>>:')
if choice == 'q':
exit()
elif choice == '1':
ip, port = '0.0.0.0', 9999
server = socketserver.ThreadingTCPServer((ip, port), MyTCPHandler)
server.serve_forever()
elif choice == '2':
useropr.interactive()
else:continueusermanagement
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#filename:usermanagement.py
import os,hashlib,time,pickle,shutil,configparser,logging
####读取配置文件####
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
config_file = os.path.join(base_dir, 'conf/server.conf')
cf = configparser.ConfigParser()
cf.read(config_file)
####设定日志目录####
if os.path.exists(cf.get('log','usermgr_log')):
logfile = cf.get('log', 'usermgr_log')
else:
logfile = os.path.join(base_dir,'log/usermgr.log')
####设定用户上传文件目录，这边用于创建用户家目录使用####
if os.path.exists(cf.get('upload','upload_dir')):
file_dir = cf.get('upload','upload_dir')
else:
file_dir = os.path.join(base_dir,'user_files')
####设定用户信息存储位置####
if os.path.exists(cf.get('db','db_dir')):
db_path = cf.get('db','db_dir')
else:
db_path = os.path.join(base_dir,'db')
def hashmd5(*args): ####用于加密密码信息
m = hashlib.md5()
m.update(str(*args).encode())
return m.hexdigest()
class useropr(object):
def __init__(self,user_name,passwd = '123456',phone_number=''):
self.user_name = user_name
self.id = time.strftime("%Y%m%d%H%M%S", time.localtime())
self.phone_number = phone_number
self.passwd = passwd
self.space_size = 104857600 ####初始分配100MB存储空间
self.member_level = 1 ####会员等级，初始为1，普通会员
@staticmethod ####使用静态方法，可以直接用类命调用，如user.search_user(username)，否则需要实例化一个对象后才能调用
def query_user(user_name): ####查询用户
db_filelist=os.listdir(db_path)
#print(db_filelist)
dict={}
for filename in db_filelist:
with open(os.path.join(db_path,filename),'rb') as f:
content=pickle.load(f)
#print(filename,content) ####开启会打印出所有用户信息
if content['username'] == user_name:
#print(filename, content)
dict={'filename':filename,'content':content}
return dict
def save_userinfo(self): ####保存用户信息
query_result = self.query_user(self.user_name) ####检查是否已存在同名用户，如果没有查询结果应该为None
if query_result == None:
user_info = {
'username':self.user_name,
'id':self.id,
'phonenumber':self.phone_number,
'passwd':hashmd5(self.passwd),
'spacesize':self.space_size,
'level':self.member_level
}
with open(os.path.join(db_path,self.id),'wb') as f:
pickle.dump(user_info,f)
print('用户信息保存完毕')
try: ####创建用户家目录
os.mkdir(os.path.join(file_dir, self.user_name))
print('用户目录创建成功！')
except Exception as e:
print('用户目录创建失败,',e)
else:
print('用户名重复，信息未保存')
@staticmethod
def change_info(user_name,**kwargs): ####修改信息
query_result = useropr.query_user(user_name) ####用于检测用户是否存在,不存在不处理
if query_result != None:
userinfo_filename = query_result['filename']
user_info = query_result['content']
print('before update:',user_info)
for key in kwargs:
if key in ('username','id'): ####用户名和ID不可更改
print(key,'项不可更改')
elif key in ('passwd','phonenumber','spacesize','level'): ####允许修改的键值
if key == 'passwd':
user_info[key] = hashmd5(kwargs[key]) ####加密密码保存
else:
user_info[key] = kwargs[key]
with open(os.path.join(db_path, userinfo_filename), 'wb') as f:
pickle.dump(user_info, f)
print(key,'项用户信息变更保存完毕')
else:
print('输入信息错误，',key,'项不存在')
print('after update:',user_info)
else:
print('用户不存在')
@staticmethod
def delete_user(user_name): ####删除用户
query_result = useropr.query_user(user_name) ####用于检测用户是否存在,不存在不处理
if query_result != None:
userinfo_filename = query_result['filename']
userfile_path=os.path.join(db_path, userinfo_filename)
os.remove(userfile_path)
query_result_again = useropr.query_user(user_name)
if query_result_again == None:
print('用户DB文件删除成功')
try:
shutil.rmtree(os.path.join(file_dir,user_name))
print('用户家目录删除成功')
except Exception as e:
print('用户家目录删除失败：',e)
else:
print('用户DB文件删除失败')
else:
print('用户不存在或者已经被删除')
@staticmethod
def query_alluser(): ####查询所有用户信息，用于调试使用
db_filelist=os.listdir(db_path)
for filename in db_filelist:
with open(os.path.join(db_path,filename),'rb') as f:
content=pickle.load(f)
print(filename,content)
@staticmethod
def interactive():
'''使用说明：
新增用户请输入类似： a=useropr(username,passwd)
a.save_userinfo()
查询用户请输入：useropr.query_user(username)
更改用户信息请输入：useropr.change_info(username,id=123,level=1,passwd=123,phonenumber=123),其中字典部分为可选项
用户删除请输入:useropr.delete_user(username)
'''
info='''
1、新增用户
2、查询用户
3、修改用户
4、删除用户
退出请按q
'''
#useropr.query_alluser() ####查询所有用户信息，调试用
while True:
print(info)
choice = input('请输入你的选择:').strip()
#print('operation choice: %s' % choice)
if choice == 'q':
exit()
else:
username = input('请输入用户名：').strip()
#print('username: %s' % username)
if username == '':
print('用户不能为空')
continue
elif choice == '1':
passwd = input('请输入密码：')
new_user = useropr(username, passwd)
new_user.save_userinfo()
elif choice == '2':
print(useropr.query_user(username))
elif choice == '3':
update_item = input('请输入要修改的项目，例如：level,passwd,phonenumber：')
print('update item: %s' % update_item)
update_value = input('请输入要修改的项目新值：')
useropr.change_info(username,**{update_item:update_value}) #### ‘**{}’ 不加**系统无法识别为字典。不能直接使用update_item=update_value，update_item会直接被当成key值，而不是其中的变量。
elif choice == '4':
useropr.delete_user(username)
else:
print('输入错误')
continue
if __name__ == '__main__':
useropr.interactive()server.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# filename:server.py
import socketserver, json, os, sys, time, shutil, configparser, logging
from usermanagement import useropr
####读取配置文件####
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
config_file = os.path.join(base_dir, 'conf/server.conf')
cf = configparser.ConfigParser()
cf.read(config_file)
####设定日志目录####
if os.path.exists(cf.get('log', 'logfile')):
logfile = cf.get('log', 'logfile')
else:
logfile = os.path.join(base_dir, 'log/server.log')
####设定用户上传文件目录####
if os.path.exists(cf.get('upload', 'upload_dir')):
file_dir = cf.get('upload', 'upload_dir')
else:
file_dir = os.path.join(base_dir, 'user_files')
####设置日志格式###
logging.basicConfig(level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename=logfile,
filemode='a+')
def TimeStampToTime(timestamp): ####输入timestamp格式化输出时间，输出格式如：2017-09-16 16:32:35
timeStruct = time.localtime(timestamp)
return time.strftime('%Y-%m-%d %H:%M:%S', timeStruct)
def ProcessBar(part, total): ####进度条模块，运行会导致程序变慢
if total != 0:
i = round(part * 100 / total)
sys.stdout.write(
'[' + '>' * i + '-' * (100 - i) + ']' + str(i) + '%' + ' ' * 3 + str(part) + '/' + str(total) + '\r')
sys.stdout.flush()
# if part == total:
# print()
class MyTCPHandler(socketserver.BaseRequestHandler):
def put(self, *args): ####接收客户端文件
# self.request.send(b'server have been ready to receive') ####发送ACK
cmd_dict = args[0]
filename = os.path.basename(cmd_dict['filename']) ####传输进来的文件名可能带有路径，将路径去掉
filesize = cmd_dict['filesize']
filemd5 = cmd_dict['filemd5']
override = cmd_dict['override']
receive_size = 0
file_path = os.path.join(self.position, filename)
if override != 'True' and os.path.exists(file_path): ####检测文件是否已经存在
self.request.send(b'file have exits, do nothing!')
else:
if os.path.isfile(file_path): ####如果文件已经存在，先删除，再计算磁盘空间大小
os.remove(file_path)
current_size = self.du() ####调用du查看用户磁盘空间大小，但是du命令的最后会发送一个结果信息给client，会和前面和后面的信息粘包，需要注意
self.request.recv(1024) ####接收客户端ack信号，防止粘包，代号：P01
print(self.user_spacesize, current_size, filesize)
if self.user_spacesize >= current_size + filesize:
self.request.send(b'begin') ####发送开始传输信号
fk = open(file_path, 'wb')
while filesize > receive_size:
if filesize - receive_size > 1024:
size = 1024
else:
size = filesize - receive_size
data = self.request.recv(size)
fk.write(data)
receive_size += len(data)
# print(receive_size,len(data)) ####打印每次接收的数据
# ProcessBar(receive_size, filesize) ####服务端进度条，不需要可以注释掉
fk.close()
receive_filemd5 = os.popen('md5sum %s' % file_path).read().split()[0]
print('\r\n', file_path, 'md5:', receive_filemd5, '原文件md5:', filemd5)
if receive_filemd5 == filemd5:
self.request.send(b'file received successfully!')
else:
self.request.send(b'Error, file received have problems!')
else:
self.request.send(
b'Error, disk space do not enough! Nothing done! Total: %d, current: %d, rest:%d, filesize:%d' % (
self.user_spacesize, current_size, self.user_spacesize - current_size, filesize))
def get(self, *args): ####发送给客户端文件
# print('get receive the cmd',args[0])
filename = args[0]['filename']
print(filename)
# self.request.send(b'server have been ready to send') ####发送ACK
file_path = os.path.join(self.position, filename)
if os.path.isfile(file_path):
filesize = os.path.getsize(file_path)
####直接调用系统命令取得MD5值，如果使用hashlib，需要写open打开文件-》read读取文件（可能文件大会很耗时）-》m.update计算三部，代码量更多，效率也低
filemd5 = os.popen('md5sum %s' % file_path).read().split()[0]
msg = {
'action': 'get',
'filename': filename,
'filesize': filesize,
'filemd5': filemd5,
'override': 'True'
}
print(msg)
self.request.send(json.dumps(msg).encode('utf-8'))
'''接下来发送文件给客户端'''
self.request.recv(1024) ####接收ACK信号，下一步发送文件
fk = open(file_path, 'rb')
send_size = 0
for line in fk:
send_size += len(line)
self.request.send(line)
# ProcessBar(send_size, filesize) ####服务端进度条，不需要可以注释掉
else:
print('文件传输完毕')
fk.close()
else:
print(file_path, '文件未找到')
self.request.send(json.dumps('Filenotfound').encode('utf-8'))
def newput(self, *args): ####接收客户端文件，具有断点续传功能
# self.request.send(b'server have been ready to receive') ####发送ACK
cmd_dict = args[0]
filename = os.path.basename(cmd_dict['filename']) ####传输进来的文件名可能带有路径，将路径去掉
filesize = cmd_dict['filesize']
filemd5 = cmd_dict['filemd5']
override = cmd_dict['override']
receive_size = 0
file_path = os.path.join(self.position, filename)
print(file_path,os.path.isdir(file_path))
if override != 'True' and os.path.exists(file_path): ####检测文件是否已经存在
if os.path.isdir(file_path):
self.request.send(b'file have exits, and is a directory, do nothing!')
elif os.path.isfile(file_path):
self.request.send(b'file have exits, do nothing!')
resume_signal = self.request.recv(1024) ####接收客户端发来的是否从文件断点续传的信号
if resume_signal == b'ready to resume from break point': ####执行断点续传功能
exits_file_size = os.path.getsize(file_path)
current_size = self.du()
time.sleep(0.5) ####防止粘包
print('用户空间上限：%d, 当前已用空间：%d, 已存在文件大小：%d, 上传文件大小：%d ' % (self.user_spacesize,current_size,exits_file_size,filesize))
if self.user_spacesize >= (current_size - exits_file_size + filesize): ####判断剩余空间是否足够
if exits_file_size < filesize:
receive_size = exits_file_size
print('服务器上已存在的文件大小为：',exits_file_size)
msg = {
'state': True,
'position': exits_file_size,
'content': 'ready to receive file'
}
self.request.send(json.dumps(msg).encode('utf-8'))
fk = open(file_path, 'ab+')
while filesize > receive_size:
if filesize - receive_size > 1024:
size = 1024
else:
size = filesize - receive_size
data = self.request.recv(size)
fk.write(data)
receive_size += len(data)
# print(receive_size,len(data)) ####打印每次接收的数据
# ProcessBar(receive_size, filesize) ####服务端进度条，不需要可以注释掉
fk.close()
receive_filemd5 = os.popen('md5sum %s' % file_path).read().split()[0]
print('\r\n', file_path, 'md5:', receive_filemd5, '原文件md5:', filemd5)
if receive_filemd5 == filemd5:
self.request.send(b'file received successfully!')
else:
self.request.send(b'Error, file received have problems!')
else: ####如果上传的文件小于当前服务器上的文件，则为同名但不同文件，不上传。实际还需要增加其他判断条件，判断是否为同一文件。
msg = {
'state': False,
'position': '',
'content': 'Error, file mismatch, do nothing!'
}
self.request.send(json.dumps(msg).encode('utf-8'))
else: ####如果续传后的用户空间大于上限，拒接续传
msg = {
'state': False,
'position':'',
'content':'Error, disk space do not enough! Nothing done! Total: %d, current: %d, rest:%d, need_size:%d' % (self.user_spacesize, current_size, self.user_spacesize - current_size, filesize - exits_file_size)
}
self.request.send(json.dumps(msg).encode('utf-8'))
else:
pass
else:
if os.path.isfile(file_path): ####如果文件已经存在，先删除，再计算磁盘空间大小
os.remove(file_path)
current_size = self.du() ####调用du查看用户磁盘空间大小，但是du命令的最后会发送一个结果信息给client，会和前面和后面的信息粘包，需要注意
self.request.recv(1024) ####接收客户端ack信号，防止粘包，代号：P01
print(self.user_spacesize, current_size, filesize)
if self.user_spacesize >= current_size + filesize:
self.request.send(b'begin') ####发送开始传输信号
fk = open(file_path, 'wb')
while filesize > receive_size:
if filesize - receive_size > 1024:
size = 1024
else:
size = filesize - receive_size
data = self.request.recv(size)
fk.write(data)
receive_size += len(data)
# print(receive_size,len(data)) ####打印每次接收的数据
# ProcessBar(receive_size, filesize) ####服务端进度条，不需要可以注释掉
fk.close()
receive_filemd5 = os.popen('md5sum %s' % file_path).read().split()[0]
print('\r\n', file_path, 'md5:', receive_filemd5, '原文件md5:', filemd5)
if receive_filemd5 == filemd5:
self.request.send(b'file received successfully!')
else:
self.request.send(b'Error, file received have problems!')
else:
self.request.send(
b'Error, disk space do not enough! Nothing done! Total: %d, current: %d, rest:%d, filesize:%d' % (
self.user_spacesize, current_size, self.user_spacesize - current_size, filesize))
def newget(self, *args): ####发送给客户端文件，具有断点续传功能
# print('get receive the cmd',args[0])
filename = args[0]['filename']
remote_local_filesize = args[0]['filesize']
print(filename)
# self.request.send(b'server have been ready to send') ####发送ACK
file_path = os.path.join(self.position, filename)
if os.path.isfile(file_path):
filesize = os.path.getsize(file_path)
####直接调用系统命令取得MD5值，如果使用hashlib，需要写open打开文件-》read读取文件（可能文件大会很耗时）-》m.update计算三部，代码量更多，效率也低
filemd5 = os.popen('md5sum %s' % file_path).read().split()[0]
msg = {
'action': 'newget',
'filename': filename,
'filesize': filesize,
'filemd5': filemd5,
'override': 'True'
}
print(msg)
self.request.send(json.dumps(msg).encode('utf-8'))
'''接下来发送文件给客户端'''
self.request.recv(1024) ####接收ACK信号，下一步发送文件
fk = open(file_path, 'rb')
fk.seek(remote_local_filesize,0)
send_size = remote_local_filesize
for line in fk:
send_size += len(line)
self.request.send(line)
# ProcessBar(send_size, filesize) ####服务端进度条，不需要可以注释掉
else:
print('文件传输完毕')
fk.close()
else:
print(file_path, '文件未找到')
self.request.send(json.dumps('Filenotfound').encode('utf-8'))
def pwd(self, *args):
current_position = self.position
result = current_position.replace(file_dir, '') ####截断目录信息，使用户只能看到自己的家目录信息
self.request.send(json.dumps(result).encode('utf-8'))
def ls(self, *args): ####列出当前目录下的所有文件信息，类型，字节数，生成时间。
result = ['%-20s%-7s%-10s%-23s' % ('filename', 'type', 'bytes', 'creationtime')] ####信息标题
for f in os.listdir(self.position):
type = 'unknown'
f_abspath = os.path.join(self.position, f) ####给出文件的绝对路径，不然程序会找不到文件
if os.path.isdir(f_abspath):
type = 'd'
elif os.path.isfile(f_abspath):
type = 'f'
result.append('%-20s%-7s%-10s%-23s' % (
f, type, os.path.getsize(f_abspath), TimeStampToTime(os.path.getctime(f_abspath))))
self.request.send(json.dumps(result).encode('utf-8'))
def du(self, *args):
'''统计纯文件和目录占用空间大小，结果小于在OS上使用du -s查询，因为有一些（例如'.','..'）隐藏文件未包含在内'''
totalsize = 0
if os.path.isdir(self.position):
dirsize, filesize = 0, 0
for root, dirs, files in os.walk(self.position):
for d_item in dirs: ####计算目录占用空间，Linux中每个目录占用4096bytes，实际上也可以按这个值来相加
if d_item != '':
dirsize += os.path.getsize(os.path.join(root, d_item))
for f_item in files: ####计算文件占用空间
if f_item != '':
filesize += os.path.getsize(os.path.join(root, f_item))
totalsize = dirsize + filesize
result = 'current directory total sizes: %d' % totalsize
else:
result = 'Error,%s is not path ,or path does not exist!' % self.position
self.request.send(json.dumps(result).encode('utf-8'))
return totalsize
def cd(self, *args):
print(*args)
user_homedir = os.path.join(file_dir, self.username)
cmd_dict = args[0]
error_tag = False
'''判断目录信息'''
if cmd_dict['dir'] == '':
self.position = user_homedir
elif cmd_dict['dir'] == '.' or cmd_dict['dir'] == '/' or '//' in cmd_dict['dir']: ####'.','/','//','///+'匹配
pass
elif cmd_dict['dir'] == '..':
if user_homedir != self.position and user_homedir in self.position: ####当前目录不是家目录，并且当前目录是家目录下的子目录
self.position = os.path.dirname(self.position)
elif '.' not in cmd_dict['dir'] and os.path.isdir(
os.path.join(self.position, cmd_dict['dir'])): ####'.' not in cmd_dict['dir'] 防止../..输入
self.position = os.path.join(self.position, cmd_dict['dir'])
else:
error_tag = True
'''发送结果'''
if error_tag:
result = 'Error,%s is not path here, or path does not exist!' % cmd_dict['dir']
self.request.send(json.dumps(result).encode('utf-8'))
else:
self.pwd()
def mv(self,*args):
print(*args)
try:
objectname = args[0]['objectname']
dstname = args[0]['dstname']
abs_objectname = os.path.join(self.position,objectname)
abs_dstname = os.path.join(self.position, dstname)
print(abs_objectname,abs_dstname,os.path.isfile(abs_objectname),os.path.isdir(abs_objectname),os.path.isdir(abs_dstname))
result = ''
if os.path.isfile(abs_objectname):
if os.path.isdir(abs_dstname) or not os.path.exists(abs_dstname):
shutil.move(abs_objectname, abs_dstname)
print('moving success')
result = 'moving success'
elif os.path.isfile(abs_dstname):
print('moving cancel, file has been exits')
result = 'moving cancel, file has been exits'
elif os.path.isdir(abs_objectname):
if os.path.isdir(abs_dstname) or not os.path.exists(abs_dstname):
shutil.move(abs_objectname, abs_dstname)
print('moving success')
result = 'moving success'
elif os.path.isfile(abs_dstname):
print('moving cancel, %s is file' % dstname)
result = 'moving cancel, %s is file' % dstname
else:
print('nothing done')
result = 'nothing done'
self.request.send(json.dumps(result).encode('utf-8'))
except Exception as e:
print(e)
result = 'moving fail,' + e
self.request.send(json.dumps(result).encode('utf-8'))
def mkdir(self, *args): ####创建目录
try:
dirname = args[0]['dirname']
if dirname.isalnum(): ####判断文件是否只有数字和字母
if os.path.exists(os.path.join(self.position, dirname)):
result = '%s have existed' % dirname
else:
os.mkdir(os.path.join(self.position, dirname))
result = '%s created succes' % dirname
else:
result = 'Illegal character %s, dirname can only by string and num here.' % dirname
except TypeError:
result = 'please input dirname'
self.request.send(json.dumps(result).encode('utf-8'))
def rm(self, *args): ####删除文件
filename = args[0]['filename']
confirm = args[0]['confirm']
file_abspath = os.path.join(self.position, filename)
if os.path.isfile(file_abspath):
if confirm == True:
os.remove(file_abspath)
result = '%s have been delete.' % filename
else:
result = 'Not file deleted'
elif os.path.isdir(file_abspath):
result = '%s is a dir, plsese using rmdir' % filename
else:
result = 'File %s not exist!' % filename
self.request.send(json.dumps(result).encode('utf-8'))
def rmdir(self, *args): ###删除目录
dirname = args[0]['dirname']
confirm = args[0]['confirm']
file_abspath = os.path.join(self.position, dirname)
if '.' in dirname or '/' in dirname: ####不能跨目录删除
result = 'should not rmdir %s this way' % dirname
elif os.path.isdir(file_abspath):
if confirm == True:
shutil.rmtree(file_abspath)
result = '%s have been delete.' % dirname
else:
result = 'Not file deleted'
elif os.path.isfile(file_abspath):
result = '%s is a file, not directory deleted' % dirname
else:
result = 'directory %s not exist!' % dirname
self.request.send(json.dumps(result).encode('utf-8'))
def auth(self):
self.data = json.loads(self.request.recv(1024).decode('utf-8'))
print(self.data)
recv_username = self.data['username']
recv_passwd = self.data['passwd']
query_result = useropr.query_user(recv_username)
print(query_result)
if query_result == None:
self.request.send(b'user does not exits')
elif query_result['content']['passwd'] == recv_passwd:
self.request.send(b'ok')
return query_result ####返回查询结果
elif query_result['content']['passwd'] != recv_passwd:
self.request.send(b'password error')
else:
self.request.send(b'unknown error')
def handle(self): ####处理类，调用以上方法
# self.position = file_dir
# print(self.position)
auth_tag = False
while auth_tag != True:
auth_result = self.auth() ####用户认证，如果通过，返回用户名，不通过为None
print('the authentication result is:', auth_result)
if auth_result != None:
self.username = auth_result['content']['username']
self.user_spacesize = auth_result['content']['spacesize']
auth_tag = True
print(self.username, self.user_spacesize)
user_homedir = os.path.join(file_dir, self.username)
if os.path.isdir(user_homedir):
self.position = user_homedir ####定锚，用户家目录
print(self.position)
while True:
print('当前连接：', self.client_address)
self.data = self.request.recv(1024).strip()
print(self.data)
logging.info(self.client_address)
if len(self.data) == 0:
print('客户端断开连接')
break ####检查发送来的命令是否为空
cmd_dict = json.loads(self.data.decode('utf-8'))
action = cmd_dict['action']
logging.info(cmd_dict)
if hasattr(self, action):
func = getattr(self, action)
func(cmd_dict)
else:
print('未支持指令：', action)
logging.info('current directory：%s' % self.position)
if __name__ == '__main__':
ip, port = '0.0.0.0', 9999
server = socketserver.ThreadingTCPServer((ip, port), MyTCPHandler)
server.serve_forever()2、客户端
client.conf
####用户端配置文件####
[DEFAULT]
logfile = ../log/client.log
download_dir= ../temp
####日志文件位置####
[log]
logfile = ../log/client.log
####下载文件存放位置####
[download]
download_dir= ../tempmain.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import configparser,os
from client import FtpClient
if __name__ == '__main__':
ftp = FtpClient()
ftp.connect('127.0.0.1',9999)
auth_tag=False
while auth_tag != True:
auth_tag=ftp.auth()
ftp.interactive()client.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# filename:client.py
import socket, json, os, sys, hashlib, getpass, logging, configparser,time
####读取配置文件####
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
config_file = os.path.join(base_dir, 'conf/client.conf')
cf = configparser.ConfigParser()
cf.read(config_file)
####设定日志目录####
if os.path.exists(cf.get('log', 'logfile')):
logfile = cf.get('log', 'logfile')
else:
logfile = os.path.join(base_dir, 'log/client.log')
####设定下载目录####
if os.path.exists(cf.get('download', 'download_dir')):
download_dir = cf.get('download', 'download_dir')
else:
download_dir = os.path.join(base_dir, 'temp')
####设置日志格式###
logging.basicConfig(level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename=logfile,
filemode='a+')
def hashmd5(*args): ####用于加密密码信息
m = hashlib.md5()
m.update(str(*args).encode())
return m.hexdigest()
def ProcessBar(part, total): ####进度条模块
if total != 0:
i = round(part * 100 / total)
sys.stdout.write(
'[' + '>' * i + '-' * (100 - i) + ']' + str(i) + '%' + ' ' * 3 + str(part) + '/' + str(total) + '\r')
sys.stdout.flush()
class FtpClient(object):
def __init__(self):
self.client = socket.socket()
def connect(self, ip, port):
self.client.connect((ip, port))
def exec_linux_cmd(self, dict): ####用于后面调用linux命令
logging.info(dict) ####将发送给服务端的命令保存到日志中
self.client.send(json.dumps(dict).encode('utf-8'))
server_response = json.loads(self.client.recv(4096).decode('utf-8'))
if isinstance(server_response, list):
for i in server_response:
print(i)
else:
print(server_response)
def help(self):
info = '''
仅支持如下命令：
ls
du
pwd
cd dirname/cd ./cd ..
mkdir dirname
rm filename
rmdir dirname
put filename
get filename
mv filename/dirname filename/dirname
newput filename (后续增加的新功能，支持断点续传)
newget filename (后续增加的新功能，支持断点续传)
'''
print(info)
def interactive(self):
while True:
self.pwd() ####打印当前目录位置
cmd = input('>>>:').strip()
if len(cmd) == 0: continue
action = cmd.split()[0]
if hasattr(self, action):
func = getattr(self, action)
func(cmd)
else:
self.help()
def put(self, *args): ####上传文件
cmd = args[0].split()
override = cmd[-1] ####override：是否覆盖参数,放在最后一位
if override != 'True':
override = 'False'
# print(cmd,override)
if len(cmd) > 1:
filename = cmd[1]
if os.path.isfile(filename):
filesize = os.path.getsize(filename)
filemd5 = os.popen('md5sum %s' % filename).read().split()[
0] ####直接调用系统命令取得MD5值，如果使用hashlib，需要写open打开文件-》read读取文件（可能文件大会很耗时）-》m.update计算三部，代码量更多，效率也低
msg = {
'action': 'put',
'filename': filename,
'filesize': filesize,
'filemd5': filemd5,
'override': override ####True ,or False
}
logging.info(msg)
self.client.send(json.dumps(msg).encode('utf-8'))
server_response = self.client.recv(1024) ####等待服务器确认信号，防止粘包
logging.info(server_response)
if server_response == b'file have exits, do nothing!':
override_tag = input('文件已存在，要覆盖文件请输入yes >>>:')
if override_tag == 'yes':
self.put('put %s True' % filename)
else:
print('文件未上传')
else:
self.client.send(b'client have ready to send') ####发送确认信号，防止粘包，代号：P01
server_response = self.client.recv(1024).decode('utf-8')
print(server_response) ####注意：用于打印服务器反馈信息，例如磁盘空间不足信息，不能取消
if server_response == 'begin':
fk = open(filename, 'rb')
send_size = 0
for line in fk:
# print(len(line))
send_size += len(line)
self.client.send(line)
ProcessBar(send_size, filesize)
else:
print('\r\n', '文件传输完毕')
fk.close()
server_response = self.client.recv(1024).decode('utf-8')
print(server_response)
else:
print('文件不存在')
else:
print('请输入文件名')
def get(self, *args): ####下载文件
cmd = args[0].split()
# print(args[0],cmd)
if len(cmd) > 1:
filename = cmd[1]
filepath = os.path.join(download_dir, filename)
if os.path.isfile(filepath): ####判断下载目录是否已存在同名文件
override_tag = input('文件已存在，要覆盖文件请输入yes >>>:').strip()
if override_tag == 'yes':
msg = {
'action': 'get',
'filename': filename,
'filesize': 0,
'filemd5': '',
'override': 'True'
}
logging.info(msg)
self.client.send(json.dumps(msg).encode('utf-8'))
server_response = json.loads(self.client.recv(1024).decode('utf-8'))
logging.info(server_response)
if server_response == 'Filenotfound':
print('File no found!')
else:
print(server_response)
self.client.send(b'client have been ready to receive') ####发送信号，防止粘包
filesize = server_response['filesize']
filemd5 = server_response['filemd5']
receive_size = 0
fk = open(filepath, 'wb')
while filesize > receive_size:
if filesize - receive_size > 1024:
size = 1024
else:
size = filesize - receive_size
data = self.client.recv(size)
fk.write(data)
receive_size += len(data)
# print(receive_size, len(data)) ####打印数据流情况
ProcessBar(receive_size, filesize) ####打印进度条
fk.close()
receive_filemd5 = os.popen('md5sum %s' % filepath).read().split()[0]
print('\r\n', filename, 'md5:', receive_filemd5, '原文件md5:', filemd5)
if receive_filemd5 == filemd5:
print('文件接收完成！')
else:
print('Error,文件接收异常！')
else:
print('下载取消')
else:
print('请输入文件名')
def newput(self, *args): ####上传文件，具有断点续传功能
cmd = args[0].split()
override = cmd[-1] ####override：是否覆盖参数,放在最后一位
if override != 'True':
override = 'False'
# print(cmd,override)
if len(cmd) > 1:
filename = cmd[1]
if os.path.isfile(filename):
filesize = os.path.getsize(filename)
filemd5 = os.popen('md5sum %s' % filename).read().split()[
0] ####直接调用系统命令取得MD5值，如果使用hashlib，需要写open打开文件-》read读取文件（可能文件大会很耗时）-》m.update计算三部，代码量更多，效率也低
msg = {
'action': 'newput',
'filename': filename,
'filesize': filesize,
'filemd5': filemd5,
'override': override ####True ,or False
}
logging.info(msg)
self.client.send(json.dumps(msg).encode('utf-8'))
server_response = self.client.recv(1024) ####等待服务器确认信号，防止粘包
logging.info(server_response)
print(server_response)
if server_response == b'file have exits, and is a directory, do nothing!':
print('文件已存在且为目录，请先修改文件或目录名字，然后再上传')
elif server_response == b'file have exits, do nothing!':
override_tag = input('文件已存在，要覆盖文件请输入yes，要断点续传请输入r >>>:').strip()
if override_tag == 'yes':
self.client.send(b'no need to do anything') ####服务端在等待是否续传的信号，发送给服务端确认(功能号：s1)
time.sleep(0.5) ####防止黏贴
self.put('put %s True' % filename)
elif override_tag == 'r':
self.client.send(b'ready to resume from break point') ####服务端在等待是否续传的信号，发送给服务端确认(功能号：s1)
self.client.recv(1024) ####这边接收服务端发送过来的du信息，不显示，直接丢弃
server_response = json.loads((self.client.recv(1024)).decode())
print(server_response)
if server_response['state'] == True:
exits_file_size = server_response['position']
fk = open(filename, 'rb')
fk.seek(exits_file_size,0)
send_size = exits_file_size
for line in fk:
# print(len(line))
send_size += len(line)
self.client.send(line)
ProcessBar(send_size, filesize)
else:
print('\r\n', '文件传输完毕')
fk.close()
server_response = self.client.recv(1024).decode('utf-8')
print(server_response)
else:
print(server_response['content'])
else:
self.client.send(b'no need to do anything') ####服务端在等待是否续传的信号，发送给服务端确认(功能号：s1)
print('文件未上传')
else:
self.client.send(b'client have ready to send') ####发送确认信号，防止粘包，代号：P01
server_response = self.client.recv(1024).decode('utf-8')
print(server_response) ####注意：用于打印服务器反馈信息，例如磁盘空间不足信息，不能取消
if server_response == 'begin':
fk = open(filename, 'rb')
send_size = 0
for line in fk:
# print(len(line))
send_size += len(line)
self.client.send(line)
ProcessBar(send_size, filesize)
else:
print('\r\n', '文件传输完毕')
fk.close()
server_response = self.client.recv(1024).decode('utf-8')
print(server_response)
else:
print('文件不存在')
else:
print('请输入文件名')
def newget(self, *args): ####下载文件，具有断点续传功能
cmd = args[0].split()
# print(args[0],cmd)
if len(cmd) > 1:
filename = cmd[1]
filepath = os.path.join(download_dir, filename)
transfer_tag = True ####传输控制信号，默认True为下载
resume_tag = False ####断点续传信号
local_filesize = 0 ####本地文件大小，后面判断是否有同名文件使用
if os.path.isfile(filepath): ####判断下载目录是否已存在同名文件
override_tag = input('文件已存在，要覆盖文件请输入yes，要断点续传请输入r >>>:').strip()
if override_tag == 'yes':
pass
elif override_tag == 'r':
local_filesize = os.path.getsize(filepath)
resume_tag = True
else:
print('下载取消')
transfer_tag = False
if transfer_tag == True:
msg = {
'action': 'newget',
'filename': filename,
'filesize': local_filesize,
'filemd5': '',
'override': 'True'
}
logging.info(msg)
self.client.send(json.dumps(msg).encode('utf-8'))
server_response = json.loads(self.client.recv(1024).decode('utf-8'))
logging.info(server_response)
if server_response == 'Filenotfound':
print('File no found!')
else:
print(server_response)
self.client.send(b'client have been ready to receive') ####发送信号，防止粘包
filesize = server_response['filesize']
filemd5 = server_response['filemd5']
receive_size = local_filesize
if resume_tag == True:
fk = open(filepath, 'ab+') ####用于断点续传
else:
fk = open(filepath, 'wb+') ####用于覆盖或者新生成文件
while filesize > receive_size:
if filesize - receive_size > 1024:
size = 1024
else:
size = filesize - receive_size
data = self.client.recv(size)
fk.write(data)
receive_size += len(data)
# print(receive_size, len(data)) ####打印数据流情况
ProcessBar(receive_size, filesize) ####打印进度条
fk.close()
receive_filemd5 = os.popen('md5sum %s' % filepath).read().split()[0]
print('\r\n', filename, 'md5:', receive_filemd5, '原文件md5:', filemd5)
if receive_filemd5 == filemd5:
print('文件接收完成！')
else:
print('Error,文件接收异常！')
else:
print('请输入文件名')
def pwd(self, *args): ####查看用户目录
msg = {
'action': 'pwd',
}
self.exec_linux_cmd(msg)
def ls(self, *args): ####查看文件信息
msg = {
'action': 'ls',
}
self.exec_linux_cmd(msg)
def du(self, *args): ####查看当前目录大小
msg = {
'action': 'du',
}
self.exec_linux_cmd(msg)
def cd(self, *args): ####切换目录
try: ####如果是直接输入cd,dirname=''
dirname = args[0].split()[1]
except IndexError:
dirname = ''
msg = {
'action': 'cd',
'dir': dirname
}
self.exec_linux_cmd(msg)
def mkdir(self, *args): ####生成目录
try: ####如果是直接输入rm,跳出
dirname = args[0].split()[1]
msg = {
'action': 'mkdir',
'dirname': dirname,
}
self.exec_linux_cmd(msg)
except IndexError:
print('Not dirname input, do nothing.')
pass
def rm(self, *args): ####删除文件
try: ####如果是直接输入rm,跳出
filename = args[0].split()[1]
msg = {
'action': 'rm',
'filename': filename,
'confirm': True ####确认是否直接删除标志
}
self.exec_linux_cmd(msg)
except IndexError:
print('Not filename input, do nothing.')
pass
def rmdir(self, *args):
try: ####如果是直接输入rm,跳出
dirname = args[0].split()[1]
msg = {
'action': 'rmdir',
'dirname': dirname,
'confirm': True ####确认是否直接删除标志
}
self.exec_linux_cmd(msg)
except IndexError:
print('Not dirname input, do nothing.')
pass
def mv(self,*args): ####实现功能：移动文件，移动目录，文件重命名，目录重命名
try:
objectname = args[0].split()[1]
dstname = args[0].split()[2]
msg = {
'action': 'mv',
'objectname': objectname,
'dstname': dstname
}
print(msg)
self.exec_linux_cmd(msg)
except Exception as e:
print(e)
pass
def auth(self):
user_name = input('请输入用户名>>>:').strip()
passwd = getpass.getpass('请输入密码>>>：').strip() ####在linux上输入密码不显示
msg = {
'username': user_name,
'passwd': hashmd5(passwd)
}
self.client.send(json.dumps(msg).encode('utf-8'))
server_response = self.client.recv(1024).decode('utf-8')
if server_response == 'ok':
print('认证通过！')
return True
else:
print(server_response)
return False
if __name__ == '__main__':
ftp = FtpClient()
ftp.connect('127.0.0.1', 9999)
auth_tag = False
while auth_tag != True:
auth_tag = ftp.auth()
ftp.interactive()注:配置文件中的中文注释，可能会使程序在启动时报出ASCII decode error，可以去掉。
另外服务端最好在Linux下启动，我在windows下启动日志输出模块会报错。
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2019-09-07 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
相关产品与服务
对象存储
对象存储（Cloud Object Storage，COS）是由腾讯云推出的无目录层次结构、无数据格式限制，可容纳海量数据且支持 HTTP/HTTPS 协议访问的分布式存储服务。腾讯云 COS 的存储桶空间无容量上限，无需分区管理，适用于 CDN 数据分发、数据万象处理或大数据计算与分析的数据湖等多种场景。