前言

為什麼要寫這篇文章呢。。。主要還是業務中有個需求，遍歷一個將近200w資料的資料夾，大部分還都是視訊檔案那種，但是這玩意用的次數還不多，做資料夾index也不是很ok，所以寫了一個指令碼來處理這個問題，從而發現了自己的一些薄弱點，將其記錄下來，方便自己，也方便未來其他的兄弟使用

基本需求

1. 把資料夾中的重複檔案找出來
2. 找出來之後用csv輸出，左邊是原始檔，右邊是重複檔案
3. 效率不能差，不能直接撐爆記憶體，不能佔用過多資源
4. 檢測的資料夾和存放csv的地方可以自己定義，加上終端互動
5. 重複檔案篩選支援md5，大小等方式

需求分析

首先要分析一點，就是我們該如何去做重複檔案的對比，並且效率還要高，首先網上過多的遞迴，os.walk的方法不可用，因為他們都會把遍歷到的內容直接做成一個大列表，塞到記憶體裡面，資料量大很容易爆掉，並且還要進行MD5，或者是大小比對，這個就非常難纏了。

基礎想法

其實說白了，拿到所有檔案列表file_list，把檔案依次對比，這裡我們可以用dict，分兩種情況

按照檔名和大小

設定兩個dict，例如record和dup，遍歷file_list,生成一個數組，比對其中的檔名和大小

按照大小和MD5值

設定兩個dict，例如record和dup，遍歷file_list,生成一個數組，比對其中的md5值和大小

具體程式碼

閒話休提，我們開始寫程式碼吧

定義遍歷函式程式碼

首先定義遍歷資料夾的部分diskwalk.py

# coding: utf-8
__author__ = "lau.wenbo"

import os,sys

class diskwalk(object):
    def __init__(self, path):
        self.path = path
    def paths(self):
        path = self.path
        # 這裡用了一個迭代器邏輯，防止所有資料塞記憶體爆掉
        path_collection = (os.path.join(root,fn) for root,dirs,files in os.walk(path) for fn in files)
        return path_collection
 

定義檢查md5值程式碼

接著我們定義檢查md5值的一個邏輯checksum.py

'''
遇到問題沒人解答？小編建立了一個Python學習交流QQ群：778463939
尋找有志同道合的小夥伴，互幫互助,群裡還有不錯的視訊學習教程和PDF電子書！
'''
# coding: utf-8
__author__ = "lau.wenbo"

import hashlib,sys

# 分塊讀MD，速度快
def create_checksum(path):
    fp = open(path)
    checksum = hashlib.md5()
    while True:
        buffer = fp.read(8192)
        if not buffer: break
        checksum.update(buffer)
    fp.close()
    checksum = checksum.digest()
    return checksum
 

定義主函式程式碼

# coding: utf-8
__author__ = "lau.wenbo"

from checksum import create_checksum
from diskwalk import diskwalk
from os.path import getsize
import csv
import os
import sys
reload(sys)
sys.setdefaultencoding('utf8')

def findDupes(path):
    record = {}
    dup = {}
    d = diskwalk(path)
    files = d.paths()
    for file in files:
        try:
            # 這裡使用了大小，檔名的對比方式，如果你需要MD5值的對比方式，可以開啟下面的註釋
            #compound_key = (getsize(file),create_checksum(file))
            compound_key = (getsize(file), file.split("/")[-1])
            if compound_key in record:
                dup[file] = record[compound_key]
            else:
                record[compound_key]=file
        except:
            continue
    return dup

if __name__ == '__main__':
    path = sys.argv[1]
    csv_path = sys.argv[2]
    if not os.path.isdir(path) or not os.path.isdir(csv_path) or csv_path[-1] != "/":
        print u"引數不是一個有效的資料夾！"
        exit()
    else:
        path = path.decode("utf-8")
        print u"待檢測的資料夾為{path}".format(path=path)
        with open(u"{csv_path}重複檔案.csv".format(csv_path=csv_path),"w+") as csvfile:
            # 原始檔 重複檔案
            header = ["Source", "Duplicate"]
            writer = csv.DictWriter(csvfile, fieldnames=header)
            writer.writeheader()
            print u"開始遍歷資料夾，尋找重複檔案，請等待........."
            print u"開始寫入CSV檔案，請等待........"
            for file in findDupes(path).items():
                writer.writerow({"Source":file[1],"Duplicate":file[0]})

結語

實現了哪些功能呢，哈哈，結尾來說一下，其實核心就是我用了一個列表生成器，加了一個迭代器，迭代器可是好東西，不會撐記憶體，不錯了，效率也還可以，200w資料判定也就20多分鐘，支援大資料量