文章目录
- 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)
目录
- 背景介绍
- 工具功能
- 核心代码解析
- 初始化配置
- 监控启动方法
- 文件检测方法
- 文件移动方法
- 使用示例
- 总结
在日常开发和自动化任务中,经常需要处理文件下载的场景。无论是网页自动化测试、爬虫抓取数据,还是自动备份文件,都需要一个可靠的方式来监控下载目录,并在文件下载完成后进行后续处理。本文介绍一个实用的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:UsersuserDownloads", # 监控的下载目录
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:projectdatareport.xlsx")
log.info("文件下载并移动成功!")
except Exception as e:
log.error(f"下载失败: {str(e)}")
本文介绍的下载监控工具提供了一个简单而强大的解决方案,用于自动化处理文件下载任务。通过合理设置超时时间、轮询频率和临时文件过滤规则,可以适应各种下载场景。工具的核心思路是通过比较目录快照的变化来检测新文件,然后进行后续处理。
这个工具类具有以下优势:
轻量级:不依赖外部库,标准库实现
可配置:灵活的参数设置适应不同需求
健壮性:完善的错误处理和超时机制
易集成:可以轻松集成到现有自动化流程中
到此这篇关于Python实现下载监控工具类:自动检测并移动下载文件的文章就介绍到这了,更多相关Python下载监控内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客!
您可能感兴趣的文章:
- 利用python编写自动整理下载目录工具
- python如何下载网络文件到本地指定文件夹
- Python实现文件下载的方法汇总与适用场景介绍
- Python下载文件的10种方法介绍
- Python使用psutil实现系统监控与资源管理
- python使用watchdog实现文件资源监控