FTP文件服務器

""" FTP服務器(File Transfer Protocol Server)是在互聯網上提供文件存儲和訪問服務的計算機，它們依照FTP協議提供服務。 FTP是File Transfer Protocol(文件傳輸協議)。顧名思義，就是專門用來傳輸文件的協議。簡單地說，支持FTP協議的服務器就是FTP服務器。 """

從上述定義中，我們知道了文件服務器的兩個核心功能：上傳和下載。

在現實生活中，常用的文件服務器有哪些了？

Xftp,WinScp,FileZilla

如何實現FTP文件服務器

""" 我們需要一個客戶端，和一個服務端（基本需求）： 1、客戶端用來發送服務請求 2、服務端用來處理服務請求，并返回相應的數據給客戶端 3、使用socket來實現客戶端和服務端之間的通信 """

實驗準備

為了模擬并發的效果，我們準備兩個客戶端，一個服務端

""" 服務端：阿里云服務器(172.21.44.11) 客戶端1：本地虛擬機1(192.168.80.129) 客戶端2：本地虛擬機1(192.168.80.130) """

主要實現功能

""" 1.文件上傳 2.文件下載 3.文件MD5校驗 4.文件進度條顯示 """

使用模塊

import os import sys import struct import json import hashlib import argparse from prettytable import PrettyTable from socket import *

客戶端代碼

""" ? FTP客戶端 time:2020/4/21 version:1.0 coder:surpass ? """ ? import os import sys import struct import json import hashlib import argparse from prettytable import PrettyTable from socket import * ? func_dic = {'upload': ['上傳文件', 'upload'],'download': ['下載文件', 'download'] } ? base_path = os.path.dirname(os.path.abspath(__file__)) ? ? def get_file_md5(file_path):if os.path.isfile(file_path):size = os.path.getsize(file_path) // 5m = hashlib.md5()with open(file_path, mode='rb') as f:for i in range(4):m.update(f.read(5))f.seek(size, 1)return m.hexdigest() ? ? def check_file_md5(filename, file_md5):size = os.path.getsize(filename) // 5m = hashlib.md5()with open(filename, mode='rb') as f:for i in range(4):m.update(f.read(5))f.seek(size, 1)if m.hexdigest() == file_md5:return True ? ? def get_pretty_tb(table_list, *args):tb = PrettyTable(field_names=table_list)for item in args:tb.add_row(item)return tb ? ? class MYFTPClient:def __init__(self, ip_address, buf_size=1024):self.ip_address = ip_addressself.buf_size = buf_sizeself.client = socket(AF_INET, SOCK_STREAM)self.client_connect() ?def client_connect(self):try:self.client.connect_ex(self.ip_address)except Exception as e:sys.stderr.write(f'connect error:{e}') ?def client_close(self):self.client.close() ?def send_head_pack(self, cmd=None, file=None, remote_path=None, file_path=None,filename=None, file_size=None, file_md5=None):"""cmd:操作命令，file:文件或文件夾集合，remote_path:遠程路徑，file_path：文件路徑filename:文件名，file_size:文件大小，file_md5:文件的md5值"""head_dic = {'cmd': cmd, 'file': file, 'remote_path': remote_path, 'file_path': file_path,'filename': filename, 'file_size': file_size, 'file_md5': file_md5}head_dic = {k: v for k, v in head_dic.items() if v is not None}# print(head_dic)head_json = json.dumps(head_dic)head_bytes = bytes(head_json, encoding='utf-8')head_struct = struct.pack('i', len(head_bytes))# 將報頭傳給服務端self.client.send(head_struct)self.client.send(head_bytes) ?def show_remote_path(self, remote_path=None):"""獲取遠端的文件夾，包括文件和文件夾"""self.send_head_pack(cmd='show_remote_path', file={}, remote_path=remote_path) ?recv_size = struct.unpack('i', self.client.recv(4))[0]file_dic = json.loads(self.client.recv(recv_size).decode('utf-8')) ?print('獲取遠程路徑'.center(60, '='))print(f"{file_dic.get('base_path')}")table_list = ['名字', '屬性', '大小', '創建日期']result = []file_list = []dir_list = []for item in file_dic['file_list']:item.insert(1, 'file')result.append(item)file_list.append(os.path.join(file_dic.get('base_path'), item[0]))for obj in file_dic['dir_list']:obj.insert(1, 'dir')result.append(obj)dir_list.append(os.path.join(file_dic.get('base_path'), obj[0]))tb_show = get_pretty_tb(table_list, *result)print(tb_show)dir_list.append(os.path.dirname(file_dic.get('base_path')))return file_list, dir_list ?def show_progress(self, func=None, filename=None, percent=None):res = int(50 * percent) * '#'sys.stdout.write('r%s %s[%-50s] %d%%' % (func, filename, res, int(100 * percent)))sys.stdout.flush() ?def upload(self, msg):cmd = msgwhile 1:filepath = input('請輸入上傳文件的路徑>>:').strip()if not filepath: continuebreakfilename = os.path.basename(filepath)file_size = os.path.getsize(filepath)file_md5 = get_file_md5(filepath)if not file_md5:raise FileNotFoundError(f'file does not exits:{filepath}')self.send_head_pack(cmd=cmd, filename=filename, file_size=file_size, file_md5=file_md5)current_size = 0with open(filepath, mode='rb') as f:for line in f:current_size += len(line)self.show_progress('upload', filename, current_size // file_size)self.client.send(line)res = self.client.recv(4)msg_len = struct.unpack('i', res)[0]msg = self.client.recv(msg_len).decode('utf-8')print(f'n{msg}') ?def download(self, msg):cmd = msgfile_all = self.show_remote_path()while 1:choice = input("請選擇相應的文件下載,或者通過選相應的路徑切換路徑:")if choice in file_all[1]:file_all = self.show_remote_path(choice)continueelif choice in file_all[0]:self.send_head_pack(cmd=cmd, file_path=choice, filename=os.path.basename(choice)) ?file_struct = struct.unpack('i', self.client.recv(4))[0]file_json = self.client.recv(file_struct).decode('utf-8')file_dic = json.loads(file_json)# print(file_dic)filename = file_dic.get('filename')file_size = file_dic.get('file_size')download_size = 0while download_size < file_size:content = self.client.recv(1024)download_size += len(content)self.show_progress('download', filename, download_size // file_size)with open(os.path.join(base_path, filename), mode='ab+') as fw:fw.write(content)else:file_path = os.path.join(base_path, filename)if check_file_md5(file_path, file_dic.get('file_md5')):sys.stdout.write(f'ndownload {filename} success!n')else:os.remove(file_path)sys.stderr.write(f'nfiled to download {filename}!n')breakelse:print('請輸入正確的文件或路徑！') ?def run(self):while 1:func_table_list = ['功能選項', '功能名稱', '操作示例']result = [(lambda x, y, z: [x, y, z])(k, v[0], v[1]) for k, v in func_dic.items()]tb_func = get_pretty_tb(func_table_list, *result)msg = input(f'{tb_func}nplease enter the operate signal>>:').strip()if not msg: continueif hasattr(self, msg):func = getattr(self, msg)func(msg)_continue = input('是否繼續進行操作(y,n):').strip().lower()if _continue == 'n': breakself.client_close() ? ? def main():parser = argparse.ArgumentParser(description='MY FTP Client v1.0')parser.add_argument('--ip', action='store', dest='ip', required=True, help='The remote ip to connect')parser.add_argument('--port', action='store', dest='port', required=True, type=int, help='The port to connect')parser.add_argument('--buf_size', action='store', dest='buf_size', type=int,help='the max size(one time) to upload/download')args = parser.parse_args()if args.buf_size:client = MYFTPClient((args.ip, args.port), args.buf_size)else:client = MYFTPClient((args.ip, args.port))client.run() ? ? if __name__ == '__main__':main()

服務端代碼

基于socketserver實現并發

""" FTP服務端 time:2020/4/21 version:v1.0 coder:surpass """ ? import os import sys import json import struct import time import hashlib import socketserver ? IP_ADDRESS = ('0.0.0.0', 9090) BUF_SIZE = 1024 ? ? def get_file_md5(file_path):if os.path.isfile(file_path):size = os.path.getsize(file_path) // 5m = hashlib.md5()with open(file_path, mode='rb') as f:for i in range(4):m.update(f.read(5))f.seek(size, 1)return m.hexdigest() ? ? def check_file_md5(filename, file_md5):size = os.path.getsize(filename) // 5m = hashlib.md5()with open(filename, mode='rb') as f:for i in range(4):m.update(f.read(5))f.seek(size, 1)if m.hexdigest() == file_md5:return True ? ? class MYFTPServer(socketserver.BaseRequestHandler):def handle(self):while 1:try:head_struct = self.request.recv(4)if not head_struct: breakhead_len = struct.unpack('i', head_struct)[0]head_dic_json = self.request.recv(head_len).decode('utf-8')head_dic = json.loads(head_dic_json)if hasattr(self, head_dic.get('cmd')):func = getattr(self, head_dic.get('cmd'))func(head_dic)except Exception as e:sys.stderr.write(f'error message:{e}')self.request.close() ?def show_remote_path(self, head_dic):if not head_dic.get('remote_path'):base_path = os.path.dirname(os.path.abspath(__file__))else:base_path = head_dic.get('remote_path')head_dic['file']['base_path'] = base_pathhead_dic['file']['file_list'] = []head_dic['file']['dir_list'] = []for obj in os.listdir(base_path):obj_abs_path = os.path.join(base_path, obj)obj_name = os.path.basename(obj_abs_path)obj_size = os.path.getsize(obj_abs_path)timestamp = os.path.getmtime(obj_abs_path)obj_mtime = time.strftime('%Y/%m/%d %X', time.localtime(timestamp))if os.path.isfile(obj_abs_path):head_dic['file']['file_list'].append([obj_name, obj_size, obj_mtime])elif os.path.isdir(obj_abs_path):obj_size = ''head_dic['file']['dir_list'].append([obj_name, obj_size, obj_mtime])head_json = json.dumps(head_dic.get('file'))head_bytes = bytes(head_json, encoding='utf-8')head_struct = struct.pack('i', len(head_bytes)) ?self.request.send(head_struct)self.request.send(head_json_bytes) ?def upload(self, head_dic):file_size = head_dic.get('file_size')recv_size = 0while recv_size < file_size:content = self.request.recv(BUF_SIZE)recv_size += len(content)with open(head_dic.get('filename'), mode='ab+') as f:f.write(content)else:if check_file_md5(head_dic.get('filename'), head_dic.get('file_md5')):msg = f"upload file {head_dic.get('filename')} success!"else:os.remove(head_dic.get('filename'))msg = f"failed to upload file,file {head_dic.get('filename')} is damaged!"msg_len = struct.pack('i', len(msg.encode('utf-8')))self.request.send(msg_len)self.request.send(msg.encode('utf-8')) ?def download(self, head_dic):file_path = head_dic.get('file_path')filename = os.path.basename(file_path)file_md5 = get_file_md5(file_path)file_size = os.path.getsize(file_path)file_dic = {'filename': filename, 'file_size': file_size, 'file_md5': file_md5}file_json = json.dumps(file_dic)file_bytes = bytes(file_json, encoding='utf-8')file_struct = struct.pack('i', len(file_bytes)) ?# 發送報頭消息self.request.send(file_struct)self.request.send(file_bytes) ?with open(file_path, mode='rb') as fr:for line in fr:self.request.send(line) ? ? def main():server = socketserver.ThreadingTCPServer(IP_ADDRESS, MYFTPServer)server.serve_forever() ? ? if __name__ == '__main__':main()

但是采用socketserver實現并發，容易遇到地址占用問題：

解決方案：

""" 1.使用Thread實現并發 2.使用 setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)解決地址占用問題 """

基于Thread實現并發

""" FTP服務端 time:2020/4/24 version:v2.0 coder:surpass """ ? import os import sys import json import struct import time import hashlib from socket import * from threading import Thread ? IP_ADDRESS = ('0.0.0.0', 9090) BUF_SIZE = 1024 BACK_LOG = 5 ? ? class MYFTPServer:def __init__(self, ip_address):self.ip_address = ip_addressself.server = socket(AF_INET, SOCK_STREAM) ?def bind(self):self.server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)self.server.bind(self.ip_address) ?def listen(self):self.server.listen(BACK_LOG) ?def accept(self):return self.server.accept() ?@staticmethoddef show_remote_path(head_dic, conn):if not head_dic.get('remote_path'):remote_path = os.path.dirname(os.path.abspath(__file__))else:remote_path = head_dic.get('remote_path')head_dic['file']['base_path'] = remote_pathhead_dic['file']['file_list'] = []head_dic['file']['dir_list'] = []for item in os.listdir(remote_path):obj_path = os.path.join(remote_path, item)obj_mtime = os.path.getmtime(obj_path)if os.path.isfile(obj_path):head_dic['file']['file_list'].append(os.path.basename(obj_path),os.path.getsize(obj_path),time.strftime('%Y-%m-%d %X', time.localtime(obj_mtime)))elif os.path.isdir(obj_path):head_dic['file']['dir_list'].append(os.path.basename(obj_path), '',time.strftime('%Y-%m-%d %X', time.localtime(obj_mtime)))head_json = json.dumps(head_dic['file'])head_bytes = bytes(head_json, encoding='utf-8')head_struct = struct.pack('i', len(head_bytes)) ?conn.send(head_struct)conn.send(head_bytes) ?@staticmethoddef check_file_md5(filename, file_size, file_md5):m = hashlib.md5()move_size = file_size // 5with open(filename, mode='rb') as fr:for i in range(4):m.update(fr.read(5))fr.seek(move_size, 1)if m.hexdigest() == file_md5:return True ?@staticmethoddef get_file_md5(file_path, file_size):m = hashlib.md5()move_size = file_size // 5with open(file_path, mode='rb') as fr:for i in range(4):m.update(fr.read(5))fr.seek(move_size, 1)return m.hexdigest() ?def upload(self, head_dic, conn):recv_size = 0file_size = head_dic.get('file_size')while recv_size < file_size:content = conn.recv(BUF_SIZE)recv_size += len(content)with open(head_dic.get('filename'), mode='ab+') as fw:fw.write(content)else:filename = head_dic.get('filename')file_size = head_dic.get('file_size')file_md5 = head_dic.get('file_md5')if self.check_file_md5(filename, file_size, file_md5):msg = f'upload file {filename} success!'else:os.remove(head_dic.get('file_path'))msg = f'failed to upload file {filename}!'msg_struct = struct.pack('i', len(msg.encode('utf-8')))conn.send(msg_struct)conn.send(msg.encode('utf-8')) ?def download(self, head_dic, conn):file_path = head_dic.get('file_path')file_size = os.path.getsize(file_path)filename = head_dic.get('filename')file_md5 = self.get_file_md5(file_path, file_size)head_dic = {'filename': filename, 'file_size': file_size, 'file_md5': file_md5}head_json = json.dumps(head_dic)head_bytes = bytes(head_json, encoding='utf-8')head_struct = struct.pack('i', len(head_bytes))conn.send(head_struct)conn.send(head_bytes) ?with open(file_path, mode='rb') as fr:for line in fr:conn.send(line) ?def communicate(self, conn, addr):print(f'收到客戶端[{addr[0]}:{addr[1]}]的信息...')while 1:try:head_struct = conn.recv(4)if not head_struct: breakhead_len = struct.unpack('i', head_struct)[0]head_bytes = conn.recv(head_len)head_dic = json.loads(head_bytes.decode('utf-8'))if hasattr(self, head_dic.get('cmd')):func = getattr(self, head_dic.get('cmd'))func(head_dic, conn)except ConnectionResetError as e:sys.stderr.write(f'connect error:{e}')conn.close() ?def run(self):self.bind()self.listen()while 1:conn, addr = self.accept()t = Thread(target=self.communicate, args=(conn, addr))t.start() ? ? if __name__ == '__main__':server = MYFTPServer(IP_ADDRESS)server.run()

上傳功能

客戶端1：

客戶端2：

服務端：

下載功能

客戶端1：

客戶端2：

服務端：

總結

以上是生活随笔為你收集整理的实现ftp_FTP文件服务器的实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。