一、自动化文件分发的两大痛点

在搭建持续集成打包分发、测试固件固化以及自动化升级链路时，很多企业依然在内网使用私有 FTP/SFTP 服务进行产物托管。

使用 Python 标准库 ftplib 进行自动化上传与下载时，开发者几乎都会被两个典型问题困扰：

1. 中文路径与文件名乱码：FTP 协议在 RFC 959 早期设计时默认只支持 7 位 ASCII 或 Latin-1 编码。当服务端的中文目录由 Windows GBK 产生、客户端在 Linux UTF-8 环境下解析时，会频繁抛出 550 No such file or directory 或产生不可读的乱码字符。
2. 大文件传输“黑盒无反馈”：直接调用 ftp.retrbinary() 下载几个 G 的测试数据包时，控制台没有任何进度提示，无法得知当前传输百分比与实时传输速率。

---

二、攻克中文路径乱码：编解码桥接技术

Python 3 的 ftplib 默认使用 latin-1 传输指令字符串。在与国内常见的 Windows FTP 服务器通信时，必须将中文路径显式进行 GBK 编码 -> Latin-1 解码 桥接处理：

def encode_ftp_path(raw_path: str) -> str:
    """
    解决包含中文字符的路径在标准 ftplib 握手时的编码失效问题
    将 UTF-8 字符串转换为 GBK 字节流，再以 latin-1 字符透传给底层 socket
    """
    if not isinstance(raw_path, str):
        return raw_path
    
    # 检查字符串中是否存在中文字符
    has_chinese = any('\u4e00' <= ch <= '\u9fff' for ch in raw_path)
    if has_chinese:
        return raw_path.encode('gbk').decode('latin-1')
    return raw_path

通过这一层透明转换，在执行 ftp.cwd(encode_ftp_path("应用软件/版本发布")) 时，服务端就能准确识别并切换至对应目录。

---

三、大文件分块回调与动态进度条实现

ftplib 的 retrbinary 和 storbinary 提供了 callback 回调函数机制。每当网络流读写一个分块（如 8KB）时，都会触发一次回调，可据此计算累计传输字节数与实时百分比：

import os
import sys
import time
from ftplib import FTP

class FtpTransferManager:
    def __init__(self, host="127.0.0.1", port=21, user="anonymous", password=""):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.ftp = None

    def connect(self):
        self.ftp = FTP()
        self.ftp.connect(self.host, self.port, timeout=30)
        self.ftp.login(self.user, self.password)
        # 强制开启二进制传输模式
        self.ftp.voidcmd('TYPE I')
        return self

    def download_with_progress(self, remote_file: str, local_save_path: str):
        """带进度条与速率反馈的二进制下载"""
        # 1. 查询远程文件总字节大小
        encoded_remote_name = encode_ftp_path(remote_file)
        total_size = self.ftp.size(encoded_remote_name)
        if not total_size or total_size <= 0:
            print(f"[Warning] 无法获取文件大小，降级为普通下载: {remote_file}")
            total_size = 1

        downloaded_size = 0
        start_time = time.time()

        # 2. 定义分块写入与控制台进度打印回调函数
        def chunk_callback(chunk: bytes):
            nonlocal downloaded_size
            f_out.write(chunk)
            downloaded_size += len(chunk)
            
            percent = (downloaded_size / total_size) * 100
            elapsed = time.time() - start_time
            speed_kb = (downloaded_size / 1024) / (elapsed if elapsed > 0 else 0.001)

            # 终端平滑刷新同一行
            sys.stdout.write(
                f"\r[下载中] {os.path.basename(local_save_path)}: "
                f"{percent:6.2f}% ({downloaded_size}/{total_size} 字节) | 速度: {speed_kb:.1f} KB/s"
            )
            sys.stdout.flush()

        # 3. 执行流式拉取
        with open(local_save_path, "wb") as f_out:
            # blocksize 设置为 64KB 提升 I/O 吞吐
            self.ftp.retrbinary(f"RETR {encoded_remote_name}", chunk_callback, blocksize=65536)
        
        sys.stdout.write("\n")
        print(f"✓ 文件成功下载至: {local_save_path}")

    def close(self):
        if self.ftp:
            try:
                self.ftp.quit()
            except Exception:
                self.ftp.close()

---

四、传输后双向完整性校验（校验和验证）

在工业级自动化发布流水线中，不能仅凭网络没有报错就认为文件传输成功。必须增加校验保障：

import hashlib

def verify_file_md5(local_file_path: str, expected_md5: str) -> bool:
    """分块计算本地大文件 MD5，防止整包载入内存"""
    hasher = hashlib.md5()
    with open(local_file_path, "rb") as f:
        while chunk := f.read(65536):
            hasher.update(chunk)
    actual_md5 = hasher.hexdigest()
    return actual_md5.lower() == expected_md5.lower()