Python 文件下载优化:提高效率,减少冗余代码
def click_download_file(self, urn, path):
'点击下载按钮,下载文件'
downloaded_files_success = list() # 下载成功的文件列表
downloaded_files_failed = list() # 下载失败的文件列表
links = self.browser.safe_get_link(By.CSS_SELECTOR, ATTACHMENT_VIEW_WINDOW_ELEMENT)
for link in links:
# 获取文件名 文件大小
file_title, download_time_min, download_time_max = self.file_title_data(link, urn)
file_size = int(file_title.split('(')[1].split(')')[0].strip()) # 文件大小
download_time = min(download_time_min, 60) # 下载时间,最小值和60的较小值
status = self.download_file(path, file_title, download_time, file_size)
if status:
downloaded_files_success.append(file_title) # 下载成功
else:
for i in range(3): # 下载失败 重试下载
status = self.download_file(path, file_title, download_time, file_size)
if status:
downloaded_files_success.append(file_title) # 重试成功
logger.info('
>> 【urn = {}】文件{}重试下载成功'.format(urn, file_title))
break
else:
logger.info('
>> 【urn = {}】文件{}重试下载失败'.format(urn, file_title))
downloaded_files_failed.append(file_title) # 下载失败
if downloaded_files_failed:
logger.info('
>> 【urn = {}】下载失败文件列表:{}'.format(urn, downloaded_files_failed))
raise FileException(message=ErrorMsg.NO_DOWNLOADABLE_FILES, file_list=downloaded_files_failed)
if downloaded_files_success:
logger.info('
>> 【urn = {}】下载成功文件列表:{}'.format(urn, downloaded_files_success))
new_file_title = self.modify_file(path, urn, downloaded_files_success) # 下载成功修改文件名
return new_file_title
def download_file(cls, path, file_name, download_time, file_size):
'文件是否下载成功'
file_path = os.path.join(path, file_name)
file_path_with_suffix = file_path + '.crdownload'
last_size = 0
retries = 0
while retries < 3:
if os.path.exists(file_path): # 文件已下载成功
return True
if os.path.exists(file_path_with_suffix): # 文件正在下载中
cur_size = os.path.getsize(file_path_with_suffix)
speed = cur_size - last_size
last_size = cur_size
if speed > 0:
time.sleep(download_time)
else:
return False
else: # 文件未开始下载,点击下载按钮
self.browser.click_with_link(link)
self.download_alert() # 文件下载过大
time.sleep(download_time)
retries += 1
return False
def file_title_data(self, link, urn):
'获取文件信息:文件名,大小'
title = link.get_attribute('title')
file_titles = re.findall(REG, title)
if not file_titles:
file_titles = re.findall(CHINE_REG, title)
if len(file_titles) == 0:
QueryResult.update(UPDATE_INVOICES_SQL, (generate_utils.get_strtime(), urn))
file_title = file_titles[0].strip() # 文件名
# [Original File] 230324043.eml (14159) kB
file_size_str = title.split('(')[1].split(')')[0].strip()
file_size = int(file_size_str) # 文件大小
download_time_min = self.calculate_download_time_limit(file_size) # 下载时间
download_time = min(download_time_min, 60)
return file_title, download_time_min, download_time
优化内容:
- 减少冗余代码: 使用
while循环代替重复的self.browser.click_with_link(link)、self.download_alert()和time.sleep(download_time_limit)代码段,避免重复代码。 - 优化下载逻辑: 将下载逻辑封装到
download_file函数中,方便管理。 - 提高效率: 根据文件大小和下载时间限制,优化下载策略,提高下载效率。
- 重试机制: 使用
for循环实现下载失败的重试机制,最多重试 3 次。 - 日志记录: 使用
logger.info记录下载成功和失败的日志信息,方便调试和排查问题。
代码解释:
-
click_download_file函数:- 获取所有下载链接。
- 遍历每个链接,获取文件名、下载时间和文件大小。
- 调用
download_file函数下载文件,并记录下载结果。 - 处理下载失败的异常情况,并输出日志信息。
-
download_file函数:- 使用
while循环判断文件是否下载成功,最多重试 3 次。 - 判断文件状态,分别处理正在下载中、未开始下载和下载成功的场景。
- 使用
time.sleep控制下载间隔时间,避免频繁检查文件状态。
- 使用
-
file_title_data函数:- 从链接标题中提取文件名、文件大小和下载时间限制。
- 计算下载时间,并返回文件名、最小下载时间和最大下载时间。
总结:
通过优化代码,可以提高文件下载效率,减少代码冗余,增强代码可读性和维护性。
注意:
- 实际使用中,需要根据具体情况调整下载逻辑和重试机制。
- 确保
download_time参数值合理,避免下载过程超时。 - 使用
logger记录日志信息,方便调试和排查问题。
原文地址: https://www.cveoy.top/t/topic/qhJT 著作权归作者所有。请勿转载和采集!