#!/usr/bin/env python3 """7z-encrypt 客户端 TUI (Python input() REPL, 中文 IME 友好) vue-tui 版输入竞争问题修复前的过渡方案 (mt-translate 同款模式: vue-tui 的 raw mode 输入无法配合 IME, Python cooked-mode input() 可以)。 流程: 直接驱动 7z-encrypt 客户端模块 CLI, 实时透传进度输出。 """ import glob import json import os import shutil import subprocess import sys from pathlib import Path CLIENT_DIR = "/home/lou/文档/7z-encrypt" PYTHON = f"{CLIENT_DIR}/.venv/bin/python" def _default_server() -> str: """从客户端 config.json 读 server.url (无则本地默认)""" try: cfg = json.loads(Path(f"{CLIENT_DIR}/config/config.json").read_text(encoding="utf-8")) return str(cfg.get("server", {}).get("url", "http://127.0.0.1:8000")) except (OSError, json.JSONDecodeError): return "http://127.0.0.1:8000" def run(args: list[str]) -> int: """子进程运行 7z-encrypt CLI, 实时透传输出 (进度可见)""" proc = subprocess.Popen( [PYTHON] + args, cwd=CLIENT_DIR, text=True, encoding="utf-8", ) try: return proc.wait() except KeyboardInterrupt: proc.kill() print("\n[已中断]") return 130 def _clean(s: str) -> str: """剥掉包裹引号 (用户复制路径常带 ' 或 ")""" return s.strip().strip("'\"") if s else "" def cleanup_artifacts(path: str) -> None: """上传成功后清理临时产物 (密文/分卷/init json/清单) 失败时不清: 断点续传依赖卷文件 + init json。 """ removed = 0 for pattern in ( path + ".enc", path + ".enc.part*", path + ".init.json", path + "_manifest.json", ): for p in glob.glob(pattern): try: os.remove(p) removed += 1 except OSError: pass if removed: print(f"[清理] 已删除 {removed} 个临时文件") def upload() -> None: print("--- 上传文件 ---") path = _clean(input("文件路径: ")) if not path or not os.path.exists(path): print(f"[错误] 文件不存在: {path}") return server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() if not server.startswith(("http://", "https://")): server = "http://" + server # 只输 IP:端口 自动补协议 chunk_size = _clean(input("卷大小 MB [10]: ")) or "10" print("\n[1/3] 7z 压缩 (-mx=9) ...") src7z = path + ".7z" r = subprocess.run( ["7z", "a", "-mx=9", "-bd", src7z, path], capture_output=True, text=True, ) if r.returncode != 0: print(f"[错误] 7z 压缩失败: {r.stderr[-300:]}") return print(f"[1/3] 压缩完成: {os.path.getsize(path) / 1048576:.1f} MB -> {os.path.getsize(src7z) / 1048576:.1f} MB") print("\n[2/3] 加密 + 分卷 + 元数据 ...") code = run([ "metadata.py", src7z, "--chunk-size", chunk_size, "--file-name", os.path.basename(path), "--compressed", ]) if code != 0: os.remove(src7z) print("[失败] 加密/分卷失败") return print("\n[3/3] 上传 ...") init_json = src7z + ".init.json" manifest = src7z + "_manifest.json" code = run(["transfer.py", "--server", server, "--init", init_json, "--manifest", manifest]) if code == 0: print("[完成]") cleanup_artifacts(src7z) # 清 .enc/.part/init/manifest os.remove(src7z) # 清压缩临时文件 else: print("[失败] 上传出错 (加密产物保留, 可续传)") os.remove(src7z) # 压缩临时文件无续传价值, 清掉 def tasks() -> None: print("--- 任务续传 ---") run(["state.py", "pending"]) def config_show() -> None: print("--- 配置 ---") run(["config.py"]) def config_edit() -> None: url = _clean(input("服务器地址: ")) if url: run(["config.py", "--set-server", url]) token = _clean(input("认证 token (留空不改): ")) if token: run(["config.py", "--set-token", token]) config_show() def files_list() -> None: """ls: 列出服务端文件 + 空间配额""" print("--- 文件列表 ---") server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() if not server.startswith(("http://", "https://")): server = "http://" + server code = run(["transfer.py", "--server", server, "--list"]) if code == 0: run(["transfer.py", "--server", server, "--quota"]) if code != 0: print("[失败] 获取列表出错") def download() -> None: """下载: 先列文件, 再选 file_id, 本地解密还原""" print("--- 下载文件 ---") server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() if not server.startswith(("http://", "https://")): server = "http://" + server code = run(["transfer.py", "--server", server, "--list"]) if code != 0: print("[失败] 获取列表出错, 无法下载") return file_id = _clean(input("file_id: ")) if not file_id: return out = _clean(input("保存路径 [当前目录/服务端文件名]: ")) args = ["transfer.py", "--server", server, "--download", file_id, "--decrypt"] if out: args += ["--out", out] code = run(args) print("[完成]" if code == 0 else "[失败] 下载出错") def delete_file() -> None: """删除: 先列文件, 再选 file_id 删除""" print("--- 删除文件 ---") server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server() if not server.startswith(("http://", "https://")): server = "http://" + server code = run(["transfer.py", "--server", server, "--list"]) if code != 0: print("[失败] 获取列表出错, 无法删除") return file_id = _clean(input("要删除的 file_id: ")) if not file_id: return code = run(["transfer.py", "--server", server, "--delete", file_id]) print("[完成]" if code == 0 else "[失败] 删除出错") def export_key() -> None: """密钥导出: 红色风险警告 + 二次确认 (防误操作)""" print("--- 密钥导出 ---") print() print("\033[91m" + "═" * 52 + "\033[0m") print("\033[91m ⚠ 危险操作警告\033[0m") print() print("\033[91m 密钥是解密你所有文件的唯一凭证!\033[0m") print("\033[91m 导出后:\033[0m") print("\033[91m · 任何人拿到此文件都能解密你服务器上的全部密文\033[0m") print("\033[91m · 密钥泄露 = 数据完全暴露, 无法撤销\033[0m") print("\033[91m · 请确保导出文件安全存放, 不要发到聊天/网盘/邮件\033[0m") print("\033[91m · 用途: 备份 / 迁移到新设备 (新设备导入后可解密)\033[0m") print("\033[91m" + "═" * 52 + "\033[0m") print() c1 = input("是否了解风险并继续? (输入 yes 继续): ").strip().lower() if c1 != "yes": print("[取消] 未导出") return c2 = input("再次确认: 输入 CONFIRM 才执行导出: ").strip() if c2 != "CONFIRM": print("[取消] 未导出") return keyring = Path(f"{CLIENT_DIR}/config/keyring.json") if not keyring.exists(): print("[错误] 密钥文件不存在, 无法导出") return out = _clean(input(f"导出到路径 [~/keyring-export.json]: ")) out = out or str(Path.home() / "keyring-export.json") shutil.copy2(keyring, out) os.chmod(out, 0o600) # 密钥文件权限 600 print(f"[完成] 密钥已导出 -> {out} (权限 600)") print("[提示] 新设备导入: 将该文件覆盖到客户端 config/keyring.json 即可") def main() -> int: print("7z-encrypt 客户端 (Python TUI)") print("安全文件传输 · 加密 / 分卷 / 上传") while True: print() print("主菜单") print(" 1. 上传文件") print(" 2. 任务续传") print(" 3. 配置") print(" 4. 文件列表 (ls)") print(" 5. 下载文件") print(" 6. 删除文件") print(" 7. 密钥导出") print(" q. 退出") choice = input("> ").strip().lower() if choice == "1": upload() elif choice == "2": tasks() elif choice == "3": config_show() edit = input("修改服务器地址? (y/N): ").strip().lower() if edit == "y": config_edit() elif choice == "4": files_list() elif choice == "5": download() elif choice == "6": delete_file() elif choice == "7": export_key() elif choice == "q": print("再见") return 0 else: print("无效输入") if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: print("\n再见") sys.exit(130)