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("———— 結束 ————")
以上為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
您可能感興趣的文章:- Python腳本實現自動登錄校園網
- python自動化運維之Telnetlib的具體使用
- 還在手動蓋樓抽獎?教你用Python實現自動評論蓋樓抽獎(一)
- Python爬蟲之自動爬取某車之家各車銷售數據
- python趣味挑戰之爬取天氣與微博熱搜并自動發給微信好友
- python 利用PyAutoGUI快速構建自動化操作腳本
- Python實現智慧校園自動評教全新版
