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()