Python统计FTP目录下各zip压缩包内CSV文件列数的方法咨询

• 全流程内存流处理：无需将FTP上的ZIP包下载到本地磁盘，也无需解压落地文件，1000个文件的处理耗时极短，无额外磁盘开销
• 遍历FTP指定目录下所有.zip后缀文件，逐个加载到内存后直接读取压缩包内的CSV对象
• 仅读取CSV首行内容，通过统计首行分隔符数量+1的方式计算列数，完全适配「字符串无双引号包裹、无需逐行校验」的场景，结果准确
• 最终输出「ZIP文件名-对应CSV列数」的映射表，可直接用于后续匹配对应版本的SSIS包

运行前提前确认两个配置：CSV实际使用的分隔符（常见为逗号、制表符\t、分号;）、CSV文件编码（常见为UTF-8、GBK），修改代码对应参数即可。

代码全部基于Python标准库实现，无需安装第三方依赖，修改开头配置段的参数即可使用。

from ftplib import FTP
import zipfile
import io
from collections import defaultdict

# -------------------------- 配置项，按需修改 --------------------------
FTP_HOST = "你的FTP服务器地址"
FTP_PORT = 21
FTP_USER = "FTP用户名"
FTP_PWD = "FTP密码"
FTP_TARGET_DIR = "/需要遍历的FTP目标目录路径"
CSV_DELIMITER = ","  # 制表符改"\t"，分号改";"
RESULT_SAVE_PATH = "csv列数统计结果.csv"  # 统计结果本地保存路径
# --------------------------------------------------------------------

def count_csv_columns():
    col_count_result = defaultdict(int)
    # 初始化FTP连接
    ftp = FTP()
    ftp.connect(FTP_HOST, FTP_PORT, timeout=30)
    ftp.login(FTP_USER, FTP_PWD)
    # 如果连接报被动模式相关错误，注释下一行代码即可切换主动模式
    ftp.set_pasv(True)
    ftp.cwd(FTP_TARGET_DIR)

    # 筛选目录下所有zip文件
    zip_file_list = [f for f in ftp.nlst() if f.lower().endswith(".zip")]
    print(f"共找到{len(zip_file_list)}个zip压缩包，开始统计列数...")

    for zip_name in zip_file_list:
        try:
            # 内存字节流接收zip文件，不落地本地
            zip_buffer = io.BytesIO()
            ftp.retrbinary(f"RETR {zip_name}", zip_buffer.write)
            zip_buffer.seek(0)

            # 读取内存中的zip包
            with zipfile.ZipFile(zip_buffer, "r") as zf:
                # 定位压缩包内的csv文件
                csv_name = [n for n in zf.namelist() if n.lower().endswith(".csv")][0]
                # 仅读取第一行计算列数
                with zf.open(csv_name, "r") as csv_f:
                    first_line = csv_f.readline().decode("utf-8").strip()
                    # 编码为GBK时把上面的utf-8替换为gbk即可
                    col_count = first_line.count(CSV_DELIMITER) + 1
                    col_count_result[zip_name] = col_count
                    print(f"文件{zip_name} 对应CSV列数：{col_count}")
        except Exception as e:
            print(f"处理{zip_name}时出错：{str(e)}")
            col_count_result[zip_name] = -1  # -1标记处理失败的文件
    ftp.quit()

    # 结果保存为本地csv方便后续调用
    with open(RESULT_SAVE_PATH, "w", encoding="utf-8") as f:
        f.write("zip文件名,对应CSV列数\n")
        for file_name, cnt in col_count_result.items():
            f.write(f"{file_name},{cnt}\n")
    print(f"统计完成，结果已保存到{RESULT_SAVE_PATH}")
    return col_count_result

if __name__ == "__main__":
    result = count_csv_columns()

• 若CSV为GBK/GB2312编码，将代码中decode("utf-8")替换为对应编码即可
• 列数返回-1的条目为处理失败的文件，常见原因为压缩包损坏、压缩包内无CSV文件、编码不匹配，可单独排查
• 若FTP目录下有子目录需要递归遍历，可在获取文件列表部分增加递归逻辑即可

内容的提问来源于stack exchange，提问作者B F