Python+wxPython从零编写一个AES加 ZIP工具(附源码)

pip install wxPython pyzipper pillow

def app_dir():
    """程序所在目录：.py 直接运行取脚本目录，PyInstaller 打包后取 exe 目录。"""
    if getattr(sys, "frozen", False):
        # onefile 模式下 sys.executable 是 exe 本身，不是解包的临时目录
        return os.path.dirname(os.path.abspath(sys.executable))
    return os.path.dirname(os.path.abspath(__file__))


BASE_DIR = app_dir()
CONFIG_PATH = os.path.join(BASE_DIR, CONFIG_NAME)
DB_PATH = os.path.join(BASE_DIR, DB_NAME)

CONFIG_PATH = "config.json"          # ❌

BASE_DIR = os.path.dirname(os.path.abspath(__file__))    # ❌ 打包后就错了

BASE_DIR = sys._MEIPASS                                   # ❌ 语义反了

self.CreateStatusBar()
self.SetStatusText(f"配置: {CONFIG_PATH}    数据库: {DB_PATH}")

import zipfile
zf = zipfile.ZipFile("a.zip")
zf.setpassword(b"123456")     # ✅ 能用：解密读取传统 ZipCrypto

zf = zipfile.ZipFile("a.zip", "w")
zf.setpassword(b"123456")     # ⚠️ 有这个方法，但对写入完全无效
zf.write("f.txt")             # 结果：一个毫无加密的普通 zip

with pyzipper.AESZipFile(
        self.zip_path, "w",
        compression=pyzipper.ZIP_DEFLATED,
        encryption=pyzipper.WZ_AES,
) as zf:
    zf.setpassword(self.password.encode("utf-8"))
    zf.setencryption(pyzipper.WZ_AES, nbits=self.nbits)
    ...
    zf.write(path, arcname)

AES_BITS = {"AES-256 (最强)": 256, "AES-192": 192, "AES-128 (兼容性好)": 128}
self.cho_enc = wx.Choice(box_opt, choices=list(AES_BITS.keys()))

used = set()
...
arcname = os.path.basename(path)
stem, ext = os.path.splitext(arcname)
n = 1
while arcname.lower() in used:
    arcname = f"{stem}({n}){ext}"
    n += 1
used.add(arcname.lower())

ProgressEvent, EVT_PROGRESS = wx.lib.newevent.NewEvent()
DoneEvent, EVT_DONE = wx.lib.newevent.NewEvent()

wx.PostEvent(self.sink, ProgressEvent(current=i, total=total, name=arcname))

self.Bind(EVT_PROGRESS, self.on_progress)
self.Bind(EVT_DONE, self.on_done)

def on_progress(self, evt):
    self.gauge.SetValue(evt.current)
    self.set_status(f"正在压缩 ({evt.current}/{evt.total}): {evt.name}")

class ZipWorker(threading.Thread):
    def __init__(self, sink, files, zip_path, password, nbits):
        super().__init__(daemon=True)
        ...
        self._cancel = threading.Event()

    def cancel(self):
        self._cancel.set()

for i, path in enumerate(self.files, 1):
    if self._cancel.is_set():
        raise InterruptedError("用户取消操作")

except InterruptedError as e:
    self._cleanup()
    wx.PostEvent(self.sink, DoneEvent(ok=False, message=str(e), ...))

def _cleanup(self):
    try:
        if os.path.exists(self.zip_path):
            os.remove(self.zip_path)
    except OSError:
        pass

except Exception:
    self._cleanup()
    wx.PostEvent(self.sink, DoneEvent(
        ok=False,
        message=traceback.format_exc(limit=2).strip().splitlines()[-1],
        elapsed=time.time() - started))

CREATE TABLE IF NOT EXISTS zip_records (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    zip_name    TEXT    NOT NULL,
    target_dir  TEXT    NOT NULL,
    zip_path    TEXT    NOT NULL,
    file_count  INTEGER NOT NULL DEFAULT 0,
    source_size INTEGER NOT NULL DEFAULT 0,
    zip_size    INTEGER NOT NULL DEFAULT 0,
    encryption  TEXT    NOT NULL,
    status      TEXT    NOT NULL,
    message     TEXT,
    elapsed     REAL,
    password    TEXT,
    created_at  TEXT    NOT NULL
);
CREATE TABLE IF NOT EXISTS zip_record_files (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    record_id INTEGER NOT NULL REFERENCES zip_records(id) ON DELETE CASCADE,
    file_path TEXT    NOT NULL,
    file_size INTEGER
);
CREATE INDEX IF NOT EXISTS idx_record_files ON zip_record_files(record_id);

def connect(self):
    conn = sqlite3.connect(self.path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn

cols = {r["name"] for r in conn.execute("PRAGMA table_info(zip_records)")}
if "password" not in cols:      # 兼容旧版本已存在的数据库
    conn.execute("ALTER TABLE zip_records ADD COLUMN password TEXT")

def connect(self):
    conn = sqlite3.connect(self.path)
    ...
    return conn

def list_records(self, keyword=""):
    ...
    with self.connect() as conn:
        return conn.execute(sql, args).fetchall()

from contextlib import closing

with closing(self.connect()) as conn:
    with conn:                      # 事务
        return conn.execute(sql, args).fetchall()

class Config:
    DEFAULTS = {
        "files": [],
        "target_dir": "",
        "zip_name": "",
        "encryption": "AES-256 (最强)",
        "remember_password": False,
        "password": "",
        "window_size": [960, 700],
    }

    def load(self):
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                saved = json.load(f)
            if isinstance(saved, dict):
                for k, v in saved.items():
                    if k in self.DEFAULTS:      # ← 关键在这一行
                        self.data[k] = v
        except (FileNotFoundError, ValueError, OSError):
            pass

self.chk_delete = wx.CheckBox(box_opt, label="压缩成功后删除原文件（移入回收站）")
self.chk_delete.SetForegroundColour(wx.Colour(0xC0, 0x39, 0x2B))

if self.chk_delete.GetValue() and wx.MessageBox(
        f"压缩成功后将把这 {len(self.files)} 个原文件移入回收站。\n"
        f"确定继续吗？", APP_NAME,
        wx.YES_NO | wx.ICON_EXCLAMATION) != wx.YES:
    return

def apply_config(self):
    self.add_files([p for p in self.cfg["files"] if os.path.isfile(p)], silent=True)

btn = wx.Button(panel, label="开始生成")
btn.SetBackgroundColour(wx.Colour(0x2D, 0x6C, 0xDF))   # ❌ Windows 上毫无效果

class FlatButton(wx.lib.buttons.GenButton):
    """扁平强调色按钮。Windows 主题下原生 wx.Button 会忽略 SetBackgroundColour，故改用自绘按钮。"""

    DISABLED = wx.Colour(0xC4, 0xC9, 0xD2)

    def __init__(self, parent, label, color, hover, size=wx.DefaultSize):
        super().__init__(parent, label=label, size=size, style=wx.BORDER_NONE)
        self._base, self._hover = color, hover
        self.SetBezelWidth(0)              # 去掉 3D 立体边，才是"扁平"
        self.SetUseFocusIndicator(False)   # 去掉焦点虚线框
        self.SetBackgroundColour(color)
        self.SetForegroundColour(wx.WHITE)
        f = self.GetFont()
        f.SetWeight(wx.FONTWEIGHT_BOLD)
        self.SetFont(f)
        self.SetCursor(wx.Cursor(wx.CURSOR_HAND))
        self.Bind(wx.EVT_ENTER_WINDOW, lambda e: self._tint(self._hover))
        self.Bind(wx.EVT_LEAVE_WINDOW, lambda e: self._tint(self._base))

    def _tint(self, colour):
        if self.IsEnabled():
            self.SetBackgroundColour(colour)
            self.Refresh()

    def Enable(self, enable=True):
        ret = super().Enable(enable)
        self.SetBackgroundColour(self._base if enable else self.DISABLED)
        self.Refresh()
        return ret

# ❌ 这会导致布局跳动
self.txt_pwd.Destroy()
self.txt_pwd = wx.TextCtrl(parent, style=0 if show else wx.TE_PASSWORD)

class PasswordCtrl(wx.Panel):
    """密码输入框。掩码/明文两个控件叠放并切换显示，避免切换时重建控件导致布局跳动。"""

    def __init__(self, parent):
        super().__init__(parent)
        self.SetBackgroundColour(parent.GetBackgroundColour())
        self.masked = wx.TextCtrl(self, style=wx.TE_PASSWORD)
        self.plain = wx.TextCtrl(self)
        self.plain.Hide()
        s = wx.BoxSizer(wx.VERTICAL)
        s.Add(self.masked, 0, wx.EXPAND)
        s.Add(self.plain, 0, wx.EXPAND)
        self.SetSizer(s)

    def GetValue(self):
        return (self.plain if self.plain.IsShown() else self.masked).GetValue()

    def SetValue(self, value):
        self.masked.SetValue(value)
        self.plain.SetValue(value)

    def show_text(self, show):
        value = self.GetValue()
        self.plain.Show(show)
        self.masked.Show(not show)
        self.SetValue(value)
        self.Layout()

def image_items(self, row):
    """收集可预览的图片，返回 (zip 句柄, [(名称, 读字节函数)], 来源说明)。

    优先用记录里的密码从 zip 内解密读取；zip 缺失或密码不可用时回退到磁盘上的原文件。
    """
    pwd = deobfuscate(row["password"])
    zip_path = row["zip_path"] or ""
    if os.path.isfile(zip_path):
        zf = None
        try:
            zf = pyzipper.AESZipFile(zip_path)
            if pwd:
                zf.setpassword(pwd.encode("utf-8"))
            names = sorted(n for n in zf.namelist() if is_image(n))
            if names:
                zf.open(names[0]).close()   # 先探一次，密码不对就走回退分支
                return zf, [(n, lambda n=n: zf.read(n)) for n in names], "ZIP 内解密"
        except Exception:
            pass
        if zf:
            zf.close()
    items = [(os.path.basename(f["file_path"]), f["file_path"])
             for f in self.db.record_files(row["id"])
             if is_image(f["file_path"]) and os.path.isfile(f["file_path"])]
    return None, [(n, lambda p=p: read_file(p)) for n, p in items], "磁盘原文件"

[(n, lambda n=n: zf.read(n)) for n in names]

def load(self):
    name, read = self.items[self.index]
    self.image, self.cache, err = None, (None, None), ""
    try:
        self.image, err = decode_image(read())     # ← 此刻才真正读取
    except Exception as e:
        err = str(e) or e.__class__.__name__

# ❌ 错误示范
fns = [lambda: zf.read(n) for n in names]
fns[0]()     # 读的是 names[-1]，不是 names[0]

>>> img = wx.Image(io.BytesIO(webp_bytes))
>>> img.IsOk()
False
>>> wx.BITMAP_TYPE_WEBP
AttributeError: module 'wx' has no attribute 'BITMAP_TYPE_WEBP'

try:                              # wxWidgets 3.2 不带 webp 解码器，用 Pillow 兜底
    from PIL import Image as PILImage, ImageOps
except ImportError:
    PILImage = None

def decode_image(data):
    """把图片字节解成 wx.Image，失败时返回 (None, 原因)。

    wx 自带解码器不认的格式（webp 最常见）交给 Pillow，顺带按 EXIF 摆正方向。
    """
    with wx.LogNull():                    # 解析失败时不弹 wx 自带的错误框
        img = wx.Image(io.BytesIO(data))
    if img.IsOk():
        return img, ""
    if PILImage is None:
        return None, "格式不支持，安装 Pillow 后可预览（pip install pillow）"
    try:
        with PILImage.open(io.BytesIO(data)) as pil:
            rgba = ImageOps.exif_transpose(pil).convert("RGBA")
        img = wx.Image(rgba.width, rgba.height)
        raw = rgba.tobytes()
        img.SetData(rgba.convert("RGB").tobytes())
        img.SetAlpha(raw[3::4])
        return img, ""
    except PILImage.UnidentifiedImageError:
        return None, "无法识别的图片格式"
    except Exception as e:
        return None, f"无法解码: {e or e.__class__.__name__}"

def on_paint(self, _):
    dc = wx.AutoBufferedPaintDC(self.canvas)
    dc.SetBackground(wx.Brush(self.BG))
    dc.Clear()
    cw, ch = self.canvas.GetClientSize()
    if not self.image or cw < 4 or ch < 4:
        return
    iw, ih = self.image.GetWidth(), self.image.GetHeight()
    scale = min(cw / iw, ch / ih, 1.0)      # 只缩不放，避免拉伸模糊
    w, h = max(1, int(iw * scale)), max(1, int(ih * scale))
    if self.cache[0] != (w, h):
        img = self.image if (w, h) == (iw, ih) else self.image.Scale(
            w, h, wx.IMAGE_QUALITY_HIGH)
        self.cache = ((w, h), img.ConvertToBitmap())
    dc.DrawBitmap(self.cache[1], (cw - w) // 2, (ch - h) // 2, True)

self.Bind(wx.EVT_CHAR_HOOK, self.on_key)

def on_key(self, evt):
    key = evt.GetKeyCode()
    if key in (wx.WXK_LEFT, wx.WXK_UP, wx.WXK_PAGEUP):
        self.step(-1)
    elif key in (wx.WXK_RIGHT, wx.WXK_DOWN, wx.WXK_PAGEDOWN, wx.WXK_SPACE):
        self.step(1)
    else:
        evt.Skip()

def step(self, delta):
    if len(self.items) > 1:
        self.index = (self.index + delta) % len(self.items)
        self.load()

def move_to_trash(paths):
    """把文件移入回收站，返回 (成功数, [(路径, 失败原因)])。

    Windows 走 SHFileOperationW + FOF_ALLOWUNDO，误删可从回收站还原；
    其他平台没有统一的回收站 API，退化为直接删除。
    """
    paths = [os.path.abspath(p) for p in paths if os.path.isfile(p)]
    if not paths:
        return 0, []
    if sys.platform != "win32":
        ok, failed = 0, []
        for p in paths:
            try:
                os.remove(p)
                ok += 1
            except OSError as e:
                failed.append((p, str(e)))
        return ok, failed

    from ctypes import wintypes

    class SHFILEOPSTRUCTW(ctypes.Structure):
        _fields_ = [
            ("hwnd", wintypes.HWND),
            ("wFunc", wintypes.UINT),
            ("pFrom", wintypes.LPCWSTR),
            ("pTo", wintypes.LPCWSTR),
            ("fFlags", ctypes.c_uint),
            ("fAnyOperationsAborted", wintypes.BOOL),
            ("hNameMappings", ctypes.c_void_p),
            ("lpszProgressTitle", wintypes.LPCWSTR),
        ]

    FO_DELETE = 3
    FOF_SILENT, FOF_NOCONFIRMATION, FOF_ALLOWUNDO, FOF_NOERRORUI = 0x04, 0x10, 0x40, 0x400

    op = SHFILEOPSTRUCTW()
    op.wFunc = FO_DELETE
    # pFrom 是双 \0 结尾的路径串
    op.pFrom = "\0".join(paths) + "\0\0"
    op.fFlags = FOF_ALLOWUNDO | FOF_NOCONFIRMATION | FOF_SILENT | FOF_NOERRORUI
    ret = ctypes.windll.shell32.SHFileOperationW(ctypes.byref(op))
    if ret != 0:
        return 0, [(p, f"SHFileOperation 错误码 {ret}") for p in paths]
    if op.fAnyOperationsAborted:
        return 0, [(p, "操作被中止") for p in paths]
    remaining = [p for p in paths if os.path.isfile(p)]
    return len(paths) - len(remaining), [(p, "仍然存在") for p in remaining]

def trash_sources(self, info):
    """把已成功压缩的原文件移入回收站，并从待压缩列表中同步移除。"""
    ok, failed = move_to_trash(info["files"])
    gone = {p for p in info["files"] if not os.path.isfile(p)}
    for i in range(len(self.files) - 1, -1, -1):
        if self.files[i] in gone:
            self.list.DeleteItem(i)
            del self.files[i]
    self.refresh_count()
    if failed:
        wx.MessageBox("以下原文件未能删除：\n" +
                      "\n".join(f"{p}\n    {why}" for p, why in failed[:5]),
                      APP_NAME, wx.ICON_WARNING)
    return f" · 已将 {ok} 个原文件移入回收站" if ok else ""

def obfuscate(text):
    """base64 混淆存储的密码。仅防止肉眼直读，不是加密。"""
    return base64.b64encode(text.encode("utf-8")).decode("ascii") if text else ""


def deobfuscate(text):
    if not text:
        return ""
    try:
        return base64.b64decode(text.encode("ascii")).decode("utf-8")
    except (ValueError, UnicodeDecodeError):
        return ""

def save(self):
    out = dict(self.data)
    if out["remember_password"] and out["password"]:
        out["password"] = obfuscate(out["password"])
    else:
        out["password"] = ""          # 不记住就清空，不留残留

with pyzipper.AESZipFile(zip_path) as zf:
    zf.setpassword(pwd.encode("utf-8"))
    bad = zf.testzip()                      # 校验所有条目的 CRC
    assert bad is None

zf, items, note = self.image_items(r)
try:
    ...
    with PreviewDialog(self, r["zip_name"], items, note) as dlg:
        dlg.ShowModal()
finally:
    if zf:
        zf.close()

# 安装依赖
pip install wxPython pyzipper pillow
# 直接运行
python zip_maker.py
# 打包（-w 不显示控制台，-F 单文件）
pyinstaller -w -F zip_maker.py

if __name__ == "__main__":
    if sys.platform == "win32":
        try:
            import ctypes
            ctypes.windll.shcore.SetProcessDpiAwareness(1)
        except Exception:
            pass
    App(False).MainLoop()