Python實現(xiàn)下載監(jiān)控工具類:自動檢測并移動下載文件

更新時間：2025年06月16日 09:14:36 作者：嘆一曲當時只道是尋常

在日常開發(fā)和自動化任務中,經(jīng)常需要處理文件下載的場景,本文介紹一個實用的Python下載監(jiān)控工具類,能夠自動檢測新下載的文件并將其移動到指定位置,感興趣的小伙伴可以了解一下

背景介紹

在日常開發(fā)和自動化任務中，經(jīng)常需要處理文件下載的場景。無論是網(wǎng)頁自動化測試、爬蟲抓取數(shù)據(jù)，還是自動備份文件，都需要一個可靠的方式來監(jiān)控下載目錄，并在文件下載完成后進行后續(xù)處理。本文介紹一個實用的Python下載監(jiān)控工具類，能夠自動檢測新下載的文件并將其移動到指定位置。

工具功能

這個Downloader類提供以下核心功能：

監(jiān)控下載目錄：實時檢測指定目錄中新出現(xiàn)的文件
忽略臨時文件：自動過濾瀏覽器下載過程中的臨時文件（如.crdownload, .tmp）
文件移動與備份：將下載完成的文件移動到目標位置，支持備份或覆蓋現(xiàn)有文件
超時控制：設置合理的等待時間，避免無限期阻塞

核心代碼解析

初始化配置

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"]

監(jiān)控啟動方法

    def watch_start(self):
        """記錄下載前的文件狀態(tài)"""
        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("暫不支持監(jiān)控多個文件下載")
                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()  # 刪除現(xiàn)有文件
            else:
                # 創(chuàng)建帶時間戳的備份文件
                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)
    
    # 初始化下載監(jiān)控器
    downloader = Downloader(
        download_dir=r"C:\Users\user\Downloads",  # 監(jiān)控的下載目錄
        time_out=120,           # 超時時間2分鐘
        poll_frequency=0.2,     # 每0.2秒檢查一次
        exist_del=False         # 目標文件存在時創(chuàng)建備份
    )
    
    # 開始監(jiān)控
    downloader.watch_start()
    
    # 這里應該觸發(fā)下載操作（例如：點擊下載鏈接）
    # 示例：使用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)}")