扫描并处理重重复文件
importosimportshutilimporthashlibimportchardetimportpandasaspdfromdocximportDocumentfromcollectionsimportdefaultdict# ====================== 全局配置区（按需修改） ======================SCAN_FOLDER=r"D:\data\mydata\t_data\test"# 需要扫描查重的根目录BACKUP_FOLDER=r"D:\data\mydata\t_data\_重复文件备份"# 重复文件移动备份目录OUTPUT_REPORT=r"D:\测试文件目录_查重清单.xlsx"# 输出查重清单ExcelLOG_PATH=r"D:\文件扫描日志.txt"# 异常日志保存路径MOVE_DUPLICATE=True# 是否自动移动重复文件副本到备份文件夹MAX_FILE_SIZE_MB=500# 大于500MB的超大文件跳过哈希计算，仅记录SIMILAR_THRESHOLD=0.85# 文本相似度阈值，大于等于判定高度相似# 跳过扫描的隐藏/系统目录SKIP_DIR_KEYWORDS={"$recycle.bin","system volume information",".git","__pycache__"}# 不处理的快捷/链接后缀SKIP_SUFFIX={".lnk",".symlink"}# 支持提取文本做相似比对的后缀TEXT_SUFFIX={".txt",".csv",".md",".docx",".xlsx",".xls"}# ====================== 模块1：工具基础函数 ======================defget_file_md5(file_path:str,block_size=65536)->str:""" 计算文件MD5哈希值，分块读取适配大文件 :param file_path: 文件绝对路径 :param block_size: 单次读取缓冲区大小 :return: md5字符串，读取失败返回空字符串 """try:md5_obj=hashlib.md5()withopen(file_path,"rb")asf:whilechunk:=f.read(block_size):md5_obj.update(chunk)returnmd5_obj.hexdigest()exceptExceptionase:log_msg=f"【哈希计算失败】{file_path}| 错误：{str(e)}"write_log(log_msg)return""defget_file_size_mb(file_path:str)->float:"""获取文件大小MB"""byte_size=os.path.getsize(file_path)returnround(byte_size/1024/1024,3)defwrite_log(msg:str):"""全局日志写入，追加模式"""withopen(LOG_PATH,"a",encoding="utf-8")asf:f.write(msg+"\n")print(msg)defsafe_move_file(src:str,dst_dir:str):""" 安全移动文件，目标存在自动重命名，避免冲突 :param src: 原文件路径 :param dst_dir: 目标文件夹 """os.makedirs(dst_dir,exist_ok=True)file_name=os.path.basename(src)dst_path=os.path.join(dst_dir,file_name)# 文件存在则循环加后缀区分idx=1whileos.path.exists(dst_path):name_no_ext,ext=os.path.splitext(file_name)dst_path=os.path.join(dst_dir,f"{name_no_ext}_副本{idx}{ext}")idx+=1try:shutil.move(src,dst_path)write_log(f"【已移动重复文件】原路径:{src}新路径:{dst_path}")exceptExceptionase:write_log(f"【文件移动失败】{src}错误:{str(e)}")# ====================== 模块2：文本提取函数（用于相似文件比对） ======================defextract_file_text(file_path:str,suffix:str)->str:""" 提取txt/docx/xlsx/csv内全部文本，用于相似度判断 :param file_path: 文件路径 :param suffix: 文件后缀小写 :return: 拼接后的纯文本，读取失败返回空字符串 """full_text=""try:ifsuffix==".txt"orsuffix==".csv"orsuffix==".md":# 自动识别编码读取文本withopen(file_path,"rb")asf:raw_data=f.read()encode_info=chardet.detect(raw_data)encode=encode_info.get("encoding","utf-8")full_text=raw_data.decode(encode,errors="ignore")elifsuffix==".docx":doc=Document(file_path)para_text=[p.textforpindoc.paragraphsifp.text.strip()]full_text=" ".join(para_text)elifsuffixin(".xlsx",".xls"):df=pd.read_excel(file_path,sheet_name=None)all_sheet_text=[]forsheet_dataindf.values():text_line=sheet_data.to_string()all_sheet_text.append(text_line)full_text=" ".join(all_sheet_text)exceptExceptionase:write_log(f"【文本提取失败】{file_path}错误:{str(e)}")return""# 清理空白字符，压缩文本减少比对开销full_text=full_text.replace("\n","").replace(" ","")returnfull_textdefcalc_text_similarity(text1:str,text2:str)->float:"""简易文本相似度（字符重合度）0~1"""set1=set(text1)set2=set(text2)ifnotset1andnotset2:return1.0inter=len(set1&set2)union=len(set1|set2)returninter/unionifunion!=0else0# ====================== 模块3：目录扫描函数 ======================defscan_all_files(root_dir:str)->list:"""遍历目录，过滤无效文件，返回全部有效文件信息列表"""file_info_list=[]forroot,dirs,filesinos.walk(root_dir):# 过滤系统隐藏目录，原地修改dirs阻止进入dirs[:]=[dfordindirsifd.lower()notinSKIP_DIR_KEYWORDS]forfile_nameinfiles:full_path=os.path.abspath(os.path.join(root,file_name))suffix=os.path.splitext(file_name)[1].lower()# 跳过快捷方式/链接ifsuffixinSKIP_SUFFIX:continuetry:byte_size=os.path.getsize(full_path)mb_size=round(byte_size/1024/1024,3)modify_time=os.path.getmtime(full_path)file_info_list.append({"file_path":full_path,"file_name":file_name,"suffix":suffix,"size_mb":mb_size,"size_byte":byte_size,"modify_time":modify_time})exceptExceptionase:write_log(f"【文件信息读取失败】{full_path}错误:{str(e)}")write_log(f"目录扫描完成，共获取有效文件：{len(file_info_list)}个")returnfile_info_list# ====================== 模块4：查重核心分组逻辑 ======================defgroup_duplicate_files(file_info_list:list):""" 核心查重分组：1.按大小分组 2.同大小计算MD5分完全重复组 3.文本相似分组 :return: duplicate_groups 完全重复组列表; similar_groups 高度相似组列表 """# 1. 先按文件字节大小分组，大小不同直接不可能重复size_group=defaultdict(list)forinfoinfile_info_list:size_group[info["size_byte"]].append(info)# 2. MD5哈希分组 = 完全重复文件组hash_groups=defaultdict(list)forsize_byte,info_listinsize_group.items():# 单个文件无重复，跳过iflen(info_list)<=1:continueforinfoininfo_list:# 超大文件跳过哈希计算ifinfo["size_mb"]>MAX_FILE_SIZE_MB:continuemd5_val=get_file_md5(info["file_path"])ifmd5_val:info["md5"]=md5_val hash_groups[md5_val].append(info)# 过滤仅单个文件的组duplicate_groups=[gforginhash_groups.values()iflen(g)>=2]# 3. 文本类文件相似度分组（非完全重复，但内容近似）similar_groups=[]# 提取所有支持文本比对、且不在完全重复组内的文件text_candidate=[]all_dup_path=set()forgroupinduplicate_groups:foritemingroup:all_dup_path.add(item["file_path"])forinfoinfile_info_list:ifinfo["suffix"]inTEXT_SUFFIXandinfo["file_path"]notinall_dup_path:info["text_content"]=extract_file_text(info["file_path"],info["suffix"])text_candidate.append(info)# 两两比对相似度used_idx=set()total_text=len(text_candidate)foriinrange(total_text):ifiinused_idx:continuegroup_temp=[text_candidate[i]]text_i=text_candidate[i]["text_content"]forjinrange(i+1,total_text):ifjinused_idx:continuetext_j=text_candidate[j]["text_content"]sim=calc_text_similarity(text_i,text_j)ifsim>=SIMILAR_THRESHOLD:group_temp.append(text_candidate[j])used_idx.add(j)iflen(group_temp)>=2:similar_groups.append(group_temp)write_log(f"查重完成：完全重复文件组{len(duplicate_groups)}组，高度相似文件组{len(similar_groups)}组")returnduplicate_groups,similar_groups# ====================== 模块5：生成输出清单 & 移动重复文件 ======================defgenerate_report_and_handle_dup(duplicate_groups,similar_groups):""" 1. 生成简易Excel查重清单 2. 可选：移动重复副本至备份文件夹 """report_rows=[]group_id=1# 写入完全重复组forgroupinduplicate_groups:base_file=group[0]# 保留第一个原始文件dup_file_list=group[1:]# 其余全部判定为重复副本base_path=base_file["file_path"]dup_paths="; ".join([f["file_path"]forfindup_file_list])# 写入报表行row={"分组ID":group_id,"重复类型":"完全重复(二进制一致)","主文件(保留)":base_path,"重复副本文件":dup_paths,"文件大小MB":base_file["size_mb"],"MD5哈希":base_file["md5"],"文件后缀":base_file["suffix"]}report_rows.append(row)# 移动重复副本文件ifMOVE_DUPLICATE:fordup_infoindup_file_list:safe_move_file(dup_info["file_path"],BACKUP_FOLDER)group_id+=1# 写入高度相似组（不移动，仅人工复核）forgroupinsimilar_groups:all_path="; ".join([f["file_path"]forfingroup])row={"分组ID":group_id,"重复类型":"高度相似(文本内容接近)","主文件(保留)":all_path,"重复副本文件":"无自动移动，建议人工复核","文件大小MB":group[0]["size_mb"],"MD5哈希":"不相同","文件后缀":group[0]["suffix"]}report_rows.append(row)group_id+=1# 导出简易Excel清单df_report=pd.DataFrame(report_rows)df_report.to_excel(OUTPUT_REPORT,index=False)write_log(f"查重清单已导出至：{OUTPUT_REPORT}")# ====================== 程序入口主函数 ======================defmain():# 清空历史日志withopen(LOG_PATH,"w",encoding="utf-8")asf:f.write("======= 文件查重扫描日志 开始 =======\n")# 1. 扫描全部文件file_info=scan_all_files(SCAN_FOLDER)# 2. 查重分组dup_groups,sim_groups=group_duplicate_files(file_info)# 3. 生成报表+处理重复文件移动generate_report_and_handle_dup(dup_groups,sim_groups)write_log("======= 文件查重全部执行完毕 =======")print(f"\n执行完成！\n清单文件：{OUTPUT_REPORT}\n操作日志：{LOG_PATH}")ifMOVE_DUPLICATE:print(f"重复副本已移动至备份目录：{BACKUP_FOLDER}")if__name__=="__main__":main()
