Python Playwright实现JS页面指定文件下载及问题排查

代码下载逻辑失效有3个核心问题：

• 事件监听注册时机错误：循环内点击按钮后才注册download事件监听，会直接漏掉点击瞬间触发的下载事件；重复注册监听还会导致单次下载触发多次冗余处理。
• 异步操作未等待：download.save_as()是异步IO方法，写在lambda中未加await等待磁盘写入完成，点击后直接进入下一轮循环、最终关闭浏览器时，未完成的写入操作会被强制中断，目标目录自然不会生成文件。
• 事件与触发动作未绑定：全局page.on("download")监听在多下载场景下，无法精准匹配单次点击对应的下载事件，容易出现文件错配、漏接问题。

逐个点击下载的场景下，用Playwright自带的expect_download上下文管理器稳定性最高：它会提前启动下载监听，和单次点击动作严格绑定，自动等待下载对象返回，只需显式等待文件保存完成再进入下一轮操作即可。
修正后的完整代码：

import asyncio
from playwright.async_api import async_playwright
from pathlib import Path

path = Path().home() / 'Downloads'
path.mkdir(exist_ok=True) # 提前创建目标目录，避免路径不存在报错
timeout = 300000                   # 5分钟超时

async def main():
    async with async_playwright() as p:
        # 调试阶段可加headless=False参数打开可见浏览器窗口，观察运行状态
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(accept_downloads=True) # 显式开启下载权限
        page = await context.new_page()
        await page.goto("https://my-fun-page.com", timeout=timeout)
        await page.wait_for_selector('ul.ant-list-items', timeout=timeout)
        downloads = page.locator("button", has=page.locator("span", has_text="_Insurer_"))
        count = await downloads.count()
        print(f"共找到{count}个待下载文件")

        for i in range(count):
            print(f"开始下载第{i+1}/{count}个文件")
            # 先启动下载监听，再触发点击，确保不会漏掉事件
            async with page.expect_download(timeout=timeout) as download_info:
                await downloads.nth(i).click(timeout=0)
            # 获取下载对象
            download = await download_info.value
            # 自定义保存路径，此处加序号前缀避免重名文件被覆盖，可按需修改命名规则
            save_file_path = path / f"{i+1}_{download.suggested_filename}"
            # 等待文件完全写入磁盘
            await download.save_as(save_file_path)
            print(f"\t文件已保存至: {save_file_path}")
        
        await browser.close()
        print("所有下载任务完成")

asyncio.run(main())

如果站点没有严格的并发限制，可以用信号量控制并发数（建议不超过5）提升下载速度，核心逻辑是提前注册一次全局下载监听，用异步任务统一处理保存：

# 并发版本核心逻辑片段
sem = asyncio.Semaphore(5) # 最多同时处理5个下载任务
save_tasks = []

async def save_download(download):
    async with sem:
        idx = len(save_tasks) + 1
        save_path = path / f"{idx}_{download.suggested_filename}"
        await download.save_as(save_path)
        print(f"已保存第{idx}个文件: {save_path}")

# 所有点击操作前仅注册一次监听
page.on("download", lambda d: save_tasks.append(asyncio.create_task(save_download(d))))

# 批量触发所有点击
await downloads.click_all(timeout=0)
# 等待所有保存任务完成
await asyncio.gather(*save_tasks)

注意：如果需要按按钮对应文本自定义文件名，建议在点击前先通过all_text_contents()拿到所有按钮的文本列表，循环时按索引对应命名即可，不要在下载触发后再查询页面元素，避免页面动态变化导致文本匹配错误。

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