You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python Docker环境下,如何优化Selenium的文件下载完成判断逻辑?

优化Selenium附件下载的文件检测逻辑

你的核心问题是误用了os.scandir()——它返回的是DirEntry对象的迭代器,而非列表,每次调用len(os.scandir(path))都会重新扫描目录并将迭代器转为列表,效率低下且逻辑不严谨。另外,仅通过文件数量判断下载完成,可能会因为同名文件覆盖、浏览器临时文件(如.crdownload、.part)导致误判。以下是几种更可靠的优化方案:

方案1:用文件名集合替代数量统计

通过记录下载前后的文件名集合,判断是否有新文件生成,同时过滤浏览器临时文件:

import os
import time
from selenium.common.exceptions import (
    NoSuchElementException,
    StaleElementReferenceException,
    ElementClickInterceptedException
)
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By

DOWNLOAD_DIR = "/tmp/downloads"

def get_attachments(driver):
    try:
        # 直接用WebDriverWait等待元素可点击,无需提前find_element
        attachments_list_element = WebDriverWait(driver, 10).until(
            EC.element_to_be_clickable((By.ID, "display-attachments-list"))
        )
        attachments_list_items = attachments_list_element.find_elements(By.TAG_NAME, "li")
        
        if not attachments_list_items:
            logger.info("No attachments found")
            scroll_to_top(driver)
            return
            
        # 创建目录,用0o777明确八进制权限(511是十进制的0777,可读性差)
        os.makedirs(DOWNLOAD_DIR, mode=0o777, exist_ok=True)
        
        for atch in attachments_list_items:
            # 记录下载前的所有文件名(过滤临时文件)
            def get_valid_files():
                return {f for f in os.listdir(DOWNLOAD_DIR) 
                        if not f.endswith(('.crdownload', '.part', '.tmp'))}
            
            before_files = get_valid_files()
            scroll_to_element(driver, atch)
            atch.click()
            logger.info(f"Downloading attachment: {atch.text}")
            
            # 等待新文件出现,最多等10秒
            for _ in range(10):
                current_files = get_valid_files()
                new_files = current_files - before_files
                if new_files:
                    logger.info(f"Download completed: {new_files.pop()}")
                    break
                time.sleep(1)
            else:
                logger.warning(f"Attachment {atch.text} download timed out")
                
    except (NoSuchElementException, StaleElementReferenceException, ElementClickInterceptedException):
        logger.info("Post does not include attachments")
    finally:
        scroll_to_top(driver)

方案2:直接监控特定文件(如果能从页面获取文件名)

如果附件的文件名可以从页面元素中提取(比如atch.text就是文件名),可以直接等待该文件出现在下载目录:

# 在循环内修改:
attachment_name = atch.text.strip()
before_files = get_valid_files()
scroll_to_element(driver, atch)
atch.click()
logger.info(f"Downloading attachment: {attachment_name}")

for _ in range(10):
    if attachment_name in get_valid_files():
        logger.info(f"Download completed: {attachment_name}")
        break
    time.sleep(1)

额外优化点

  • 避免重复扫描目录:将目录扫描逻辑封装成函数,减少重复代码
  • 权限可读性:将511改为0o777,明确表示八进制权限
  • 简化元素等待:去掉冗余的driver.find_element,直接用WebDriverWait完成等待和元素定位
  • 超时处理:添加超时警告,方便排查下载失败的情况

内容的提问来源于stack exchange,提问作者Hunter

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.22 12:37:25