當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

实现ftp_FTP文件服务器的实现

發(fā)布時(shí)間：2025/1/21 编程问答 24 豆豆

生活随笔收集整理的這篇文章主要介紹了实现ftp_FTP文件服务器的实现小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

FTP文件服務(wù)器

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

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

在現(xiàn)實(shí)生活中，常用的文件服務(wù)器有哪些了？

Xftp,WinScp,FileZilla

如何實(shí)現(xiàn)FTP文件服務(wù)器

""" 我們需要一個(gè)客戶端，和一個(gè)服務(wù)端（基本需求）： 1、客戶端用來發(fā)送服務(wù)請求 2、服務(wù)端用來處理服務(wù)請求，并返回相應(yīng)的數(shù)據(jù)給客戶端 3、使用socket來實(shí)現(xiàn)客戶端和服務(wù)端之間的通信 """

實(shí)驗(yàn)準(zhǔn)備

為了模擬并發(fā)的效果，我們準(zhǔn)備兩個(gè)客戶端，一個(gè)服務(wù)端

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

主要實(shí)現(xiàn)功能

""" 1.文件上傳 2.文件下載 3.文件MD5校驗(yàn) 4.文件進(jìn)度條顯示 """

使用模塊

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:遠(yuǎn)程路徑，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))# 將報(bào)頭傳給服務(wù)端self.client.send(head_struct)self.client.send(head_bytes) ?def show_remote_path(self, remote_path=None):"""獲取遠(yuǎn)端的文件夾，包括文件和文件夾"""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('獲取遠(yuǎn)程路徑'.center(60, '='))print(f"{file_dic.get('base_path')}")table_list = ['名字', '屬性', '大小', '創(chuàng)建日期']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("請選擇相應(yīng)的文件下載,或者通過選相應(yīng)的路徑切換路徑:")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 = ['功能選項(xiàng)', '功能名稱', '操作示例']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('是否繼續(xù)進(jìn)行操作(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()

服務(wù)端代碼

基于socketserver實(shí)現(xiàn)并發(fā)

""" FTP服務(wù)端 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)) ?# 發(fā)送報(bào)頭消息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實(shí)現(xiàn)并發(fā)，容易遇到地址占用問題：

解決方案：

""" 1.使用Thread實(shí)現(xiàn)并發(fā) 2.使用 setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)解決地址占用問題 """

基于Thread實(shí)現(xiàn)并發(fā)

""" FTP服務(wù)端 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：

服務(wù)端：

下載功能