如何通过Python的ftplib读取FTP服务器上CSV文件的最后一行？

要避免完整下载CSV文件，核心思路是利用FTP的REST命令从文件末尾分段读取数据，再从读取的片段中提取最后一行，无需下载整个文件。

方法一：基于文件大小的高效读取（推荐） #

大部分主流FTP服务器支持SIZE命令获取文件大小，结合REST命令定位到文件末尾附近，读取一段足够覆盖最后一行的字节即可。

代码实现 #

from ftplib import FTP

def get_last_ftp_csv_line(host, username, password, csv_path, chunk_size=1024):
    # 连接并登录FTP服务器
    ftp = FTP(host)
    ftp.login(username, password)

    # 获取目标CSV文件大小
    try:
        file_size = ftp.size(csv_path)
    except Exception as e:
        print(f"无法获取文件大小: {e}")
        ftp.quit()
        return ""

    if file_size == 0:
        ftp.quit()
        return ""

    # 计算起始读取位置，确保不小于文件开头
    start_pos = max(0, file_size - chunk_size)
    # 发送REST命令，设置读取起始点
    ftp.sendcmd(f"REST {start_pos}")

    # 读取从起始点到文件末尾的内容
    raw_data = b""
    def collect_chunk(chunk):
        nonlocal raw_data
        raw_data += chunk
    ftp.retrbinary(f"RETR {csv_path}", collect_chunk)

    ftp.quit()

    # 转换为字符串并处理换行（兼容Windows/Linux/macOS换行格式）
    text_content = raw_data.decode("utf-8", errors="ignore")
    lines = text_content.splitlines()

    # 提取最后一个非空行（可根据需求调整是否保留空行）
    last_valid_line = ""
    for line in reversed(lines):
        if line.strip():
            last_valid_line = line
            break
    return last_valid_line

# 使用示例
if __name__ == "__main__":
    last_line = get_last_ftp_csv_line(
        host="your_ftp_host",
        username="your_ftp_user",
        password="your_ftp_pass",
        csv_path="/data/records.csv"
    )
    print("CSV最后一行内容:", last_line)

关键说明 #

• chunk_size：可根据CSV单行的最大长度调整，默认1KB足够覆盖大多数场景，若单行过长可设为4KB或更大。
• 编码设置：需与目标CSV文件的编码一致（如gbk、utf-8-sig），避免乱码。
• 空行处理：代码默认跳过末尾的空行，若需保留最后一行（包括空行），可删除反转遍历的过滤逻辑，直接取lines[-1]。

方法二：兼容不支持SIZE命令的服务器 #

若FTP服务器不支持SIZE命令，可通过逐步向前分段读取的方式，直到找到换行符或文件开头。

代码实现 #

from ftplib import FTP

def get_last_ftp_csv_line_fallback(host, username, password, csv_path, chunk_size=1024):
    ftp = FTP(host)
    ftp.login(username, password)

    collected_data = b""
    current_start_pos = 0

    # 尝试获取文件大小，失败则从末尾逐步往前读
    try:
        file_size = ftp.size(csv_path)
        current_start_pos = max(0, file_size - chunk_size)
    except:
        # 服务器不支持SIZE命令，从较大的起始位置开始尝试
        current_start_pos = 1024 * 10  # 先从10KB位置开始

    while True:
        ftp.sendcmd(f"REST {current_start_pos}")
        temp_data = b""
        def append_chunk(chunk):
            nonlocal temp_data
            temp_data += chunk
        ftp.retrbinary(f"RETR {csv_path}", append_chunk)

        collected_data = temp_data + collected_data
        text_content = collected_data.decode("utf-8", errors="ignore")
        lines = text_content.splitlines()

        # 找到多行或已读到文件开头，停止读取
        if len(lines) > 1 or current_start_pos == 0:
            break

        # 继续向前读取一块数据
        current_start_pos = max(0, current_start_pos - chunk_size)

    ftp.quit()

    # 提取最后有效行
    last_valid_line = ""
    for line in reversed(lines):
        if line.strip():
            last_valid_line = line
            break
    return last_valid_line

注意事项 #

• 该方法兼容性更强，但效率略低于方法一，仅在服务器不支持SIZE命令时使用。

内容的提问来源于stack exchange，提问作者Janzaib M Baloch