我的个人电脑有一个很羞耻的入口——~/Downloads。这里堆积着三个月前下载的文件、老板发的合同、还有一堆后缀奇怪的设计素材。每次想找点东西,都得翻半天。终于,某个工作周二的下午,我怒写了一段Python程序,让它帮我盯着这个文件夹,只要有了新文件,就自动按类别扔进对应子目录。
当时只花了不到一小时,但效果立竿见影。我把这段脚本整理了一下,分享出来。它不算多高大上,但思路很适合写自动化脚本的新手参考。
先想清楚要按什么规则归类
最常见的归类方式就是按扩展名。图片放images,文档放docs,压缩包放archives,程序安装包放programs,其他的扔进others。我实际用的时候,还会额外加上一个“最近项目”目录,把工作相关的文件单独拎走,不过下面代码里我们先做最基础的版本。
定的映射关系大概是这样:
image_exts = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
doc_exts = {'.pdf', '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx', '.txt', '.md'}
zip_exts = {'.zip', '.rar', '.7z', '.tar', '.gz'}
program_exts = {'.exe', '.msi', '.dmg', '.deb', '.rpm'}
第一版:一个简单的轮询脚本
其实刚开始没想用监听,就写了个死循环,每隔三秒扫描一次目录,发现新文件就移动。代码特别朴实,直接用了os、shutil和time。
import os
import time
import shutil
from pathlib import Path
def get_category(ext):
if ext in image_exts:
return 'images'
elif ext in doc_exts:
return 'docs'
elif ext in zip_exts:
return 'archives'
elif ext in program_exts:
return 'programs'
else:
return 'others'
def organize_folder(folder):
folder = Path(folder)
if not folder.exists():
print('文件夹不存在')
return
for item in folder.iterdir():
if item.is_file():
ext = item.suffix.lower()
cat = get_category(ext)
target_dir = folder / cat
target_dir.mkdir(exist_ok=True)
target_path = target_dir / item.name
if target_path.exists():
# 避免重名覆盖,加个时间戳
target_path = target_dir / f"{item.stem}_{int(time.time())}{item.suffix}"
shutil.move(str(item), str(target_path))
print(f"已移动: {item.name} -> {cat}/")
if __name__ == "__main__":
watch_dir = os.path.expanduser("~/Downloads")
print(f"开始监控: {watch_dir}")
while True:
organize_folder(watch_dir)
time.sleep(3)
这段代码能跑,但有个问题:它扫描的是整个目录,如果里面的文件很多,每3秒扫一遍会很耗资源,而且就算你只是打开一个文件然后关闭,它也会把文件重新“整理”到另一个地方。所以我后来改用了一个更“懒”的方式,只处理新增的文件,而且等文件写入完成后再移动。
升级版:尽量不碰正在使用的文件
文件下载到一半就被移动,会导致下载失败。我的解决方式是:发现新文件后,先记录下修改时间,然后等几秒再检查一次,如果文件大小和修改时间没有变化,才移动它。
import os
import time
import shutil
from pathlib import Path
def organize_pending_file(file_path):
"""延迟处理单个文件,避免移动正在下载的文件"""
if not file_path.exists():
return
size1 = file_path.stat().st_size
mtime1 = file_path.stat().st_mtime
time.sleep(2)
size2 = file_path.stat().st_size
mtime2 = file_path.stat().st_mtime
if size1 != size2 or mtime1 != mtime2:
print(f"文件还在变化,暂时不移动: {file_path.name}")
return
ext = file_path.suffix.lower()
cat = get_category(ext)
target_dir = file_path.parent / cat
target_dir.mkdir(exist_ok=True)
target_path = target_dir / file_path.name
if target_path.exists():
target_path = target_dir / f"{file_path.stem}_{int(time.time())}{file_path.suffix}"
shutil.move(str(file_path), str(target_path))
print(f"移走: {file_path.name} -> {cat}/")
然后在主循环里,只对“最近30秒内修改过”的文件调用这个函数。但这样还是要遍历整个目录。说实话,如果文件数量不多,问题不大。我电脑里的Downloads有几百个文件,轮询尚可接受。
再进一步:用 watchdog 做实时监听
真正的实时体验还是要靠事件驱动。Python里有个很成熟的库叫watchdog,它底层会根据不同系统调用对应的文件事件接口,不占CPU。
安装很简单:
pip install watchdog
完整的监听脚本如下:
import time
import shutil
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 文件分类映射
image_exts = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
doc_exts = {'.pdf', '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx', '.txt', '.md'}
zip_exts = {'.zip', '.rar', '.7z', '.tar', '.gz'}
program_exts = {'.exe', '.msi', '.dmg', '.deb', '.rpm'}
def get_category(ext):
if ext in image_exts:
return 'images'
elif ext in doc_exts:
return 'docs'
elif ext in zip_exts:
return 'archives'
elif ext in program_exts:
return 'programs'
return 'others'
class MyHandler(FileSystemEventHandler):
def on_created(self, event):
# 可能是文件夹创建,不处理
if event.is_directory:
return
path = Path(event.src_path)
print(f"发现新文件: {path.name}")
# 等1秒,给文件写入留点时间
time.sleep(1)
self._try_move(path)
def _try_move(self, path):
try:
ext = path.suffix.lower()
cat = get_category(ext)
target_dir = path.parent / cat
target_dir.mkdir(exist_ok=True)
target_path = target_dir / path.name
if target_path.exists():
target_path = target_dir / f"{path.stem}_{int(time.time())}{path.suffix}"
shutil.move(str(path), str(target_path))
print(f"已整理: {path.name} -> {cat}/")
except Exception as e:
# 文件可能被占用,或者已经不存在了
print(f"移动失败: {e}")
if __name__ == "__main__":
watch_dir = str(Path.home() / "Downloads")
event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, watch_dir, recursive=False)
observer.start()
print(f"开始实时监听: {watch_dir}")
try:
while True:
time.sleep(2)
except KeyboardInterrupt:
observer.stop()
observer.join()
这里on_created会在文件创建的那一刻触发,我用time.sleep(1)先等文件稳定下来。如果下载速度很慢,可能还是要加一个“确认写入完成”的机制。但日常用下来,大部分文件在1秒内都已经写完,表现还算不错。
我实际用的时候踩到的几个坑
1. 自己人打自己人。脚本在工作时会把文件移动到子目录,这个移动动作本身会触发新的事件。比如把report.pdf移进docs/,docs文件夹里就会触发一次on_created,然后脚本又会把它当作“新文件”尝试整理。因为docs/在目标目录内,但是路径变了,容易造成循环或不必要的操作。
解决方案其实很简单:给脚本加一个白名单,只处理watch_dir下的第一层文件。或者干脆忽略掉子目录里的事件。我的做法是在on_created里判断父目录是不是Downloads本身:
if path.parent.name != watch_name:
return
这样整理后的文件不会被二次整理。
2. 一些文件后缀是大写。比如.PDF或者.ZIP,所以我在每个判断前都先path.suffix.lower()。
3. 重名文件被覆盖。用时间戳追加在文件名后面,保留原扩展名,不会覆盖已有文件。
4. 移动某些系统文件会权限错误。在_try_move里包了一层try-except,失败就打个日志,不影响后续事件。
这个脚本给我们带来了什么
现在我的下载文件夹基本不会堆积超过10个文件,桌面也更清爽了。每天的工作文件在下载完成的瞬间就被归到了docs,我再也不用在巨大的下载列表里翻找“新建文本文档”。
更重要的是,我通过这次小工具,掌握了watchdog的基本用法。后来还给团队写了个内部工具:用同样的机制监听一个共享文件夹,当有人丢进来一个Excel时,自动提取数据并更新到数据库。那又是另一个故事了。
如果你也为乱七八糟的文件夹头疼,不妨复制上面的代码改一改。先从轮询版开始,等你发现它不够用了,再升级成watchdog版。写点小代码改善环境,是真的会让人开心。

