Python 文件自動去重
平日里一來無聊，二來手巧，果然下載了好多無(luan)比(qi)珍(ba)貴(zao)的資料，搞得我小小的硬盤（已經(jīng)擴(kuò)到6T了）捉襟見肘，
有次無意間，發(fā)現(xiàn)有兩個居然長得一毛一樣，在房子這么小的情況下，我怎能忍兩個一毛一樣的東西不要臉皮的躺在我的硬盤里，果斷搞掉一個，整理一下，本來想文件名一樣的就保留一份，但問題出現(xiàn)了，居然有名字一樣，內(nèi)容卻完全不一樣的文件，想我背朝黃土面朝天吹著空調(diào)吃著西瓜下載下來的東西，刪除是不可能的，這輩子都是不可能刪除的。可是我也又不能把這數(shù)以億計的文件挨個打開看看里面一樣不一樣吧，這個工程我大概夠我做了好久好久了，有沒有辦法搞個軟件幫幫我呢，答案是肯定的，要不然我也不用在這里寫這個博客了（應(yīng)該是苦逼的一個一個打開比較吧），說正題，Python提供了一個比較文件內(nèi)容的東西，那就是。。。。。。。。。。哈希算法
MD5消息摘要算法（英語：MD5 Message-Digest Algorithm），一種被廣泛使用的密碼散列函數(shù)，可以產(chǎn)生出一個128位（16字節(jié)）的散列值（hash value），用于確保信息傳輸完整一致。MD5由美國密碼學(xué)家羅納德·李維斯特（Ronald Linn Rivest）設(shè)計，于1992年公開，用以取代MD4算法。
說了這么長，總結(jié)出來就一句，這玩意就是文件的指紋，幾乎每個文件是唯一的（碰到重復(fù)的，恭喜你，可以去買彩票了），那我們就把這個指紋拿出來，一個一個比對，肯定不能會有漏網(wǎng)的文件，既不會錯殺三千，也不使一文件漏網(wǎng)，原理上通了，那么我們就要去搞個代碼來幫我完成這個工作，作為最好用的語言，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
#主函數(shù)
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("重復(fù)：%s" % a)
else:
md5list.append(md5)
print("一共%s張照片" % len(md5list))
效果
python3 大文件去重
一、生成待去重數(shù)據(jù)
每行是固定位數(shù)的數(shù)字串
import os
from random import randint
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點(diǎn)()
# 用來配置的變量
位數(shù) = 13
行數(shù) = 500 * 10000
輸出目錄 = "./a_輸入"
輸出文件 = f"{輸出目錄}/隨機(jī)數(shù).txt"
# 預(yù)處理
_00 = "".join(["0" for i in range(位數(shù) - 1)])
_100 = "1" + _00
最小值 = int(_100)
_1000 = _100 + "0"
最大值 = int(_1000)
if not os.path.exists(輸出目錄):
os.makedirs(輸出目錄)
#-- 輸出文件 = 文件名防重_追加數(shù)字(輸出文件)
# 實際處理
with open(輸出文件,"a") as f:
for i in range(行數(shù)):
f.write(f"{randint(最小值, 最大值)}\n")
百分比 = (i+1) / 行數(shù) * 100
if 百分比 == int(百分比):
print(f"已完成{int(百分比)}%")
#-- 打點(diǎn)()
#-- print(f"\n總耗時：{計時(0)}")
print("———— 結(jié)束 ————")
二、通過set按行去重
1. 按原值比較
（1）讀取全部數(shù)據(jù)
（2）用split來分行
（3）通過set數(shù)據(jù)結(jié)構(gòu)來去除重復(fù)數(shù)據(jù)
（4）將set的數(shù)據(jù)寫入文件
import os
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點(diǎn)()
# 用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結(jié)果.txt"
# 預(yù)處理
# 目錄不存在就手動建立
if not os.path.exists(輸出目錄):
os.makedirs(輸出目錄)
if not os.path.exists(輸入目錄):
os.makedirs(輸入目錄)
#-- 輸出文件 = 文件名防重_追加數(shù)字(輸出文件)
# 獲取待去重文件
待去重文件列表 = []
待去重文件列表 = [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 = []
文件個數(shù) = 0
for 文件 in 待去重文件列表:
文件個數(shù) += 1
print(f"\n處理第{文件個數(shù)}個文件")
#-- 打點(diǎn)()
# (1)讀全部
with open(文件, "rb") as f:
data = f.read()
# (2)split分行
lines = data.split(換行符)
all_lines.extend(lines)
#-- 打點(diǎn)()
#-- print(f"分行完畢，耗時：{計時()}")
# (3)集合去重
all_lines_set = set(all_lines)
all_lines_set.remove(b"")
#-- 打點(diǎn)()
#-- print(f"\n\n去重完畢，耗時：{計時()}")
# (4)循環(huán)寫入
with open(輸出文件,"ab") as f_rst:
for line in all_lines_set:
f_rst.write(line + 換行符)
#-- 打點(diǎn)()
#-- print(f"\n寫入完畢，耗時：{計時()}")
print(f"\n輸出文件：{輸出文件}")
#-- 打點(diǎn)()
#-- print(f"\n\n總耗時：{計時(0)}")
print("———— 結(jié)束 ————")
附：
（2）用正則表達(dá)式來分行
import re
# (2)正則分行 二進(jìn)制的話要加b， b''' '''
regx = '''[\w\~`\!\@\#\$\%\^\\*\(\)\_\-\+\=\[\]\{\}\:\;\,\.\/\\>\&;]+'''
lines = re.findall(regx, data)
2. 按md5比較
import hashlib
import os
#-- from u_工具 import *
print("———— 開始 ————")
#-- 打點(diǎn)()
# 用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結(jié)果.txt"
# 預(yù)處理
# 目錄不存在就手動建立
if not os.path.exists(輸出目錄):
os.makedirs(輸出目錄)
if not os.path.exists(輸入目錄):
os.makedirs(輸入目錄)
#-- 輸出文件 = 文件名防重_追加數(shù)字(輸出文件)
# 獲取待去重文件
待去重文件列表 = [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:
文件個數(shù) = 0
for 文件 in 待去重文件列表:
文件個數(shù) += 1
print(f"\n處理第{文件個數(shù)}個文件")
# 計算總行數(shù)
with open(文件, 'rb') as f:
行數(shù) = 0
buf_size = 1024 * 1024
buf = f.read(buf_size)
while buf:
行數(shù) += buf.count(b'\n')
buf = f.read(buf_size)
# 讀取、分行、去重、寫入
#-- 打點(diǎn)()
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 / 行數(shù) * 10
if 百分比 == int(百分比):
print(f"已完成{int(百分比)*10}%")
#-- 打點(diǎn)()
#-- print(f"耗時：{計時()}")
print(f"\n輸出文件：{輸出文件}")
#-- 打點(diǎn)()
#-- print(f"\n\n總耗時：{計時(0)}")
print("———— 結(jié)束 ————")
三、二路歸并
import hashlib
import os
import platform
import queue
import shutil
from uuid import uuid1
from u_工具 import *
print("———— 開始 ————")
打點(diǎn)()
# 1.用來配置的變量
輸入目錄 = "./a_輸入"
輸出目錄 = "./b_輸出"
輸出文件 = f"{輸出目錄}/去重結(jié)果.txt"
臨時目錄 = "./c_臨時"
小文件大小 = 50 * 1024 * 1024 # 50M
# 2.預(yù)處理
# 目錄不存在就手動建立
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(臨時目錄)
輸出文件 = 文件名防重_追加數(shù)字(輸出文件)
# 獲取待去重文件
# 待去重文件列表 = [f"{輸入目錄}/{i}" for i in os.listdir(輸入目錄)]
待去重文件列表 = []
getDeepFilePaths(待去重文件列表,輸入目錄,"txt")
print(f"總共{len(待去重文件列表)}個文件")
換行符 = b"\n"
if platform.system().lower() == 'windows':
換行符 = b"\r\n"
# 3.實際處理
# (1)分割大文件
打點(diǎn)()
待排序文件列表 = []
待補(bǔ)全數(shù)據(jù) = 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(待補(bǔ)全數(shù)據(jù) + data[0])
待排序文件列表.append(新路徑)
try:
待補(bǔ)全數(shù)據(jù) = data[1]
except:
待補(bǔ)全數(shù)據(jù) = b""
buf = f.read(小文件大小)
新路徑 = f"{臨時目錄}/無序_{序號(1)}_{uuid1()}.txt"
with open(新路徑, 'ab') as ff:
ff.write(待補(bǔ)全數(shù)據(jù) + 換行符)
待排序文件列表.append(新路徑)
待補(bǔ)全數(shù)據(jù) = b""
del buf,data,待補(bǔ)全數(shù)據(jù)
打點(diǎn)()
print(f"\n分割大文件完成，共耗時：{計時()}")
# (2)排序小文件
打點(diǎn)()
序號_重置(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
打點(diǎn)()
print(f"\n排序小文件完成，共耗時：{計時()}")
# (3)歸并小文件
打點(diǎn)("歸并前")
序號_重置(1)
個數(shù) = 待歸并文件隊列.qsize()
歸并次數(shù) = 個數(shù) - 1
print(f"\n\n歸并共{歸并次數(shù)}次")
當(dāng)前次數(shù) = 0
while 個數(shù) > 1:
當(dāng)前次數(shù) += 1
print(f"\n執(zhí)行第{當(dāng)前次數(shù)}次歸并")
文件路徑a = 待歸并文件隊列.get()
文件路徑b = 待歸并文件隊列.get()
新文件路徑 = f"{臨時目錄}/{序號(1)}_{uuid1()}.txt"
if 當(dāng)前次數(shù) == 歸并次數(shù):
新文件路徑 = 輸出文件
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)
個數(shù) = 待歸并文件隊列.qsize()
打點(diǎn)()
print(f"耗時：{計時()}")
打點(diǎn)("歸并后")
print(f"\n\n歸并小文件完成，共耗時：{計時('歸并前','歸并后')}")
print(f"\n輸出文件：{輸出文件}")
打點(diǎn)()
print(f"\n\n總耗時：{計時(0)}")
print("———— 結(jié)束 ————")
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
您可能感興趣的文章:- Python腳本實現(xiàn)自動登錄校園網(wǎng)
- python自動化運(yùn)維之Telnetlib的具體使用
- 還在手動蓋樓抽獎?教你用Python實現(xiàn)自動評論蓋樓抽獎(一)
- Python爬蟲之自動爬取某車之家各車銷售數(shù)據(jù)
- python趣味挑戰(zhàn)之爬取天氣與微博熱搜并自動發(fā)給微信好友
- python 利用PyAutoGUI快速構(gòu)建自動化操作腳本
- Python實現(xiàn)智慧校園自動評教全新版
