Python Selenium下载APK时等待完成及断网异常处理求助

核心问题分析 #

1. 线程创建错误：Thread(target=task(url)) 会直接同步执行task(url)，线程并未真正异步启动，导致后续等待逻辑完全失效。
2. 单一文件检测逻辑：仅判断.apk文件是否存在，断网时Chrome生成的.crdownload临时文件不会转为.apk，直接导致循环无法退出。
3. 无超时机制：遇到网络异常时没有终止条件，程序陷入无限循环。

解决方案：两种可靠的等待实现方式 #

方式一：改进文件监控（带超时与临时文件检测）

通过监控下载目录的临时文件状态，结合超时判断，避免无限循环：

• 检测.crdownload文件的大小变化，若长时间无更新则判定下载停滞
• 设置合理的超时阈值（可根据文件大小调整）
• 修复线程创建的基础错误

修改后的代码：

import glob
import os
import time
from threading import Thread
from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def waiting_download(download_dir, timeout=300):
    start_time = time.time()
    last_file_size = 0
    last_check_time = time.time()
    crdownload_path = os.path.join(download_dir, "*.crdownload")
    apk_path = os.path.join(download_dir, "*.apk")
    
    while time.time() - start_time < timeout:
        # 检查是否生成完整APK
        apk_files = glob.glob(apk_path)
        if len(apk_files) > 0:
            return True
        
        # 监控临时下载文件状态
        crdownload_files = glob.glob(crdownload_path)
        if len(crdownload_files) > 0:
            current_size = os.path.getsize(crdownload_files[0])
            # 10秒内文件无变化，判定下载停滞
            if current_size == last_file_size:
                if time.time() - last_check_time > 10:
                    print("下载停滞，疑似网络异常")
                    return False
            else:
                last_file_size = current_size
                last_check_time = time.time()
        time.sleep(1)
    
    print("下载超时")
    return False

# 初始化WebDriver
chromedriver = 'C:\\Users\\ASUS\\anaconda3\\phd_implement\\chromedriver.exe'
os.environ["webdriver.chrome.driver"] = chromedriver
chrome_options = Options()
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_argument("--disable-popup-blocking")
download_dir = "C:\\Users\\ASUS\\anaconda3\\phd_implement\\download_apk\\tmp\\"
prefs = {
    'profile.default_content_setting_values.automatic_downloads': 1,
    "download.default_directory": download_dir
}
chrome_options.add_experimental_option("prefs", prefs)
web = webdriver.Chrome(options=chrome_options)
web.maximize_window()

# 下载链接
url = "https://download.apkcombo.com/com.avast.android.secure.browser/Avast%20Secure%20Browser_7.5.2_apkcombo.com.apk?ecp=Y29tLmF2YXN0LmFuZHJvaWQuc2VjdXJlLmJyb3dzZXIvNy41LjIvNDYzMC5jZWZhYTAzZTYwOTY3NGY0Y2Y2YWRiMTk1N2VkMzhlODMzYTExNGQzLmFwaw==&iat=1681553964&sig=4cc693ff4b5dfe4adea4030b65b8e454&size=198815248&from=cf&version=latest&lang=en&fp=1dce2e74bf37995f5efd03cd5877afb5&ip=130.25.102.251"

def task(url):
    web.get(url)

# 修复线程创建：传入函数对象与参数
thread = Thread(target=task, args=(url,))
thread.start()

print('Waiting for the download file...')
if waiting_download(download_dir):
    print('Download completed')
else:
    print('Download failed or timed out')

thread.join()
web.close()

方式二：利用Chrome DevTools监控下载状态（更精准）

直接通过Chrome的DevTools API获取下载实时状态，能准确判断下载完成、失败或停滞：

import os
import time
from threading import Thread
from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def wait_for_download(driver, timeout=300):
    start_time = time.time()
    # 启用DevTools的下载监控
    driver.execute_cdp_cmd("Page.enable", {})
    driver.execute_cdp_cmd("Downloads.enable", {})
    
    download_completed = False
    download_failed = False
    
    # 监听下载状态更新事件
    def on_download_event(event):
        nonlocal download_completed, download_failed
        if event["params"]["state"] == "completed":
            download_completed = True
        elif event["params"]["state"] == "interrupted":
            download_failed = True
    
    driver.add_event_listener("Downloads.downloadUpdated", on_download_event)
    
    while time.time() - start_time < timeout:
        if download_completed:
            return True
        if download_failed:
            print("下载被中断")
            return False
        time.sleep(1)
    
    print("下载超时")
    return False

# 初始化WebDriver
chromedriver = 'C:\\Users\\ASUS\\anaconda3\\phd_implement\\chromedriver.exe'
os.environ["webdriver.chrome.driver"] = chromedriver
chrome_options = Options()
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_argument("--disable-popup-blocking")
download_dir = "C:\\Users\\ASUS\\anaconda3\\phd_implement\\download_apk\\tmp\\"
prefs = {
    'profile.default_content_setting_values.automatic_downloads': 1,
    "download.default_directory": download_dir
}
chrome_options.add_experimental_option("prefs", prefs)
web = webdriver.Chrome(options=chrome_options)
web.maximize_window()

# 下载链接
url = "https://download.apkcombo.com/com.avast.android.secure.browser/Avast%20Secure%20Browser_7.5.2_apkcombo.com.apk?ecp=Y29tLmF2YXN0LmFuZHJvaWQuc2VjdXJlLmJyb3dzZXIvNy41LjIvNDYzMC5jZWZhYTAzZTYwOTY3NGY0Y2Y2YWRiMTk1N2VkMzhlODMzYTExNGQzLmFwaw==&iat=1681553964&sig=4cc693ff4b5dfe4adea4030b65b8e454&size=198815248&from=cf&version=latest&lang=en&fp=1dce2e74bf37995f5efd03cd5877afb5&ip=130.25.102.251"

def task(url):
    web.get(url)

thread = Thread(target=task, args=(url,))
thread.start()

print('Waiting for the download file...')
if wait_for_download(web):
    print('Download completed')
else:
    print('Download failed or timed out')

thread.join()
web.close()

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