使用aioftp与asyncio实现FTP文件下载失败，寻求技术解决方案

我尝试用aioftp结合asyncio实现FTP文件下载，但未成功。目前改用aiohttp和asyncio编写了代码，希望能得到帮助，以下是我的代码：

import asyncio
import requests
import aiohttp
import urllib.request
import urllib.error
import time
import urllib.request

t1 = time.time()


async def job(session, url):
    filename = url.split("/")[-1]
    ftp = await session.get(url)
    ftpcode = await ftp.read()
    with open("new/" + str(filename), 'wb') as f:
        f.write(ftpcode)
    return str(url)

    # urllib.request.urlretrieve(url, filename)

async def main(loop, URL):
    async with aiohttp.ClientSession() as session:
        tasks = [loop.create_task(job(session, URL[x])) for x in range(2)]
   
        finished, unfinished = await asyncio.wait(tasks)
        all_results = [r.result() for r in finished]
        print("ALL RESULT:", all_results)


URL = [
    "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/8e/71/PMC5334499.tar.gz",
    "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/ad/b5/PMC7444942.tar.gz",
    "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/e8/38/PMC8903393.tar.gz",
    "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/f3/7d/PMC9130726.tar.gz"
]

loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop, URL))
loop.close()

t2 = time.time()
print(int(t2-t1))

当前代码无法正常工作的核心原因是：aiohttp的ClientSession并不支持FTP协议，它仅处理HTTP/HTTPS类请求。直接用session.get()请求FTP链接会导致请求失败，无法获取文件内容。

要异步下载FTP文件，应该使用专门的aioftp库，以下是完整可运行的代码：

import asyncio
import aioftp
import time
import os

t1 = time.time()

async def download_ftp_file(url, save_path):
    # 解析FTP链接的主机地址和远程文件路径
    url_parts = url.split("://")[1].split("/")
    host = url_parts[0]
    remote_path = "/" + "/".join(url_parts[1:])
    filename = remote_path.split("/")[-1]
    local_file_path = f"{save_path}/{filename}"

    # 建立匿名FTP连接（NCBI的FTP服务器支持匿名访问）
    async with aioftp.ClientSession(host) as client:
        # 异步下载文件到本地
        await client.download(remote_path, local_file_path, write_into=True)
    return url

async def main(url_list, save_path="new"):
    # 确保保存目录存在
    os.makedirs(save_path, exist_ok=True)
    # 创建所有下载任务
    tasks = [asyncio.create_task(download_ftp_file(url, save_path)) for url in url_list]
    # 等待所有任务完成
    finished, unfinished = await asyncio.wait(tasks)
    all_results = [r.result() for r in finished]
    print("已完成下载的链接:", all_results)

if __name__ == "__main__":
    URL = [
        "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/8e/71/PMC5334499.tar.gz",
        "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/ad/b5/PMC7444942.tar.gz",
        "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/e8/38/PMC8903393.tar.gz",
        "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_package/f3/7d/PMC9130726.tar.gz"
    ]
    
    asyncio.run(main(URL))
    
    t2 = time.time()
    print(f"总耗时: {int(t2-t1)} 秒")

• 解析FTP链接，分离出主机地址和远程文件路径，适配NCBI的FTP服务器结构
• 使用aioftp.ClientSession建立异步FTP连接，NCBI服务器支持匿名登录，无需账号密码
• 通过client.download()异步下载文件，write_into=True直接写入本地文件，避免内存占用过高
• 自动创建保存目录，避免因目录不存在导致报错
• 使用asyncio.run()替代旧的事件循环调用方式，符合Python 3.7+的异步代码规范

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