如何将本地读取WAV文件的Python脚本修改为从FTP服务器读取

端口说明 #

• 21端口是传统FTP协议的默认端口，普通FTP服务直接使用21即可
• 22端口是SFTP（基于SSH的文件传输协议）的默认端口，需先确认你的服务器部署的服务类型，以下代码以标准FTP为例，若使用SFTP可替换为paramiko库实现对应逻辑

修改思路 #

原脚本为本地文件读写逻辑，修改后核心调整点：

1. 输入项新增FTP服务器地址、端口、账号、密码、远程WAV文件存储目录
2. 用FTP协议替代本地OS方法，列取远程目录下的WAV文件
3. 将远程文件内容读取到内存中，无需保存到本地磁盘即可解析属性
4. 通过FTP协议接口获取文件的创建、修改、访问时间

完整修改后代码 #

import csv
from datetime import timedelta
from time import ctime
from wavinfo import WavInfoReader
# 新增FTP和内存IO依赖
import ftplib
from io import BytesIO


# OUTPUT FILE HEADERS
EXTRACTED_HEADERS = ['copyright', 'product', 'album', 'genre', 'artist', 'created_date', 'title']
DURATION_HEADER = 'duration'
CREATED_TIME_HEADER = 'created_time'
MODIFIED_TIME_HEADER = 'modified_time'
ACCESS_TIME_HEADER = 'access_time'

# ERROR FILE HEADERS
FILE_NAME_HEADER = 'file_name'
ERROR_HEADER = 'error'


def get_input():
    ftp_host = input('FTP服务器地址: ').strip()
    while not ftp_host:
        ftp_host = input('FTP服务器地址必填（按Ctrl+C退出）: ').strip()
    ftp_port = input('FTP端口（默认21）: ').strip() or 21
    ftp_user = input('FTP用户名（匿名访问填anonymous）: ').strip() or 'anonymous'
    ftp_passwd = input('FTP密码（匿名访问可填任意邮箱）: ').strip() or 'test@example.com'
    remote_dir = input('FTP上WAV文件存储目录: ').strip()
    while not remote_dir:
        remote_dir = input('FTP上WAV文件存储目录必填（按Ctrl+C退出）: ').strip()
    o_file = input('输出文件路径（默认output.csv）:').strip() or 'output.csv'
    e_file = input('错误日志路径（默认error.csv）: ').strip() or 'error.csv'
    return ftp_host, int(ftp_port), ftp_user, ftp_passwd, remote_dir, o_file, e_file


def write_output_file(output_list, file):
    with open(file, 'w+', encoding='utf-8') as f:
        dict_writer = csv.DictWriter(f, fieldnames=[FILE_NAME_HEADER] + EXTRACTED_HEADERS + [DURATION_HEADER, CREATED_TIME_HEADER, MODIFIED_TIME_HEADER, ACCESS_TIME_HEADER])
        dict_writer.writeheader()
        for op in output_list:
            dict_writer.writerow(op)


def write_error_file(failure_list, file):
    with open(file, 'w+', encoding='utf-8') as f:
        dict_writer = csv.DictWriter(f, fieldnames=[FILE_NAME_HEADER, ERROR_HEADER])
        dict_writer.writeheader()
        for fl in failure_list:
            dict_writer.writerow(fl)


def run():
    ftp_host, ftp_port, ftp_user, ftp_passwd, remote_dir, output_file, error_file = get_input()
    output = []
    failure = []
    # 连接FTP服务器
    try:
        ftp = ftplib.FTP()
        ftp.connect(ftp_host, ftp_port, timeout=10)
        ftp.login(ftp_user, ftp_passwd)
        ftp.cwd(remote_dir)
        # 获取目录下所有文件
        only_files = ftp.nlst()
    except Exception as e:
        print(f"FTP连接失败: {str(e)}")
        return

    for file_name in only_files:
        # 只处理wav文件
        if not file_name.endswith('wav'):
            continue
        properties = {
            FILE_NAME_HEADER: file_name
        }
        try:
            # 将文件下载到内存
            bio = BytesIO()
            ftp.retrbinary(f'RETR {file_name}', bio.write)
            bio.seek(0)
            # 解析Wav信息
            w = WavInfoReader(bio, info_encoding='utf-8', bext_encoding='utf-8')
            info_dict = w.info.to_dict()
            for header in EXTRACTED_HEADERS:
                properties[header] = info_dict.get(header, '')

            # 计算时长
            properties[DURATION_HEADER] = str(timedelta(seconds=w.data.byte_count / w.fmt.byte_rate))

            # 获取文件时间信息
            mtime = ftp.sendcmd(f'MDTM {file_name}')[4:]
            # 转换时间格式
            from datetime import datetime
            mtime_dt = datetime.strptime(mtime, "%Y%m%d%H%M%S")
            mtime_stamp = ctime(mtime_dt.timestamp())
            # FTP默认只提供修改时间，其余时间可根据需要调整
            properties[CREATED_TIME_HEADER] = mtime_stamp
            properties[MODIFIED_TIME_HEADER] = mtime_stamp
            properties[ACCESS_TIME_HEADER] = mtime_stamp
            output.append(properties)
            bio.close()
        except Exception as e:
            failure.append({
                FILE_NAME_HEADER: file_name,
                ERROR_HEADER: str(e)
            })
    ftp.quit()
    write_output_file(output, output_file)
    write_error_file(failure, error_file)
    print(f"处理完成，成功{len(output)}个，失败{len(failure)}个")


if __name__ == '__main__':
    run()

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