python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python下载监控

Python实现下载监控工具类:自动检测并移动下载文件

作者:叹一曲当时只道是寻常

在日常开发和自动化任务中,经常需要处理文件下载的场景,本文介绍一个实用的Python下载监控工具类,能够自动检测新下载的文件并将其移动到指定位置,感兴趣的小伙伴可以了解一下

背景介绍

在日常开发和自动化任务中,经常需要处理文件下载的场景。无论是网页自动化测试、爬虫抓取数据,还是自动备份文件,都需要一个可靠的方式来监控下载目录,并在文件下载完成后进行后续处理。本文介绍一个实用的Python下载监控工具类,能够自动检测新下载的文件并将其移动到指定位置。

工具功能

这个Downloader类提供以下核心功能:

核心代码解析

初始化配置

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_dir
        self.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 = path
        self.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.suffix
                timestamp = 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)}")

总结

本文介绍的下载监控工具提供了一个简单而强大的解决方案,用于自动化处理文件下载任务。通过合理设置超时时间、轮询频率和临时文件过滤规则,可以适应各种下载场景。工具的核心思路是通过比较目录快照的变化来检测新文件,然后进行后续处理。

这个工具类具有以下优势:

轻量级:不依赖外部库,标准库实现

可配置:灵活的参数设置适应不同需求

健壮性:完善的错误处理和超时机制

易集成:可以轻松集成到现有自动化流程中

到此这篇关于Python实现下载监控工具类:自动检测并移动下载文件的文章就介绍到这了,更多相关Python下载监控内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文