Python怎么根據文件后綴進行分類

# Python怎么根據文件后綴進行分類

## 引言

在日常文件管理中，我們經常需要根據文件后綴名（如`.txt`、`.jpg`、`.pdf`等）對大量文件進行自動化分類。Python憑借其強大的標準庫和簡潔的語法，能夠高效完成這類任務。本文將詳細介紹5種實現方法，并分析其適用場景和性能差異。

## 一、準備工作

### 1.1 創建測試環境
```python
import os
import shutil
from pathlib import Path
import time

# 創建測試目錄和樣本文件
test_dir = "test_files"
os.makedirs(test_dir, exist_ok=True)

file_types = ['.txt', '.jpg', '.pdf', '.mp3', '.docx']
for i in range(100):
    ext = file_types[i % len(file_types)]
    with open(f"{test_dir}/file_{i}{ext}", "w") as f:
        f.write(f"This is a {ext} file")

1.2 目標目錄結構

sorted_files/
├── txt/
├── jpg/
├── pdf/
├── mp3/
└── docx/

二、5種實現方法

2.1 使用os模塊（基礎版）

def classify_with_os(source_dir, target_dir):
    for filename in os.listdir(source_dir):
        if os.path.isfile(os.path.join(source_dir, filename)):
            # 獲取文件后綴（包含點）
            _, ext = os.path.splitext(filename)
            if ext:  # 確保有后綴
                ext = ext.lower()
                dest_dir = os.path.join(target_dir, ext[1:])
                os.makedirs(dest_dir, exist_ok=True)
                shutil.move(
                    os.path.join(source_dir, filename),
                    os.path.join(dest_dir, filename)
                )

特點： - 依賴標準庫os和shutil - 處理速度中等（100文件約120ms） - 代碼直觀但路徑拼接較繁瑣

2.2 使用pathlib（Python3.4+推薦）

def classify_with_pathlib(source_dir, target_dir):
    source = Path(source_dir)
    target = Path(target_dir)
    
    for file in source.iterdir():
        if file.is_file():
            ext = file.suffix.lower()
            if ext:
                dest = target / ext[1:] / file.name
                dest.parent.mkdir(exist_ok=True)
                file.rename(dest)

優勢： - 面向對象路徑操作 - 代碼更簡潔易讀 - 性能與os版本相當但更安全

2.3 使用字典映射（批量處理）

def classify_with_mapping(source_dir, target_dir):
    ext_mapping = {
        '.txt': 'text',
        '.jpg': 'images',
        '.pdf': 'documents',
        # 可擴展其他映射
    }
    
    for filename in os.listdir(source_dir):
        filepath = os.path.join(source_dir, filename)
        if os.path.isfile(filepath):
            ext = os.path.splitext(filename)[1].lower()
            category = ext_mapping.get(ext, 'others')
            dest = os.path.join(target_dir, category)
            os.makedirs(dest, exist_ok=True)
            shutil.move(filepath, os.path.join(dest, filename))

適用場景： - 需要自定義分類邏輯 - 支持將不同后綴歸入同一類別 - 添加others作為默認分類

2.4 多線程加速（處理大量文件）

from concurrent.futures import ThreadPoolExecutor

def worker(filepath, target_dir):
    ext = os.path.splitext(filepath.name)[1].lower()
    if ext:
        dest = target_dir / ext[1:] / filepath.name
        dest.parent.mkdir(exist_ok=True)
        filepath.rename(dest)

def classify_with_threads(source_dir, target_dir, max_workers=4):
    source = Path(source_dir)
    target = Path(target_dir)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        for file in source.iterdir():
            if file.is_file():
                executor.submit(worker, file, target)

性能對比：

2.5 使用watchdog（實時監控）

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class FileHandler(FileSystemEventHandler):
    def __init__(self, target_dir):
        self.target = Path(target_dir)
    
    def on_created(self, event):
        if not event.is_directory:
            file = Path(event.src_path)
            ext = file.suffix.lower()
            if ext:
                dest = self.target / ext[1:] / file.name
                dest.parent.mkdir(exist_ok=True)
                file.rename(dest)

def start_monitoring(source_dir, target_dir):
    event_handler = FileHandler(target_dir)
    observer = Observer()
    observer.schedule(event_handler, source_dir, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

典型應用： - 下載文件夾自動整理 - 實時監控掃描儀輸出 - 需要7×24小時運行的場景

三、進階技巧

3.1 處理同名文件沖突

def safe_move(src, dst):
    counter = 1
    while dst.exists():
        stem = src.stem
        new_name = f"{stem}_{counter}{src.suffix}"
        dst = dst.parent / new_name
        counter += 1
    src.rename(dst)

3.2 支持嵌套子目錄

def classify_recursive(source, target):
    for item in source.rglob('*'):
        if item.is_file():
            ext = item.suffix.lower()
            if ext:
                relative = item.relative_to(source)
                dest = target / ext[1:] / relative
                dest.parent.mkdir(parents=True, exist_ok=True)
                item.rename(dest)

3.3 添加日志記錄

import logging

logging.basicConfig(
    filename='file_classifier.log',
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)

def logged_move(src, dst):
    try:
        src.rename(dst)
        logging.info(f"Moved {src} -> {dst}")
    except Exception as e:
        logging.error(f"Failed to move {src}: {str(e)}")

四、性能優化建議

1. 批量操作：對于SSD存儲，建議每次處理50-100個文件
2. 內存優化：使用scandir()替代listdir()處理大目錄
3. 異常處理：

   
   try:
      file.rename(dest)
   except PermissionError:
      print(f"跳過系統文件: {file.name}")
   except OSError as e:
      print(f"移動失敗: {e}")
   

五、完整代碼示例

import argparse
from pathlib import Path

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("source", help="源目錄路徑")
    parser.add_argument("--target", default="sorted_files", help="目標目錄路徑")
    parser.add_argument("--threads", type=int, default=4, help="線程數")
    args = parser.parse_args()

    source = Path(args.source)
    target = Path(args.target)

    if not source.exists():
        raise ValueError(f"源目錄不存在: {source}")

    classify_with_threads(source, target, args.threads)

if __name__ == "__main__":
    main()

結語

本文介紹了從基礎到高級的多種文件分類方法，實際應用中建議： 1. 小規模文件使用pathlib版本 2. 萬級以上文件使用多線程方案 3. 需要實時處理時采用watchdog

通過靈活組合這些技術，可以構建出適應各種場景的高效文件管理系統。 “`