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_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)}")