自动化下载文件:根据URN和CSS选择器下载附件
{ "def click_download_files(self, urn, el):\n """文件点击下载""" downloaded_files = list() # 下载成功的文件列表\n downloaded_failed = list() # 下载失败文件列表\n download_links = self.driver.find_elements(By.CSS_SELECTOR, el)\n try:\n for link in download_links:\n file_title = link.get_attribute("title")\n t_f_n = re.findall(REG, file_title)\n\n if len(t_f_n) == 0:\n QueryResult.update(UPDATE_INVOICES_SQL, (generate_utils.get_strtime(), urn))\n continue\n # 2023-3-27 测试中发现存在文件名为空的错误数据导致程序死循环\n double_check = r'^.'\n tfn = re.findall(double_check, t_f_n[0].strip().strip())\n if len(tfn) > 0:\n QueryResult.update(UPDATE_INVOICES_SQL, (generate_utils.get_strtime(), urn))\n continue\n file_title = t_f_n[0].strip()\n new_file_title = urn + "__" + t_f_n[0].strip() # 修改后的文件名\n\n if link.is_enabled(): # 文件是否可点击下载\n file_path = os.path.join(self.get_download_path(), new_file_title) # 完整的文件路径\n # 判断文件是否已经存在\n if os.path.exists(file_path) and os.path.getsize(file_path) > 0: # 检查文件是否存在且大小大于0\n # 删除原文件\n os.remove(file_path)\n\n link.click()\n # 等待确认对话框出现\n alert = Alert(self.driver)\n alert_text = alert.text\n print("alert_text=", alert_text)\n # 如果提示框文本包含所需的确认消息 文件下载内容过大\n if "This files has big size. It might take long time to download. Do you want to continue?" in alert_text:\n # 点击确定按钮\n alert.accept()\n\n # # 确认对话框\n # self.driver.switch_to.alert.accept()\n time.sleep(60)\n # 等待文件下载完成\n max_retry = 3\n retry_count = 0\n file_downloaded = False\n # 文件下载失败,重试\n while retry_count < max_retry and not file_downloaded:\n time.sleep(1) # 等待1秒\n # expected_size = 100 # 预期文件大小\n file_path = os.path.join(self.get_download_path(), file_title) # 完整的文件路径\n # 在判断文件已经下载完成后,等待额外的时间是为了确保文件已经完全下载完成。有时候,虽然文件已经下载完成,\n # 但是文件系统可能还在处理文件写入操作,因此需要等待一段时间以确保文件完全写入磁盘。\n if os.path.exists(file_path) and os.path.getsize(file_path) > 0: # 检查文件是否存在且大小大于0\n if not file_path.endswith('.crdownload'):\n # 等待额外的时间以确保文件完全下载完成\n time.sleep(5) # 等待5秒\n file_downloaded = True\n\n # # 判断文件大小是否符合预期\n # if os.path.getsize(file_path) > expected_size:\n # file_downloaded = True\n # time.sleep(5)\n # else:\n # # 删除下载失败的文件\n # os.remove(file_path)\n\n retry_count += 1\n\n if retry_count < max_retry and not file_downloaded:\n # 重试下载失败的文件\n time.sleep(10)\n link.click()\n\n if not file_downloaded:\n # 记录下载失败的文件名\n downloaded_failed.append(file_title)\n else:\n # 记录成功下载的文件名\n downloaded_files.append(new_file_title)\n\n # 修改文件名\n new_file_path = os.path.join(self.get_download_path(), new_file_title) # 修改后的文件路径\n shutil.move(file_path, new_file_path) # 使用shutil.move来重命名文件\n\n else:\n # 文件不可点击下载\n ValidateData().add_record(urn, self.name, Operate.DOWNLOAD.value,\n file_title + " " + ErrorMsg.THE_DOWNLOAD_FILE_DOES_NOT_EXIST, el)\n ValidateData().update_urn_is_load(urn)\n logger.info("\n >> 【urn = {}】文件下载列表: 成功文件【{}】,失败文件: 【{}】".format(urn, downloaded_files,\n downloaded_failed))\n return downloaded_files, downloaded_failed\n except Exception as e:\n log_error(e)\n content = "download_files 下载附件时出错!!"\n # ProcessMsg().execution_error(urn_no, e, content)\n ValidateData().add_record(urn, self.name, Operate.DOWNLOAD.value, content, str(traceback.format_exc()), )\n return downloaded_files, downloaded_failed\n\nclick_download_files(self, urn, ATTACHMENT_VIEW_WINDOW_ELEMENT)
原文地址: https://www.cveoy.top/t/topic/p7Iq 著作权归作者所有。请勿转载和采集!