Python 如何實現文件自動去重

Posted on by

Python 文件自動去重

平日裡一來無聊，二來手巧，果然下載瞭好多無(luan)比(qi)珍(ba)貴(zao)的資料，搞得我小小的硬盤（已經擴到6T瞭）捉襟見肘，

有次無意間，發現有兩個居然長得一毛一樣，在房子這麼小的情況下，我怎能忍兩個一毛一樣的東西不要臉皮的躺在我的硬盤裡，果斷搞掉一個，整理一下，本來想文件名一樣的就保留一份，但問題出現瞭，居然有名字一樣，內容卻完全不一樣的文件，想我背朝黃土面朝天吹著空調吃著西瓜下載下來的東西，刪除是不可能的，這輩子都是不可能刪除的。可是我也又不能把這數以億計的文件挨個打開看看裡面一樣不一樣吧，這個工程我大概夠我做瞭好久好久瞭，有沒有辦法搞個軟件幫幫我呢，答案是肯定的，要不然我也不用在這裡寫這個博客瞭（應該是苦逼的一個一個打開比較吧），說正題，Python提供瞭一個比較文件內容的東西，那就是。。。。。。。。。。哈希算法

MD5消息摘要算法（英語：MD5 Message-Digest Algorithm），一種被廣泛使用的密碼散列函數，可以產生出一個128位（16字節）的散列值（hash value），用於確保信息傳輸完整一致。MD5由美國密碼學傢羅納德·李維斯特（Ronald Linn Rivest）設計，於1992年公開，用以取代MD4算法。

說瞭這麼長，總結出來就一句，這玩意就是文件的指紋，幾乎每個文件是唯一的（碰到重復的，恭喜你，可以去買彩票瞭），那我們就把這個指紋拿出來，一個一個比對，肯定不能會有漏網的文件，既不會錯殺三千，也不使一文件漏網，原理上通瞭，那麼我們就要去搞個代碼來幫我完成這個工作，作為最好用的語言，Python就這樣被我翻瞭牌子

# -*- coding:utf-8 -*-
import os
import hashlib
import time
import sys
#搞到文件的MD5
def get_ms5(filename):
    m = hashlib.md5()
    mfile = open(filename , "rb")
    m.update(mfile.read())
    mfile.close()
    md5_value = m.hexdigest()
    return md5_value
#搞到文件的列表
def get_urllist():
    base = ("D:\\lwj\\spider\\pic\\")#這裡就是你要清繳的文件們瞭
    list = os.listdir(base)
    urllist = []
    for i in list:
        url = base + i
        urllist.append(url)
 
    return urllist
#主函數
if __name__ == '__main__':
    md5list = []
    urllist = get_urllist()
    print("test1")
    for a in urllist:
        md5 = get_ms5(a)
        if(md5 in md5list):
            os.remove(a)
            print("重復：%s" % a)
        else:
            md5list.append(md5)
            print("一共%s張照片" % len(md5list))

效果

python3 大文件去重

一、生成待去重數據

每行是固定位數的數字串

import os
from random import randint
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點()
 
# 用來配置的變量
位數 = 13
行數 = 500 * 10000
 
輸出目錄 = "./a_輸入"
輸出文件 = f"{輸出目錄}/隨機數.txt"
 
# 預處理
_00 = "".join(["0" for i in range(位數 - 1)])
_100 = "1" + _00
最小值 = int(_100)
_1000 = _100 + "0"
最大值 = int(_1000)
 
if not os.path.exists(輸出目錄):
    os.makedirs(輸出目錄)
#-- 輸出文件 = 文件名防重_追加數字(輸出文件)
 
# 實際處理
with open(輸出文件,"a") as f:
    for i in range(行數):
 
        f.write(f"{randint(最小值, 最大值)}\n")
 
        百分比 = (i+1) / 行數 * 100
        if 百分比 == int(百分比):
            print(f"已完成{int(百分比)}%")
#-- 打點()
#-- print(f"\n總耗時：{計時(0)}")
print("———— 結束 ————")

二、通過set按行去重

1. 按原值比較

（1）讀取全部數據

（2）用split來分行

（3）通過set數據結構來去除重復數據

（4）將set的數據寫入文件

import os
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點()
 
# 用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結果.txt"
# 預處理
# 目錄不存在就手動建立
if not os.path.exists(輸出目錄):
    os.makedirs(輸出目錄)
if not os.path.exists(輸入目錄):
    os.makedirs(輸入目錄)
#-- 輸出文件 = 文件名防重_追加數字(輸出文件)
 
# 獲取待去重文件
待去重文件列表 = []
待去重文件列表 = [f"{輸入目錄}/{i}" for i in os.listdir(輸入目錄)]
#-- getDeepFilePaths(待去重文件列表,輸入目錄,"txt")
print(f"\n總共{len(待去重文件列表)}個文件")
 
換行符 = b"\n"
if platform.system().lower() == 'windows':
    換行符 = b"\r\n"
 
# 實際處理
all_lines = []
文件個數 = 0
for 文件 in 待去重文件列表:
    文件個數 += 1
    print(f"\n處理第{文件個數}個文件")
 
    #-- 打點()
    # (1)讀全部
    with open(文件, "rb") as f:
        data = f.read()
 
    # (2)split分行
    lines = data.split(換行符)
    all_lines.extend(lines)
    #-- 打點()
    #-- print(f"分行完畢，耗時：{計時()}")
 
# (3)集合去重
all_lines_set = set(all_lines)
all_lines_set.remove(b"")
#-- 打點()
#-- print(f"\n\n去重完畢，耗時：{計時()}")
 
# (4)循環寫入
with open(輸出文件,"ab") as f_rst:
    for line in all_lines_set:
        f_rst.write(line + 換行符)
#-- 打點()
#-- print(f"\n寫入完畢，耗時：{計時()}")
print(f"\n輸出文件：{輸出文件}")
 
#-- 打點()
#-- print(f"\n\n總耗時：{計時(0)}")
print("———— 結束 ————")

附：

（2）用正則表達式來分行

import re
 
# (2)正則分行 二進制的話要加b， b''' '''
regx = '''[\w\~`\!\@\#\$\%\^\&\*\(\)\_\-\+\=\[\]\{\}\:\;\,\.\/\<\>\?]+'''
lines = re.findall(regx, data)

2. 按md5比較 

import hashlib
import os
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點()
 
# 用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結果.txt"
 
# 預處理
# 目錄不存在就手動建立
if not os.path.exists(輸出目錄):
    os.makedirs(輸出目錄)
if not os.path.exists(輸入目錄):
    os.makedirs(輸入目錄)
#-- 輸出文件 = 文件名防重_追加數字(輸出文件)
 
# 獲取待去重文件
待去重文件列表 = [f"{輸入目錄}/{i}" for i in os.listdir(輸入目錄)]
#-- 待去重文件列表 = []
#-- getDeepFilePaths(待去重文件列表,輸入目錄,"txt")
print(f"\n總共{len(待去重文件列表)}個文件")
 
def gen_md5(data):
    md5 = hashlib.md5()
    if repr(type(data)) == "<class 'str'>":
        data = data.encode('utf-8')
    md5.update(data)
    return md5.hexdigest()
 
# 實際處理
md5集 = set()
with open(輸出文件, "a") as f_rst:
    文件個數 = 0
    for 文件 in 待去重文件列表:
        文件個數 += 1
        print(f"\n處理第{文件個數}個文件")
 
        # 計算總行數
        with open(文件, 'rb') as f:
            行數 = 0
            buf_size = 1024 * 1024
            buf = f.read(buf_size)
            while buf:
                行數 += buf.count(b'\n')
                buf = f.read(buf_size)
 
        # 讀取、分行、去重、寫入
        #-- 打點()
        i = 0
        for line_帶換行 in open(文件):
            i += 1
            line = line_帶換行.strip()
            md5值 = gen_md5(line)
            if md5值 not in md5集:
                md5集.add(md5值)
                f_rst.write(line_帶換行)
 
            百分比 = i / 行數 * 10
            if 百分比 == int(百分比):
                print(f"已完成{int(百分比)*10}%")
                #-- 打點()
                #-- print(f"耗時：{計時()}")
 
print(f"\n輸出文件：{輸出文件}")
 
#-- 打點()
#-- print(f"\n\n總耗時：{計時(0)}")
print("———— 結束 ————")

三、二路歸並

import hashlib
import os
import platform
import queue
import shutil
from uuid import uuid1
from u_工具 import *
 
print("———— 開始 ————")
打點()
 
# 1.用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結果.txt"
臨時目錄 = "./c_臨時"
小文件大小 = 50 * 1024 * 1024  # 50M
 
# 2.預處理
# 目錄不存在就手動建立
if not os.path.exists(輸出目錄):
    os.makedirs(輸出目錄)
if not os.path.exists(輸入目錄):
    os.makedirs(輸入目錄)
if not os.path.exists(臨時目錄):
    os.makedirs(臨時目錄)
shutil.rmtree(臨時目錄)
os.makedirs(臨時目錄)
輸出文件 = 文件名防重_追加數字(輸出文件)
 
# 獲取待去重文件
# 待去重文件列表 = [f"{輸入目錄}/{i}" for i in os.listdir(輸入目錄)]
待去重文件列表 = []
getDeepFilePaths(待去重文件列表,輸入目錄,"txt")
print(f"總共{len(待去重文件列表)}個文件")
 
換行符 = b"\n"
if platform.system().lower() == 'windows':
    換行符 = b"\r\n"
 
# 3.實際處理
 
# (1)分割大文件
打點()
待排序文件列表 = []
待補全數據 = b""
for 文件 in 待去重文件列表:
    with open(文件, 'rb') as f:
        buf = f.read(小文件大小)
        while buf:
            data = buf.split(換行符,1)
            新路徑 = f"{臨時目錄}/無序_{序號(1)}_{uuid1()}.txt"
            with open(新路徑, 'ab') as ff:
                ff.write(待補全數據 + data[0])
            待排序文件列表.append(新路徑)
            try:
                待補全數據 = data[1]
            except:
                待補全數據 = b""
            buf = f.read(小文件大小)
    新路徑 = f"{臨時目錄}/無序_{序號(1)}_{uuid1()}.txt"
    with open(新路徑, 'ab') as ff:
            ff.write(待補全數據 + 換行符)
            待排序文件列表.append(新路徑)
    待補全數據 = b""
del buf,data,待補全數據
打點()
print(f"\n分割大文件完成，共耗時：{計時()}")
 
# (2)排序小文件
打點()
序號_重置(1)
待歸並文件隊列 = queue.Queue()
for 文件 in 待排序文件列表:
    with open(文件, "rb") as f:
        data = f.read()
    data = set(data.split(換行符))
    if b"" in data:
        data.remove(b"")
    if 換行符 in data:
        data.remove(換行符)
    data = sorted(data)
 
    新路徑 = f"{臨時目錄}/有序_{序號(1)}_{uuid1()}.txt"
    with open(新路徑, 'ab') as ff:
        for line in data:
            ff.write(line + 換行符)
    待歸並文件隊列.put(新路徑)
    os.remove(文件)
del data
打點()
print(f"\n排序小文件完成，共耗時：{計時()}")
 
# (3)歸並小文件
打點("歸並前")
序號_重置(1)
個數 = 待歸並文件隊列.qsize()
歸並次數 = 個數 - 1
print(f"\n\n歸並共{歸並次數}次")
當前次數 = 0
while 個數 > 1:
    當前次數 += 1
    print(f"\n執行第{當前次數}次歸並")
    文件路徑a = 待歸並文件隊列.get()
    文件路徑b = 待歸並文件隊列.get()
    新文件路徑 = f"{臨時目錄}/{序號(1)}_{uuid1()}.txt"
    if 當前次數 == 歸並次數:
        新文件路徑 = 輸出文件
    with open(文件路徑a,"rb") as 文件a, open(文件路徑b,"rb") as 文件b, open(新文件路徑,"wb") as ff:
        # region 歸並操作
        is_a_over = False
        is_b_over = False
 
        a = 文件a.readline().strip()
        b = 文件b.readline().strip()
        last = None
 
        while not (is_a_over and is_b_over):
 
            if is_a_over:
                b = 文件b.readline()
                if not b:
                    is_b_over = True
                else:
                    ff.write(b)
 
            elif is_b_over:
                a = 文件a.readline()
                if not a:
                    is_a_over = True
                else:
                    ff.write(a)
 
            else:
                # region 處理初始賦值
                if not a:
                    is_a_over = True
                    if not b:
                        is_b_over = True
                        continue
                    else:
                        ff.write(b + 換行符)
                        continue
 
                if not b:
                    is_b_over = True
                    ff.write(a + 換行符)
                    continue
                # endregion
 
                if a <= b:
                    if a == b or a == last:
                        a = 文件a.readline().strip()
                        if not a:
                            is_a_over = True
                            ff.write(b + 換行符)
                        continue
                    else:
                        last = a
                        ff.write(last + 換行符)
                        a = 文件a.readline().strip()
                        if not a:
                            is_a_over = True
                            ff.write(b + 換行符)
                        continue
                else:
                    if b == last:
                        b = 文件b.readline().strip()
                        if not b:
                            is_b_over = True
                            ff.write(a + 換行符)
                        continue
                    else:
                        last = b
                        ff.write(last + 換行符)
                        b = 文件b.readline().strip()
                        if not b:
                            is_b_over = True
                            ff.write(a + 換行符)
                        continue
        # endregion
 
    待歸並文件隊列.put(新文件路徑)
    os.remove(文件路徑a)
    os.remove(文件路徑b)
    個數 = 待歸並文件隊列.qsize()
    打點()
    print(f"耗時：{計時()}")
 
打點("歸並後")
print(f"\n\n歸並小文件完成，共耗時：{計時('歸並前','歸並後')}")
print(f"\n輸出文件：{輸出文件}")
 
打點()
print(f"\n\n總耗時：{計時(0)}")
print("———— 結束 ————")

以上為個人經驗，希望能給大傢一個參考，也希望大傢多多支持WalkonNet。

推薦閱讀：