设为首页收藏本站
查看: 53|回复: 3

python去重小脚本

[复制链接]
H.U.C_刺客
  • TA的每日心情
    奋斗
    6 小时前

    • 签到天数: 2929 天

    [LV.Master]伴坛终老
    发表于 3 天前 | 显示全部楼层 |阅读模式
    1.png (130.74 KB, 下载次数: 0)
    2.png (114.06 KB, 下载次数: 0)
    3.png (89.65 KB, 下载次数: 0)
    4.png (102.08 KB, 下载次数: 0)
    5.png (103.72 KB, 下载次数: 0)
    6.png (122.47 KB, 下载次数: 0)
    7.png (164.87 KB, 下载次数: 0)
    8.png (106.19 KB, 下载次数: 0)
    9.png (198.37 KB, 下载次数: 0)
    10.png (98.63 KB, 下载次数: 0)

    1. # -*- coding: utf-8 -*-
    2. """
    3. 企业文件管理系统 v3.0
    4. 功能：1. 网络共享扫描  2. 智能文件去重  3. 操作审计报告
    5. """

    7. import os
    8. import csv
    9. import hashlib
    10. import time
    11. import sys
    12. from datetime import datetime
    13. from collections import defaultdict

    15. # 尝试加载GUI模块
    16. try:
    17.     import tkinter as tk
    18.     from tkinter import filedialog, messagebox
    19.     HAS_GUI = True
    20. except ImportError:
    21.     HAS_GUI = False

    23. # 配置参数
    24. DEFAULT_CHUNK_SIZE = 65536  # 64KB分块哈希计算
    25. MAX_LOG_ENTRIES = 1000      # 最大日志记录数
    26. SAFE_DIRECTORIES = {'Windows', 'Program Files'}  # 受保护目录

    28. class FileManager:
    29.     def __init__(self):
    30.         self.operation_log = []
    31.         self.error_log = []

    33.     def log_operation(self, event_type, path, metadata=None):
    34.         """记录操作日志"""
    35.         entry = {
    36.             "timestamp": datetime.now(),
    37.             "type": event_type,
    38.             "path": path,
    39.             "metadata": metadata or {}
    40.         }
    41.         self.operation_log.append(entry)
    42.         if len(self.operation_log) > MAX_LOG_ENTRIES:
    43.             self.operation_log.pop(0)

    45.     def get_hash(self, file_path):
    46.         """计算文件哈希值（SHA-256）"""
    47.         hasher = hashlib.sha256()
    48.         try:
    49.             with open(file_path, 'rb') as f:
    50.                 while chunk := f.read(DEFAULT_CHUNK_SIZE):
    51.                     hasher.update(chunk)
    52.             return hasher.hexdigest()
    53.         except Exception as e:
    54.             self.error_log.append(f"哈希计算失败 [{file_path}]: {str(e)}")
    55.             return None

    57.     def scan_network_share(self, target_path):
    58.         """深度扫描网络共享"""
    59.         scan_report = []
    60.         start_time = time.time()

    62.         def recursive_scan(current_path):
    63.             try:
    64.                 with os.scandir(current_path) as entries:
    65.                     for entry in entries:
    66.                         try:
    67.                             if entry.is_dir(follow_symlinks=False):
    68.                                 if not entry.name.startswith('$'):
    69.                                     recursive_scan(entry.path)
    70.                             else:
    71.                                 stat = entry.stat()
    72.                                 scan_report.append({
    73.                                     "path": entry.path,
    74.                                     "size": stat.st_size,
    75.                                     "modified": stat.st_mtime,
    76.                                     "hash": self.get_hash(entry.path)
    77.                                 })
    78.                         except PermissionError:
    79.                             self.error_log.append(f"权限不足: {entry.path}")
    80.                         except Exception as e:
    81.                             self.error_log.append(f"扫描错误: {entry.path} ({str(e)})")
    82.             except Exception as e:
    83.                 self.error_log.append(f"扫描中止: {current_path} ({str(e)})")

    85.         recursive_scan(target_path)
    86.         self.log_operation("NETSCAN", target_path, {
    87.             "duration": time.time() - start_time,
    88.             "files_scanned": len(scan_report)
    89.         })
    90.         return scan_report

    92.     def deduplicate_files(self, target_path):
    93.         """智能文件去重引擎"""
    94.         hash_registry = defaultdict(list)
    95.         deleted_records = []
    96.         preserved_files = set()

    98.         # 第一阶段：构建哈希索引
    99.         for root, dirs, files in os.walk(target_path):
    100.             # 跳过系统目录
    101.             dirs[:] = [d for d in dirs if d not in SAFE_DIRECTORIES]
    102.             
    103.             for file in files:
    104.                 file_path = os.path.join(root, file)
    105.                 file_hash = self.get_hash(file_path)
    106.                 if file_hash:
    107.                     hash_registry[file_hash].append(file_path)

    109.         # 第二阶段：处理重复文件
    110.         for file_hash, files in hash_registry.items():
    111.             if len(files) > 1:
    112.                 # 按修改时间排序（保留最新）
    113.                 sorted_files = sorted(
    114.                     files,
    115.                     key=lambda x: os.path.getmtime(x),
    116.                     reverse=True
    117.                 )
    118.                 preserved_file = sorted_files[0]
    119.                 preserved_files.add(preserved_file)

    121.                 # 记录删除操作
    122.                 for duplicate in sorted_files[1:]:
    123.                     try:
    124.                         file_stat = os.stat(duplicate)
    125.                         deleted_records.append({
    126.                             "deleted_path": duplicate,
    127.                             "preserved_path": preserved_file,
    128.                             "size": file_stat.st_size,
    129.                             "modified": file_stat.st_mtime,
    130.                             "hash": file_hash
    131.                         })
    132.                     except Exception as e:
    133.                         self.error_log.append(f"记录失败: {duplicate} ({str(e)})")

    135.         # 第三阶段：执行删除
    136.         success_count = 0
    137.         for record in deleted_records:
    138.             try:
    139.                 os.remove(record["deleted_path"])
    140.                 success_count += 1
    141.                 self.log_operation("DELETION", record["deleted_path"], {
    142.                     "preserved": record["preserved_path"],
    143.                     "hash": record["hash"][:12]
    144.                 })
    145.             except Exception as e:
    146.                 self.error_log.append(f"删除失败: {record['deleted_path']} ({str(e)})")

    148.         # 生成报告
    149.         report_path = self.generate_report(
    150.             target_path,
    151.             preserved_files,
    152.             deleted_records,
    153.             success_count
    154.         )

    156.         return {
    157.             "total_files": sum(len(v) for v in hash_registry.values()),
    158.             "duplicates": len(deleted_records),
    159.             "deleted": success_count,
    160.             "report_path": report_path
    161.         }

    163.     def generate_report(self, target_path, preserved, deleted, success_count):
    164.         """生成审计报告"""
    165.         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    166.         report_dir = os.path.join(os.getcwd(), "reports")
    167.         os.makedirs(report_dir, exist_ok=True)
    168.         report_path = os.path.join(report_dir,
    169.             f"DedupeReport_{os.path.basename(target_path)}_{timestamp}.csv")

    171.         with open(report_path, 'w', newline='', encoding='utf-8-sig') as f:
    172.             writer = csv.writer(f)
    173.             
    174.             # Header
    175.             writer.writerow(["企业文件管理审计报告"])
    176.             writer.writerow([f"目标目录: {target_path}"])
    177.             writer.writerow([f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"])
    178.             writer.writerow([])
    179.             
    180.             # 删除记录
    181.             writer.writerow(["删除文件路径", "大小 (MB)", "最后修改时间", "保留文件路径", "哈希片段"])
    182.             for record in deleted:
    183.                 writer.writerow([
    184.                     record["deleted_path"],
    185.                     round(record["size"] / 1048576, 2),
    186.                     datetime.fromtimestamp(record["modified"]).strftime('%Y-%m-%d %H:%M'),
    187.                     record["preserved_path"],
    188.                     record["hash"][:12]
    189.                 ])
    190.             
    191.             # 统计摘要
    192.             writer.writerow([])
    193.             writer.writerow(["总文件数", sum(len(v) for v in deleted) + len(preserved)])
    194.             writer.writerow(["发现重复", len(deleted)])
    195.             writer.writerow(["成功删除", success_count])
    196.             writer.writerow(["保留文件", len(preserved)])
    197.             writer.writerow(["错误数量", len(self.error_log)])

    199.         return report_path

    201. class UserInterface:
    202.     @staticmethod
    203.     def select_directory(gui_title="选择目录"):
    204.         """跨平台目录选择"""
    205.         if HAS_GUI:
    206.             root = tk.Tk()
    207.             root.withdraw()
    208.             root.attributes('-topmost', True)
    209.             path = filedialog.askdirectory(title=gui_title)
    210.             root.destroy()
    211.             return path
    212.         else:
    213.             print("\n命令行模式：")
    214.             while True:
    215.                 path = input("请输入完整路径（或输入Q退出）: ").strip()
    216.                 if path.upper() == 'Q':
    217.                     return None
    218.                 if os.path.isdir(path):
    219.                     return os.path.normpath(path)
    220.                 print("错误：无效路径")

    222.     @classmethod
    223.     def display_report(cls, report_path):
    224.         """报告展示"""
    225.         if HAS_GUI:
    226.             root = tk.Tk()
    227.             root.withdraw()
    228.             messagebox.showinfo(
    229.                 "操作完成",
    230.                 f"报告已生成至：\n{report_path}"
    231.             )
    232.             root.destroy()
    233.         else:
    234.             print(f"\n操作完成，报告路径：\n{report_path}")

    236. if __name__ == "__main__":
    237.     fm = FileManager()
    238.     ui = UserInterface()

    240.     # 操作菜单
    241.     print("\n企业文件管理系统")
    242.     print("1. 扫描网络共享")
    243.     print("2. 清理重复文件")
    244.     print("3. 退出")

    246.     choice = input("请选择操作：").strip()

    248.     if choice == '1':
    249.         target = ui.select_directory("选择扫描目录")
    250.         if target:
    251.             report = fm.scan_network_share(target)
    252.             print(f"扫描完成，发现 {len(report)} 个文件")

    254.     elif choice == '2':
    255.         target = ui.select_directory("选择清理目录")
    256.         if target:
    257.             # 安全确认
    258.             if HAS_GUI:
    259.                 confirm = messagebox.askyesno(
    260.                     "确认删除",
    261.                     "将永久删除重复文件！\n请确认已备份重要数据。"
    262.                 )
    263.             else:
    264.                 confirm = input("将永久删除文件，确认操作？(y/N): ").lower() == 'y'

    266.             if confirm:
    267.                 result = fm.deduplicate_files(target)
    268.                 ui.display_report(result['report_path'])
    269.                 print(f"删除完成：{result['deleted']}/{result['duplicates']} 成功")

    271.     print("\n操作日志：")
    272.     for entry in fm.operation_log[-3:]:
    273.         print(f"[{entry['timestamp']}] {entry['type']}: {entry['path']}")

    275.     if fm.error_log:
    276.         print("\n最近错误：")
    277.         for error in fm.error_log[-3:]:
    278.             print(f"! {error}")
    复制代码

    1. python python去重小脚本.py
    复制代码

    点评

    alglsf666
    666 :)  发表于 前天 20:00
    回复

    使用道具 举报

    H.U.C_刺客
  • TA的每日心情
    奋斗
    6 小时前

    • 签到天数: 2929 天

    [LV.Master]伴坛终老
     楼主| 发表于 前天 13:28 | 显示全部楼层


    python扫描重复文件
    1. import os
    2. import hashlib
    3. import tkinter as tk
    4. from tkinter import filedialog
    5. import xlwt  # 改用xlwt生成兼容性更好的xls格式
    6. import time

    8. def select_directory():
    9.     """选择文件夹对话框"""
    10.     root = tk.Tk()
    11.     root.withdraw()
    12.     return filedialog.askdirectory()

    14. def file_hash(filepath):
    15.     """计算文件的MD5哈希值（兼容旧系统）"""
    16.     hash_md5 = hashlib.md5()
    17.     try:
    18.         with open(filepath, "rb") as f:
    19.             while True:
    20.                 chunk = f.read(4096)
    21.                 if not chunk:
    22.                     break
    23.                 hash_md5.update(chunk)
    24.         return hash_md5.hexdigest()
    25.     except Exception as e:
    26.         print("[Error] 无法读取文件 {}: {}".format(filepath, str(e)))
    27.         return None

    29. def scan_files(directory, max_depth=None):
    30.     """扫描目录及子目录文件（兼容旧路径格式）"""
    31.     file_dict = {}
    32.     directory = os.path.normpath(directory)  # 规范化路径格式

    34.     for root, dirs, files in os.walk(directory):
    35.         # 计算当前深度
    36.         rel_path = os.path.relpath(root, directory)
    37.         current_depth = rel_path.count(os.sep) if rel_path != '.' else 0
    38.         
    39.         if max_depth is not None and current_depth >= max_depth:
    40.             del dirs[:]
    41.             continue

    43.         for file in files:
    44.             file_path = os.path.join(root, file)
    45.             try:
    46.                 file_size = os.path.getsize(file_path)
    47.                 mtime = os.path.getmtime(file_path)
    48.                 modified_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime))
    49.                 file_hash_value = file_hash(file_path)
    50.                
    51.                 if file_hash_value:
    52.                     key = (file_hash_value, file_size)
    53.                     if key not in file_dict:
    54.                         file_dict[key] = []
    55.                     file_dict[key].append({
    56.                         "path": file_path,
    57.                         "size": file_size,
    58.                         "modified": modified_time
    59.                     })
    60.             except Exception as e:
    61.                 print("[Error] 跳过文件 {}: {}".format(file_path, str(e)))
    62.     return file_dict

    64. def export_to_excel(duplicates, output_file="duplicates.xls"):
    65.     """导出重复文件到Excel 97-2003格式"""
    66.     wb = xlwt.Workbook(encoding='utf-8')
    67.     ws = wb.add_sheet('重复文件报告')

    69.     # 设置样式
    70.     header_style = xlwt.easyxf(
    71.         'font: bold on; align: wrap on, vert centre, horiz center;'
    72.         'borders: left thin, right thin, top thin, bottom thin;'
    73.     )
    74.     group_style = xlwt.easyxf(
    75.         'pattern: pattern solid, fore_colour pale_blue;'
    76.         'font: bold on;'
    77.     )

    79.     # 写入表头
    80.     ws.write(0, 0, "重复文件组", header_style)
    81.     ws.write(0, 1, "文件路径", header_style)
    82.     ws.write(0, 2, "文件大小 (Bytes)", header_style)
    83.     ws.write(0, 3, "最后修改时间", header_style)
    84.    
    85.     row = 1
    86.     group_num = 1
    87.     for key, files in duplicates.items():
    88.         if len(files) > 1:
    89.             # 合并单元格显示组信息
    90.             ws.write(row, 0, "重复文件组 {} (共 {} 个)".format(group_num, len(files)), group_style)
    91.             ws.write(row, 1, "MD5: {}".format(key[0]), group_style)
    92.             ws.row(row).set_style(xlwt.easyxf('pattern: pattern solid, fore_colour gray25;'))
    93.             
    94.             row +=1
    95.             for file in files:
    96.                 ws.write(row, 0, group_num)
    97.                 ws.write(row, 1, file["path"])
    98.                 ws.write(row, 2, file["size"])
    99.                 ws.write(row, 3, file["modified"])
    100.                 row +=1
    101.             group_num +=1
    102.    
    103.     # 调整列宽
    104.     ws.col(0).width = 3000
    105.     ws.col(1).width = 12000
    106.     ws.col(2).width = 4000
    107.     ws.col(3).width = 4000

    109.     wb.save(output_file)
    110.     print("报告已生成：{}".format(os.path.abspath(output_file)))

    112. def main():
    113.     """主程序"""
    114.     target_dir = select_directory()
    115.     if not target_dir:
    116.         print("操作已取消")
    117.         return

    119.     try:
    120.         depth_input = input("请输入扫描深度（0=仅当前目录，回车=不限深度）: ").strip()
    121.         max_depth = int(depth_input) if depth_input else None
    122.     except ValueError:
    123.         max_depth = None

    125.     print("扫描中...（可能需要较长时间）")
    126.     all_files = scan_files(target_dir, max_depth)
    127.     duplicates = {k: v for k, v in all_files.items() if len(v) > 1}

    129.     if duplicates:
    130.         print("发现 {} 组重复文件".format(len(duplicates)))
    131.         export_to_excel(duplicates)
    132.     else:
    133.         print("未发现重复文件")

    135. if __name__ == "__main__":
    136.     # 配置适用于旧系统的编码环境
    137.     import sys
    138.     if sys.version_info < (3,):
    139.         reload(sys)
    140.         sys.setdefaultencoding('utf-8')
    141.     main()
    复制代码
    回复 支持 1 反对 0

    使用道具 举报

    alglsf666
  • TA的每日心情
    奋斗
    4 小时前

    • 签到天数: 341 天

    [LV.8]以坛为家I
    发表于 前天 19:59 | 显示全部楼层
    刺客老哥牛逼
    回复 支持 反对

    使用道具 举报

    返回列表
    高级模式
    B Color Image Link Quote Code Smilies
    您需要登录后才可以回帖 登录 | 注册

    本版积分规则

    红盟社区--红客联盟 

    Processed in 0.066789 second(s), 28 queries.

    站点统计| 举报| Archiver| 手机版| 黑屋 |   

    备案号：冀ICP备20006029号-1 Powered by HUC © 2001-2021 Comsenz Inc.

    首页

    资讯

    社区

    互助

    家园

    群组

    文库

    文图

    组织活动

    总排行榜

    个人日志

    音乐电台

    爱心赞助

    每日签到

    微信机器

    微信图片

    手机扫我进入移动触屏客户端

    关注我们可获取更多热点资讯

    Honor accompaniments. theme macfee

    快速回复 返回顶部 返回列表