Python实现下载监控工具:自动检测并移动下载文件
文章目录
- Python实现下载监控工具:自动检测并移动下载文件
- 背景介绍
- 工具功能
- 核心代码解析
- 初始化配置
- 监控启动方法
- 文件检测方法
- 文件移动方法
- 使用示例
- 总结
Python实现下载监控工具:自动检测并移动下载文件
背景介绍
在日常开发和自动化任务中,经常需要处理文件下载的场景。无论是网页自动化测试、爬虫抓取数据,还是自动备份文件,都需要一个可靠的方式来监控下载目录,并在文件下载完成后进行后续处理。本文介绍一个实用的Python下载监控工具类,能够自动检测新下载的文件并将其移动到指定位置。
工具功能
这个Downloader
类提供以下核心功能:
- 监控下载目录:实时检测指定目录中新出现的文件
- 忽略临时文件:自动过滤浏览器下载过程中的临时文件(如.crdownload, .tmp)
- 文件移动与备份:将下载完成的文件移动到目标位置,支持备份或覆盖现有文件
- 超时控制:设置合理的等待时间,避免无限期阻塞
核心代码解析
初始化配置
class Downloader():def __init__(self, download_dir, time_out=60, poll_frequency=0.5, exist_del=True, ignore_suffix_list=None):# 下载目录self.download_dir = download_dirself.time_out = time_out # 超时时间(秒)self.poll_frequency = poll_frequency # 轮询间隔(秒)self.exist_del = exist_del # 目标文件存在时是否删除# 忽略的临时文件后缀self.ignore_suffix_list = ignore_suffix_list or [".crdownload", ".tmp"]
监控启动方法
def watch_start(self):"""记录下载前的文件状态"""path = Path(self.download_dir)files = set(path.glob("*"))self.path = pathself.origin_files = files
文件检测方法
def get_download_file(self):"""检测新下载的文件"""start_time = perf_counter()while (perf_counter() - start_time <= self.time_out):now_files = set(self.path.glob("*"))# 计算新增文件diff_list = list(now_files.difference(self.origin_files))if not diff_list:sleep(self.poll_frequency)continue# 过滤临时文件final_diff_list = []for item in diff_list:if Path(item).suffix not in self.ignore_suffix_list:final_diff_list.append(item)# 处理有效文件if final_diff_list:if len(final_diff_list) > 1:raise Exception("暂不支持监控多个文件下载")return str(Path(final_diff_list[0]).resolve())return None
文件移动方法
def download_file(self, file_path):"""移动下载的文件到指定位置"""download_file = self.get_download_file()if download_file is None:raise Exception("下载超时,未获取到文件")log.info(f"找到文件 {download_file}, 开始移动")target = Path(file_path)# 处理目标文件已存在的情况if target.exists():if self.exist_del:target.unlink() # 删除现有文件else:# 创建带时间戳的备份文件suffix = target.suffixtimestamp = datetime.now().strftime("%Y%m%d%H%M%S")backup_path = target.with_name(f"{target.stem}_{timestamp}_bak{suffix}")shutil.move(file_path, backup_path)# 移动文件到目标位置shutil.move(download_file, file_path)
使用示例
if __name__ == "__main__":log.basicConfig(level=log.INFO)# 初始化下载监控器downloader = Downloader(download_dir=r"C:\Users\user\Downloads", # 监控的下载目录time_out=120, # 超时时间2分钟poll_frequency=0.2, # 每0.2秒检查一次exist_del=False # 目标文件存在时创建备份)# 开始监控downloader.watch_start()# 这里应该触发下载操作(例如:点击下载链接)# 示例:使用Selenium下载文件# driver.find_element(By.ID, "downloadButton").click()try:# 获取并移动下载的文件downloader.download_file(r"D:\project\data\report.xlsx")log.info("文件下载并移动成功!")except Exception as e:log.error(f"下载失败: {str(e)}")
总结
本文介绍的下载监控工具提供了一个简单而强大的解决方案,用于自动化处理文件下载任务。通过合理设置超时时间、轮询频率和临时文件过滤规则,可以适应各种下载场景。工具的核心思路是通过比较目录快照的变化来检测新文件,然后进行后续处理。
这个工具类具有以下优势:
- 轻量级:不依赖外部库,标准库实现
- 可配置:灵活的参数设置适应不同需求
- 健壮性:完善的错误处理和超时机制
- 易集成:可以轻松集成到现有自动化流程中
希望这个工具能为你的自动化任务带来便利!完整代码已在文章开头提供,欢迎直接使用和扩展。