一、使用方法

1、将以下代码保存为 copy_tool_gui.py

2、运行 python copy_tool_gui.py

3、在界面中设置：

点击 “添加” 选择要复制的源文件夹（可多选） 选择目标根目录 编辑敏感词（默认已预置，可直接修改） 可选填写排除目录名（如 .git）或排除路径前缀（如 build/temp）

4、勾选/取消各项检查

5、点击 “开始复制” 若发现敏感文件等问题，会弹出确认对话框，选择“是”继续，否则跳过当前源。

二、源代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
文件夹复制安全检查工具 - 图形界面版（支持正则匹配）
"""

import os
import sys
import re
import shutil
import threading
import queue
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
from typing import List, Tuple, Optional, Set, Dict, Pattern

# ==================== 核心检测函数（升级正则支持） ====================

DEFAULT_SENSITIVE_WORDS = [
    'key', 'password', 'secret', 'token', 'api_key',
    'auth', 'credential', 'private_key', 'pwd'
]

TEXT_EXTENSIONS = {
    '.txt', '.log', '.cfg', '.conf', '.ini', '.properties',
    '.json', '.yaml', '.yml', '.xml', '.toml',
    '.py', '.js', '.java', '.c', '.cpp', '.h', '.sh', '.bash',
    '.env', '.config', '.sql', '.csv', '.md', '.rst'
}

def detect_file_type(file_path: str) -> Optional[str]:
    try:
        with open(file_path, 'rb') as f:
            header = f.read(12)
            if header.startswith(b'\x50\x4b\x03\x04') or \
               header.startswith(b'\x50\x4b\x05\x06') or \
               header.startswith(b'\x50\x4b\x07\x08'):
                return 'zip'
            elif header.startswith(b'\x25\x50\x44\x46'):
                return 'pdf'
            elif header.startswith(b'\xff\xd8\xff'):
                return 'jpg'
            elif header.startswith(b'\x89\x50\x4e\x47\x0d\x0a\x1a\x0a'):
                return 'png'
            elif header.startswith(b'GIF87a') or header.startswith(b'GIF89a'):
                return 'gif'
            elif header.startswith(b'BM'):
                return 'bmp'
            elif header.startswith(b'MZ'):
                return 'exe'
            elif header.startswith(b'PK'):
                return 'zip'
            else:
                return None
    except Exception:
        return None

def is_zip_encrypted(file_path: str) -> bool:
    if detect_file_type(file_path) != 'zip':
        return False
    try:
        with open(file_path, 'rb') as f:
            f.seek(6)
            flag_bytes = f.read(2)
            if len(flag_bytes) == 2:
                flags = int.from_bytes(flag_bytes, byteorder='little')
                return (flags & 0x0001) != 0
    except Exception:
        pass
    return False

def check_header_tampering(file_path: str) -> Optional[str]:
    ext = os.path.splitext(file_path)[1].lower()
    EXT_TYPE_MAP = {
        '.zip': 'zip', '.jar': 'zip', '.war': 'zip',
        '.pdf': 'pdf',
        '.jpg': 'jpg', '.jpeg': 'jpg',
        '.png': 'png',
        '.gif': 'gif',
        '.bmp': 'bmp',
        '.exe': 'exe', '.dll': 'exe',
    }
    expected = EXT_TYPE_MAP.get(ext)
    if expected is None:
        return None
    actual = detect_file_type(file_path)
    if actual is None:
        return None
    if actual != expected:
        return f"格式篡改: 扩展名 {ext}，实际类型为 {actual}"
    return None

def is_text_file(file_path: str) -> bool:
    ext = os.path.splitext(file_path)[1].lower()
    return ext in TEXT_EXTENSIONS

def check_file_for_sensitive_content(
    file_path: str,
    sensitive_patterns: List[Pattern],
    use_regex: bool,
    raw_words: List[str]
) -> List[str]:
    """
    检查文件内容是否匹配敏感词。
    若 use_regex 为 True，则使用编译好的正则表达式列表（已包含忽略大小写）。
    否则使用原始词列表进行子串匹配（忽略大小写）。
    返回匹配到的敏感词（或正则表达式原文）列表。
    """
    if not os.path.isfile(file_path) or not is_text_file(file_path):
        return []
    matched = set()
    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            for line in f:
                line_lower = line.lower() if not use_regex else line
                if use_regex:
                    for pattern in sensitive_patterns:
                        if pattern.search(line):
                            matched.add(pattern.pattern)  # 存储正则原文
                else:
                    # 普通子串匹配（不区分大小写）
                    line_lower = line.lower()
                    for word in raw_words:
                        if word.lower() in line_lower:
                            matched.add(word)
                if use_regex and len(matched) == len(sensitive_patterns):
                    break
                if not use_regex and len(matched) == len(raw_words):
                    break
    except Exception:
        pass
    return list(matched)

# ==================== 扫描与复制 ====================

def collect_files(
    source: str,
    exclude_dirs: Optional[List[str]] = None,
    exclude_paths: Optional[List[str]] = None
):
    exclude_dirs_set = set(exclude_dirs) if exclude_dirs else set()
    exclude_paths_list = [p.rstrip('/\\') for p in exclude_paths] if exclude_paths else []

    for root, dirs, files in os.walk(source):
        rel = os.path.relpath(root, source)
        should_skip = False

        if rel != '.':
            for pat in exclude_paths_list:
                if rel == pat or rel.startswith(pat + os.sep):
                    should_skip = True
                    break
            if not should_skip and os.path.basename(root) in exclude_dirs_set:
                should_skip = True

        if should_skip:
            dirs[:] = []
            continue

        for f in files:
            abs_path = os.path.join(root, f)
            rel_path = os.path.relpath(abs_path, source)
            yield abs_path, rel_path

def compile_regex_patterns(raw_words: List[str]) -> List[Pattern]:
    """编译正则表达式列表，忽略大小写；若编译失败则抛出异常"""
    patterns = []
    for word in raw_words:
        try:
            pat = re.compile(word, re.IGNORECASE)
            patterns.append(pat)
        except re.error as e:
            raise ValueError(f"无效的正则表达式 '{word}': {e}")
    return patterns

def scan_directory(
    source_dir: str,
    sensitive_words: List[str],
    use_regex: bool,
    check_header: bool = True,
    check_zip: bool = True,
    exclude_dirs: Optional[List[str]] = None,
    exclude_paths: Optional[List[str]] = None,
    log_func=None
) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
    """返回 (文件列表, 问题列表)"""
    if log_func:
        log_func(f"开始扫描: {source_dir}")

    # 编译正则（如果启用）
    patterns = None
    if use_regex:
        try:
            patterns = compile_regex_patterns(sensitive_words)
        except ValueError as e:
            if log_func:
                log_func(f"正则编译错误: {e}")
            raise

    file_list = list(collect_files(source_dir, exclude_dirs, exclude_paths))
    if log_func:
        log_func(f"找到 {len(file_list)} 个文件")

    issues = []
    for idx, (abs_path, rel_path) in enumerate(file_list, 1):
        if log_func and idx % 100 == 0:
            log_func(f"已检查 {idx}/{len(file_list)} 个文件")

        # 敏感词检查
        matched = check_file_for_sensitive_content(
            abs_path, patterns, use_regex, sensitive_words
        )
        if matched:
            issues.append((rel_path, f"包含敏感词: {', '.join(matched)}"))

        # 格式篡改
        if check_header:
            tamper = check_header_tampering(abs_path)
            if tamper:
                issues.append((rel_path, tamper))

        # ZIP加密
        if check_zip:
            if is_zip_encrypted(abs_path):
                issues.append((rel_path, "加密ZIP文件"))

    if log_func:
        log_func(f"扫描完成，发现 {len(issues)} 个问题")
    return file_list, issues

def copy_files(file_list: List[Tuple[str, str]], source: str, dest_root: str, log_func=None):
    folder_name = os.path.basename(os.path.normpath(source))
    dest_base = os.path.join(dest_root, folder_name)

    for abs_path, rel_path in file_list:
        dest_file = os.path.join(dest_base, rel_path)
        os.makedirs(os.path.dirname(dest_file), exist_ok=True)
        shutil.copy2(abs_path, dest_file)
        if log_func:
            log_func(f"复制: {rel_path}")

# ==================== 图形界面应用 ====================

class CopyApp(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("文件夹安全复制工具（支持正则）")
        self.geometry("820x720")
        self.resizable(True, True)

        self.log_queue = queue.Queue()
        self.confirm_queue = queue.Queue()

        self.build_ui()

        self.after(100, self.process_log_queue)
        self.after(100, self.process_confirm_queue)

        self.running = False

    # ---------- UI 构建 ----------
    def build_ui(self):
        main_frame = ttk.Frame(self, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)

        # 源文件夹
        src_frame = ttk.LabelFrame(main_frame, text="源文件夹", padding="5")
        src_frame.grid(row=0, column=0, columnspan=2, sticky="ew", padx=5, pady=5)

        self.src_listbox = tk.Listbox(src_frame, height=4, selectmode=tk.EXTENDED)
        self.src_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0,5))

        btn_frame = ttk.Frame(src_frame)
        btn_frame.pack(side=tk.RIGHT, fill=tk.Y)

        ttk.Button(btn_frame, text="添加", command=self.add_source).pack(fill=tk.X, pady=2)
        ttk.Button(btn_frame, text="移除", command=self.remove_source).pack(fill=tk.X, pady=2)

        # 目标目录
        dest_frame = ttk.LabelFrame(main_frame, text="目标根目录", padding="5")
        dest_frame.grid(row=1, column=0, columnspan=2, sticky="ew", padx=5, pady=5)

        self.dest_var = tk.StringVar()
        ttk.Entry(dest_frame, textvariable=self.dest_var).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0,5))
        ttk.Button(dest_frame, text="浏览...", command=self.browse_dest).pack(side=tk.RIGHT)

        # 敏感词（含正则开关）
        word_frame = ttk.LabelFrame(main_frame, text="敏感词（每行一个）", padding="5")
        word_frame.grid(row=2, column=0, sticky="nsew", padx=5, pady=5)

        self.word_text = scrolledtext.ScrolledText(word_frame, height=6, width=40)
        self.word_text.pack(fill=tk.BOTH, expand=True)
        self.word_text.insert(tk.END, "\n".join(DEFAULT_SENSITIVE_WORDS))

        # 正则复选框
        self.use_regex = tk.BooleanVar(value=False)
        ttk.Checkbutton(word_frame, text="启用正则表达式匹配", variable=self.use_regex).pack(anchor=tk.W, pady=5)

        # 排除设置
        exclude_frame = ttk.LabelFrame(main_frame, text="排除设置", padding="5")
        exclude_frame.grid(row=2, column=1, sticky="nsew", padx=5, pady=5)

        ttk.Label(exclude_frame, text="排除目录名（逗号分隔）:").pack(anchor=tk.W)
        self.exclude_dir_var = tk.StringVar()
        ttk.Entry(exclude_frame, textvariable=self.exclude_dir_var).pack(fill=tk.X, pady=(0,10))

        ttk.Label(exclude_frame, text="排除路径前缀（逗号分隔）:").pack(anchor=tk.W)
        self.exclude_path_var = tk.StringVar()
        ttk.Entry(exclude_frame, textvariable=self.exclude_path_var).pack(fill=tk.X)

        # 选项
        opt_frame = ttk.LabelFrame(main_frame, text="检查选项", padding="5")
        opt_frame.grid(row=3, column=0, columnspan=2, sticky="ew", padx=5, pady=5)

        self.check_header = tk.BooleanVar(value=True)
        self.check_zip = tk.BooleanVar(value=True)
        self.force_mode = tk.BooleanVar(value=False)

        ttk.Checkbutton(opt_frame, text="文件头篡改检测", variable=self.check_header).pack(side=tk.LEFT, padx=10)
        ttk.Checkbutton(opt_frame, text="加密ZIP检测", variable=self.check_zip).pack(side=tk.LEFT, padx=10)
        ttk.Checkbutton(opt_frame, text="强制模式（跳过确认）", variable=self.force_mode).pack(side=tk.LEFT, padx=10)

        # 执行按钮
        btn_frame2 = ttk.Frame(main_frame)
        btn_frame2.grid(row=4, column=0, columnspan=2, pady=10)
        self.start_btn = ttk.Button(btn_frame2, text="开始复制", command=self.start_copy)
        self.start_btn.pack(side=tk.LEFT, padx=5)
        self.stop_btn = ttk.Button(btn_frame2, text="停止", command=self.stop_copy, state=tk.DISABLED)
        self.stop_btn.pack(side=tk.LEFT, padx=5)

        # 日志区域
        log_frame = ttk.LabelFrame(main_frame, text="日志", padding="5")
        log_frame.grid(row=5, column=0, columnspan=2, sticky="nsew", padx=5, pady=5)
        self.log_text = scrolledtext.ScrolledText(log_frame, height=12, state='normal')
        self.log_text.pack(fill=tk.BOTH, expand=True)

        main_frame.columnconfigure(0, weight=1)
        main_frame.columnconfigure(1, weight=1)
        main_frame.rowconfigure(2, weight=1)
        main_frame.rowconfigure(5, weight=1)

    # ---------- 回调函数 ----------
    def log(self, msg: str):
        self.log_queue.put(msg)

    def confirm(self, issues: List[Tuple[str, str]], source: str) -> bool:
        event = threading.Event()
        self.confirm_queue.put((issues, source, event))
        event.wait()
        return event.result

    # ---------- 队列处理 ----------
    def process_log_queue(self):
        while not self.log_queue.empty():
            msg = self.log_queue.get_nowait()
            self.log_text.insert(tk.END, msg + "\n")
            self.log_text.see(tk.END)
        self.after(100, self.process_log_queue)

    def process_confirm_queue(self):
        while not self.confirm_queue.empty():
            issues, source, event = self.confirm_queue.get_nowait()
            self.log(f"--- 源目录: {source} 发现问题 {len(issues)} 个 ---")
            for rel, msg in issues[:20]:
                self.log(f"  {rel}: {msg}")
            if len(issues) > 20:
                self.log(f"  ... 还有 {len(issues)-20} 个问题")
            result = messagebox.askyesno(
                "确认复制",
                f"源文件夹 '{os.path.basename(source)}' 发现 {len(issues)} 个问题文件。\n是否继续复制？",
                parent=self
            )
            event.result = result
            event.set()
        self.after(100, self.process_confirm_queue)

    # ---------- 界面事件 ----------
    def add_source(self):
        dirs = filedialog.askdirectory(title="选择要复制的源文件夹", parent=self)
        if dirs:
            self.src_listbox.insert(tk.END, dirs)

    def remove_source(self):
        selected = self.src_listbox.curselection()
        for idx in reversed(selected):
            self.src_listbox.delete(idx)

    def browse_dest(self):
        dir_ = filedialog.askdirectory(title="选择目标根目录", parent=self)
        if dir_:
            self.dest_var.set(dir_)

    # ---------- 复制任务 ----------
    def start_copy(self):
        if self.running:
            messagebox.showwarning("警告", "复制任务正在运行，请先停止")
            return

        sources = list(self.src_listbox.get(0, tk.END))
        if not sources:
            messagebox.showerror("错误", "请至少添加一个源文件夹")
            return
        dest = self.dest_var.get().strip()
        if not dest:
            messagebox.showerror("错误", "请选择目标根目录")
            return
        if not os.path.exists(dest):
            try:
                os.makedirs(dest)
            except Exception as e:
                messagebox.showerror("错误", f"无法创建目标目录: {e}")
                return

        words_text = self.word_text.get("1.0", tk.END).strip()
        sensitive_words = [w.strip() for w in words_text.splitlines() if w.strip()]
        if not sensitive_words:
            messagebox.showerror("错误", "请至少输入一个敏感词")
            return

        use_regex = self.use_regex.get()
        if use_regex:
            # 提前验证所有正则是否有效
            try:
                compile_regex_patterns(sensitive_words)
            except ValueError as e:
                messagebox.showerror("正则错误", str(e))
                return

        exclude_dirs = [d.strip() for d in self.exclude_dir_var.get().split(',') if d.strip()]
        exclude_paths = [p.strip() for p in self.exclude_path_var.get().split(',') if p.strip()]

        check_header = self.check_header.get()
        check_zip = self.check_zip.get()
        force = self.force_mode.get()

        self.log_text.delete("1.0", tk.END)
        self.log("=== 开始复制任务 ===")
        self.log(f"源文件夹: {len(sources)} 个")
        self.log(f"目标根目录: {dest}")
        self.log(f"敏感词: {sensitive_words}")
        self.log(f"使用正则: {use_regex}")
        self.log(f"排除目录名: {exclude_dirs}")
        self.log(f"排除路径前缀: {exclude_paths}")
        self.log(f"文件头检查: {check_header}, ZIP加密检测: {check_zip}, 强制模式: {force}")

        self.start_btn.config(state=tk.DISABLED)
        self.stop_btn.config(state=tk.NORMAL)
        self.running = True

        self.copy_thread = threading.Thread(
            target=self.run_copy,
            args=(sources, dest, sensitive_words, use_regex, exclude_dirs, exclude_paths,
                  check_header, check_zip, force),
            daemon=True
        )
        self.copy_thread.start()

    def run_copy(self, sources, dest, sensitive_words, use_regex, exclude_dirs, exclude_paths,
                 check_header, check_zip, force):
        success_count = 0
        for src in sources:
            if not self.running:
                self.log("任务被用户停止")
                break
            try:
                result = self.copy_single_folder(
                    src, dest, sensitive_words, use_regex, exclude_dirs, exclude_paths,
                    check_header, check_zip, force
                )
                if result:
                    success_count += 1
            except Exception as e:
                self.log(f"处理 {src} 时发生异常: {e}")
        self.log(f"=== 任务完成，成功复制 {success_count}/{len(sources)} 个目录 ===")
        self.after(0, self.finish_copy)

    def copy_single_folder(self, source, dest_root, sensitive_words, use_regex, exclude_dirs, exclude_paths,
                           check_header, check_zip, force) -> bool:
        folder_name = os.path.basename(os.path.normpath(source))
        dest_path = os.path.join(dest_root, folder_name)

        self.log(f"\n--- 处理源: {source} ---")

        try:
            file_list, issues = scan_directory(
                source, sensitive_words, use_regex, check_header, check_zip,
                exclude_dirs, exclude_paths, log_func=self.log
            )
        except ValueError as e:
            self.log(f"正则错误: {e}")
            return False

        if issues and not force:
            if not self.confirm(issues, source):
                self.log(f"用户取消复制，跳过 {source}")
                return False

        if os.path.exists(dest_path):
            if not force:
                resp = messagebox.askyesno(
                    "目标已存在",
                    f"目标目录 {dest_path} 已存在，是否删除并重新复制？",
                    parent=self
                )
                if not resp:
                    self.log(f"跳过 {source}（用户取消覆盖）")
                    return False
            try:
                shutil.rmtree(dest_path)
                self.log(f"已删除旧目录: {dest_path}")
            except Exception as e:
                self.log(f"删除目录失败: {e}")
                return False

        try:
            self.log(f"开始复制 {len(file_list)} 个文件...")
            copy_files(file_list, source, dest_root, log_func=self.log)
            self.log(f"✅ 复制成功: {source} -> {dest_path}")
            return True
        except Exception as e:
            self.log(f"❌ 复制失败: {e}")
            return False

    def finish_copy(self):
        self.running = False
        self.start_btn.config(state=tk.NORMAL)
        self.stop_btn.config(state=tk.DISABLED)

    def stop_copy(self):
        self.log("用户请求停止...")
        self.running = False
        if self.copy_thread and self.copy_thread.is_alive():
            self.copy_thread.join(timeout=1.0)

# ==================== 启动 ====================
if __name__ == "__main__":
    app = CopyApp()
    app.mainloop()

三、功能

1、支持ZIP加密文件的识别

2、支持纯文本文件敏感识别

3、支持文件类型不一致识别

4、支持正则

ps：本脚本仅仅对纯文本文件进行检查，可以通过AI进行进一步优化，增加图片识别，PDF，DOCX等文档

四、适用场景

适用于敏感文件初次筛选

来源：deepseek AI 生成