Python基于UDP协议实现轻量级FTP文件传输服务

应用场景与通信机制

在内网环境中，有时需要绕过物理介质（如U盘）直接进行设备间的数据流转。本方案利用Python标准库中的 socket 模块，基于无连接的UDP协议构建简易FTP服务端与客户端。双方约定以自定义的字典结构作为通信载体，通过 pickle.dumps() 与 pickle.loads() 完成对象到字节流的序列化与反序列化，从而实现在局域网内的目录浏览、路径切换及文件定向下载。

支持的客户端指令如下：

• ll 或 ls：列出当前目录下的所有文件或目录
• pwd：显示当前所在的相对路径
• get 文件名：下载指定文件至客户端配置的本地目录
• get 目录 / get all：递归下载指定文件夹或当前目录全部内容
• cd / cd .. / cd 指定目录：切换客户端逻辑访问路径

实际启动后，服务端会将绑定的根目录暴露给客户端，客户端执行指令即可返回结构化结果。服务端运行时的监听状态如下所示：

代码实现与操作步骤

第1步：准备基础文件处理工具类 file_handler.py

该模块封装了文件系统的基础操作，供服务端调用。核心功能包括获取目录下的文件元数据列表，以及按偏移量分段读取二进制文件内容。首次使用前需确保同级目录下已创建此文件。

import os
import logging
import traceback
LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s"
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
_logger = logging.getLogger()
class FileHandler:
    def __init__(self, logger=None):
        self.__logger = logger if logger else _logger
    def list_dir(self, dir_path: str, return_absolute_path=False) -> list:
        ret_list = []
        try:
            if os.path.exists(dir_path):
                ret = os.listdir(dir_path)
                for item in ret:
                    full_path = os.path.join(dir_path, item)
                    if os.path.isdir(full_path):
                        rel_name = full_path if return_absolute_path else item
                        ret_list.append((rel_name, 0, "d"))
                    elif os.path.isfile(full_path):
                        size = os.path.getsize(full_path)
                        rel_name = full_path if return_absolute_path else item
                        ret_list.append((rel_name, size, "f"))
        except Exception:
            self.__logger.error("Can not list dir: [%s]" % dir_path + traceback.format_exc())
        finally:
            return ret_list
    def seek_file(self, file_path: str, start_index: int, length=1024) -> tuple:
        content_bytes = b''
        next_index = -1
        if not os.path.exists(file_path):
            raise Exception("File[%s] not exists !!!" % file_path)
        file_size = os.path.getsize(file_path)
        if start_index >= file_size:
            return content_bytes, next_index
        try:
            with open(file_path, "rb") as fh:
                fh.seek(start_index)
                content_bytes = fh.read(length)
                if start_index + length < file_size:
                    next_index = start_index + length
        except Exception:
            self.__logger.error("Seek file exception !!! " + traceback.format_exc())
        finally:
            return content_bytes, next_index
if __name__ == '__main__':
    pass  # 仅作模块导入测试，实际运行时由服务端实例化调用

第2步：启动UDP服务端监听请求

服务端负责接收客户端发送的字节流，反序列化后解析 type 字段走向命令路由或下载路由。默认监听地址为 127.0.0.1，端口 8090，暴露的根目录配置为 D:\var。启动后服务进入死循环等待 recvfrom。

import os, pickle, sys, socket, logging, time, traceback
from file_handler import FileHandler
LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s"
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
_logger = logging.getLogger()
class FTPServerUDP:
    def __init__(self, host="", port=None, root_dir="", logger=None):
        self.__host = host or "127.0.0.1"
        self.__port = port or 8090
        self.__root_dir = root_dir or r"D:\var"
        self.__logger = logger or _logger
        self.__file_handler = FileHandler()
        self.__socket_obj = self._bind_socket()
        self.__msg_unsupported = "Unsupported message type"
        self.__msg_error = "Server internal error"
        self.__msg_not_exists = "Target path not exists"
        self.__msg_ok = "ok"
    def _bind_socket(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.bind((self.__host, self.__port))
        return sock
    def _check_recv(self, msg_bytes):
        try:
            d = pickle.loads(msg_bytes)
        except Exception:
            return {}
        if not isinstance(d, dict) or not {"type","body","pwd","uuid"}.issubset(set(d.keys())):
            return {}
        return d
    def _send_back(self, resp_dict, client_addr):
        try:
            bts = pickle.dumps(resp_dict)
            self.__socket_obj.sendto(bts, client_addr)
        except Exception:
            resp_dict["status"] = 0
            resp_dict["body"] = self.__msg_error
            self.__socket_obj.sendto(pickle.dumps(resp_dict), client_addr)
    def _handle_cmd(self, req, resp, addr):
        cmd = req.get("body","").lower().strip()
        resp["pwd"] = req.get("pwd",[])
        if not cmd:
            resp.update({"body":self.__msg_unsupported,"status":0}); self._send_back(resp,addr); return
        base_dir = self.__root_dir
        if req.get("pwd"): base_dir = os.path.join(self.__root_dir, *req["pwd"])
        if cmd in ("ls","ll"):
            resp["body"] = self.__file_handler.list_dir(base_dir)
            resp["status"] = 1
        elif cmd.startswith("cd "):
            target = cmd.split(" ")[-1]
            if target in ("..", "/"):
                resp["pwd"] = [] if target=="/" else req["pwd"][:-1]
                resp["status"] = 1
            elif os.path.isdir(os.path.join(base_dir, target)):
                resp["pwd"].append(target)
                resp["status"] = 1
            else:
                resp.update({"body":self.__msg_not_exists,"status":0})
        elif cmd == "pwd":
            resp["body"] = req["pwd"]
            resp["status"] = 1
        else:
            resp.update({"body":self.__msg_unsupported,"status":0})
        self._send_back(resp,addr)
    def _handle_download(self, req, resp, addr):
        resp["type"] = "download"
        try:
            fname = req["body"]["file_name"]
            pwd = req["pwd"]
            s_idx = req["body"]["start_index"]
            length = min(req["body"].get("length", 1024*10), 1024*10)
            abspath = os.path.join(self.__root_dir, *pwd, fname)
            chunk, nxt_idx = self.__file_handler.seek_file(abspath, s_idx, length)
            resp["body"] = {"file_name":fname,"pwd":pwd,"content_bytes":chunk,"next_index":nxt_idx}
            resp["status"] = 1
        except Exception:
            resp.update({"body":self.__msg_error,"status":0})
        self._send_back(resp,addr)
    def run(self):
        print(f"FTP UDP Server running at {self.__host}:{self.__port}")
        while True:
            try:
                msg_tuple = self.__socket_obj.recvfrom(1024*1024)
                recv_dict = self._check_recv(msg_tuple[0])
                if not recv_dict: continue
                resp = {"type":"cmd","pwd":[],"status":1,"uuid":recv_dict.get("uuid","")}
                if recv_dict.get("type")=="cmd": self._handle_cmd(recv_dict, resp, msg_tuple[1])
                elif recv_dict.get("type")=="download": self._handle_download(recv_dict, resp, msg_tuple[1])
            except Exception as e:
                print(f"Server Loop Error: {e}"); time.sleep(1)
if __name__ == '__main__':
    ftp_srv = FTPServerUDP(host="127.0.0.1", port=8090, root_dir=r"D:\var")
    ftp_srv.run()

第3步：启动客户端发起交互请求

客户端内置终端交互循环，通过 input() 捕获用户指令并拼接字典。由于UDP不可靠，客户端实现了UUID校验与重试机制，单次请求失败最多尝试3次。下载线程会锁定输入状态，完成后自动恢复。

import socket, sys, time, pickle, os, logging, traceback, uuid
LOG_FORMAT = "%(asctime)s - %(levelname)s [%(filename)s-%(funcName)s] Line: %(lineno)s] - %(message)s"
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
_logger = logging.getLogger()
class FTPClientUDP:
    def __init__(self, host="127.0.0.1", port=8090, download_path="D:\my_download", logger=None):
        self.__host = host; self.__port = port
        self.__dl_root = download_path
        self.__logger = logger or _logger
        os.makedirs(download_path, exist_ok=True)
        self.__sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.__is_downloading = False
        self.__pwd = []
        self.__server_addr = (host, port)
    def _send_retry(self, cmd_dict, times=3):
        bts = pickle.dumps(cmd_dict)
        for _ in range(times):
            self.__sock.sendto(bts, self.__server_addr)
            try:
                raw, _ = self.__sock.recvfrom(1024*1024)
                resp = pickle.loads(raw)
                if cmd_dict.get("uuid") == resp.get("uuid"): return resp
                time.sleep(0.1)
            except Exception: break
        return {}
    def _exec_dl_single(self, fname, fsize, pwd):
        dl_req = {"type":"download","body":{"file_name":fname,"pwd":pwd,"start_index":0,"length":1024*1024},"pwd":pwd,"status":1,"uuid":uuid.uuid1().hex}
        abs_path = os.path.join(self.__dl_root, *pwd, fname)
        os.makedirs(os.path.dirname(abs_path), exist_ok=True)
        try:
            with open(abs_path, "wb") as fh:
                while True:
                    resp = self._send_retry(dl_req)
                    if not resp.get("status"): break
                    chunk = resp["body"]["content_bytes"]
                    fh.write(chunk)
                    nxt = resp["body"].get("next_index", -1)
                    if nxt == -1 or nxt == len(chunk): break
                    dl_req["body"]["start_index"] = resp["body"]["next_index"]
                    dl_req["uuid"] = uuid.uuid1().hex
                    pct = (resp["body"]["next_index"] / fsize * 100) if fsize>0 else 0
                    sys.stdout.write(f"\r[{fname}] -> {pct:.1f}%"); sys.stdout.flush()
                print(f"\nDownload completed: {abs_path}")
        except Exception as e: self.__logger.error(f"DL Fail: {e}")
    def run(self):
        while True:
            try:
                if not self.__is_downloading:
                    cmd = input("Input: ").strip()
                    msg = {"type":"cmd","body":cmd,"pwd":list(self.__pwd),"status":1,"uuid":uuid.uuid1().hex}
                    resp = self._send_retry(msg)
                    body = resp.get("body")
                    if isinstance(body, list):
                        for name, size, typ in body: print(f"{name:<20} {size:>10} {'Dir' if typ=='d' else 'File'}")
                    else: print(body)
                    if resp.get("status"): self.__pwd = resp.get("pwd", [])
            except KeyboardInterrupt: print("\nExit."); break
            except Exception as e: self.__logger.error(e); time.sleep(1)
if __name__ == '__main__':
    cli = FTPClientUDP(host="127.0.0.1", port=8090, download_path="D:\my_download")
    cli.run()

常见运行问题排查

在实际部署过程中，若遇到通信中断或下载异常，可按以下方向核查：

1. 数据包丢失导致卡死：UDP属于不可靠传输，若网络波动严重可能导致心跳包丢失。客户端内置了UUID比对与最大3次重试逻辑，若连续失败请检查本机防火墙是否放行了 8090 端口的入站/出站规则。
2. Pickle版本兼容性：序列化依赖Python原生协议，若服务端与客户端运行在完全不同的Python大版本上，可能出现反序列化报错。建议两端保持相同的解释器版本。
3. 路径越界防护：服务端代码中已通过startswith校验防止客户端通过 cd .. 逃逸至授权根目录之外。若需扩大权限范围，需调整安全拦截阈值或改用虚拟文件系统挂载。

总结

本方案完整展示了如何利用Python标准库构建基于UDP的轻量级文件分发服务。通过标准化的字典封包与分段读取机制，能够在不依赖第三方库的前提下实现基础的目录导航与断点续传逻辑。以上就是Python使用Socket的UDP协议搭建简易FTP文件服务器的详细内容，更多关于网络文件传输与Python Socket实战的资料请关注本站其它相关文章！