在selenium中点击按钮下载判断文件是否下载成功可根据文件名或者文件大小判断且如果多个文件下载其中一个文件失败需要重试重试一定次数后记录失败的文件且文件如果全部下载完成对每个文件修改名字文件是否下载完成可以考虑文件名或者文件大小且修改文件名重复下载可能使用osrname考虑报错参考如下代码 修改def click_download_filesself urn el 文件点击下载
以下是修改后的代码:
import os
import time
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
def click_download_files(self, urn, el):
"""文件点击下载"""
file_names = list()
download_links = self.driver.find_elements(By.CSS_SELECTOR, el)
try:
for link in download_links:
file_title = link.get_attribute("title")
t_f_n = re.findall(REG, file_title)
if len(t_f_n) == 0:
QueryResult.update(UPDATE_INVOICES_SQL, (generate_utils.get_strtime(), urn))
continue
# 2023-3-27 测试中发现存在文件名为空的错误数据导致程序死循环
double_check = r'^\.'
tfn = re.findall(double_check, t_f_n[0].strip().strip())
if len(tfn) > 0:
QueryResult.update(UPDATE_INVOICES_SQL, (generate_utils.get_strtime(), urn))
continue
file_title = t_f_n[0].strip()
new_file_title = urn + "__" + t_f_n[0].strip() # 修改后的文件名
if link.is_enabled(): # 文件是否可点击下载
ActionChains(self.driver).move_to_element(link).click().perform()
# 文件下载时间
err_flag, seconds = self.download_wait(self.get_download_path(), file_title, 600)
if not err_flag:
logger.info("\n >> 【urn = {}】附件下载完成,文件名: 【{}】,用时: 【{}】".format(urn, file_title, seconds))
os.rename(self.get_download_path() + "\\" + file_title,
self.get_download_path() + "\\" + new_file_title)
file_names.append(new_file_title)
logger.info("\n >> 【urn = {}】防止冲突,文件改名: old = 【{}】, new = 【{}】".format(urn, file_title, new_file_title))
else:
# 下载超时
ValidateData().add_record(urn, self.name, Operate.DOWNLOAD.value, file_title + " " + ErrorMsg.DOWNLOAD_FILE_TIMEOUT, el)
# ValidateData().update_urn_is_load(urn)
QueryResult.update(UPDATE_INVOICES_1_SQL, (generate_utils.get_strtime(), urn))
else:
# 文件不可点击下载
ValidateData().add_record(urn, self.name, Operate.DOWNLOAD.value, file_title + " " + ErrorMsg.THE_DOWNLOAD_FILE_DOES_NOT_EXIST, el)
ValidateData().update_urn_is_load(urn)
except Exception as e:
# 处理异常,记录失败的文件
failed_files = [os.path.join(self.get_download_path(), file_name) for file_name in file_names]
retry_count = 0
while retry_count < MAX_RETRY:
retry_count += 1
for file_path in failed_files:
try:
os.remove(file_path)
except:
pass
else:
file_name = os.path.basename(file_path)
new_file_name = "retry_" + file_name
os.rename(file_path, os.path.join(self.get_download_path(), new_file_name))
file_names.append(new_file_name)
logger.info("\n >> 【urn = {}】重试下载,文件改名: old = 【{}】, new = 【{}】".format(urn, file_name, new_file_name))
failed_files.remove(file_path)
if not failed_files:
break
time.sleep(RETRY_INTERVAL)
if failed_files:
logger.error("\n >> 【urn = {}】文件下载失败,重试次数已达到上限".format(urn))
# 记录下载失败的文件
with open("failed_files.txt", "a") as f:
for file_path in failed_files:
f.write(file_path + "\n")
``
原文地址: https://www.cveoy.top/t/topic/h9m8 著作权归作者所有。请勿转载和采集!