7z-encrypt-tui/scripts/tui.py

#!/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
from urllib import request as urlrequest

CLIENT_DIR = "/home/lou/文档/7z-encrypt"
PYTHON = f"{CLIENT_DIR}/.venv/bin/python"
# 编译产物优先 (Nuitka 原生二进制), 无则回退源码解释执行
TRANSFER_BIN = f"{CLIENT_DIR}/bin/sz-transfer"
TRANSFER_CMD = [TRANSFER_BIN] if os.path.exists(TRANSFER_BIN) else [PYTHON, "transfer.py"]


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, 实时透传输出 (进度可见)

    transfer.py 走编译产物 (sz-transfer 原生二进制), 其余模块走 venv python
    """
    if args and args[0] == "transfer.py":
        cmd = TRANSFER_CMD + args[1:]
    else:
        cmd = [PYTHON] + args
    proc = subprocess.Popen(
        cmd,
        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 auth_register() -> None:
    """注册账号 (成功即登录)"""
    print("--- 注册账号 ---")
    user = _clean(input("用户名 (3-32 位字母数字或下划线): "))
    pw = input("密码 (至少 6 位): ")
    if not user or not pw:
        print("[错误] 用户名/密码不能为空")
        return
    server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server()
    if not server.startswith(("http://", "https://")):
        server = "http://" + server
    code = run(["transfer.py", "--server", server, "--register", user, pw])
    print("[完成]" if code == 0 else "[失败] 注册出错")


def auth_login() -> None:
    """登录账号"""
    print("--- 登录 ---")
    user = _clean(input("用户名: "))
    pw = input("密码: ")
    if not user or not pw:
        print("[错误] 用户名/密码不能为空")
        return
    server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server()
    if not server.startswith(("http://", "https://")):
        server = "http://" + server
    code = run(["transfer.py", "--server", server, "--login", user, pw])
    print("[完成]" if code == 0 else "[失败] 登录出错")


def auth_logout() -> None:
    """注销当前账号 (token 失效)"""
    print("--- 注销 ---")
    c = input("确认退出当前账号? (输入 yes): ").strip().lower()
    if c != "yes":
        print("[取消]")
        return
    server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server()
    if not server.startswith(("http://", "https://")):
        server = "http://" + server
    code = run(["transfer.py", "--server", server, "--logout"])
    print("[完成] 已注销" if code == 0 else "[失败] 注销出错")


def regenerate_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    · 建议先导出旧密钥备份 (菜单 7)\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
    code = run(["config.py", "--regenerate-key", "--yes"])
    print("[完成]" if code == 0 else "[失败] 重新生成出错")


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 _keyring_path() -> Path:
    """密钥库路径: config.json 显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)"""
    try:
        cfg = json.loads(Path(f"{CLIENT_DIR}/config/config.json").read_text(encoding="utf-8"))
        raw = cfg.get("crypto", {}).get("keyring_path", "")
        if raw:
            return Path(raw)
    except (OSError, json.JSONDecodeError):
        pass
    env = os.environ.get("SZ_KEYRING")
    if env:
        return Path(env)
    xdg = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share")
    return Path(xdg) / "7z-encrypt" / "keyring.json"


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 = _keyring_path()
    if not keyring.exists():
        print(f"[错误] 密钥文件不存在: {keyring}")
        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 import_key() -> None:
    """密钥导入: 红色风险警告 + 二次确认 + 自动备份现有密钥库"""
    print("--- 密钥导入 ---")
    src = _clean(input("密钥文件路径: "))
    if not src or not os.path.exists(src):
        print("[错误] 文件不存在")
        return
    try:
        with open(src, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict) or not data:
            print("[错误] 不是有效的密钥库文件 (期望 {key_id: hex})")
            return
        for kid, val in data.items():
            if not isinstance(val, str) or len(val) != 64:
                print(f"[错误] 密钥 {kid} 格式非法 (期望 32 字节 hex)")
                return
    except (OSError, ValueError) as e:
        print(f"[错误] 无法读取密钥文件: {e}")
        return

    keyring = _keyring_path()
    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    · 导入前会自动备份当前密钥 -> keyring.json.bak\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.json.bak (防手滑可回滚)
    if keyring.exists():
        shutil.copy2(keyring, str(keyring) + ".bak")
        print(f"[备份] 现有密钥已备份 -> config/keyring.json.bak")
    shutil.copy2(src, keyring)
    os.chmod(keyring, 0o600)
    print(f"[完成] 密钥已导入 (来自 {src}, 权限 600)")
    print("[提示] 导入后即可解密该密钥加密的密文; 出错可恢复 keyring.json.bak")


def _load_token() -> str:
    """从客户端 config.json 读 server.token"""
    try:
        cfg = json.loads(Path(f"{CLIENT_DIR}/config/config.json").read_text(encoding="utf-8"))
        return str(cfg.get("server", {}).get("token", ""))
    except (OSError, json.JSONDecodeError):
        return ""


def _check_login() -> str | None:
    """验证当前 token: 有效返回用户名, 否则 None (不登录不能操作文件)"""
    server = _default_server()
    token = _load_token()
    if not server or not token:
        return None
    try:
        req = urlrequest.Request(
            server.rstrip("/") + "/api/auth/me",
            headers={"Authorization": f"Bearer {token}"},
        )
        with urlrequest.urlopen(req, timeout=10) as resp:
            return str(json.loads(resp.read()).get("username", ""))
    except Exception:
        return None


def _require_login() -> bool:
    """文件操作前置校验: 未登录直接报错 (不登入不能进行文件操作)"""
    if _check_login() is not None:
        return True
    print("\033[91m[错误] 未登录, 不能进行文件操作, 请先登录 (菜单 10)\033[0m")
    return False


def main() -> int:
    print("7z-encrypt 客户端 (Python TUI)")
    print("安全文件传输 · 加密 / 分卷 / 上传")
    while True:
        user = _check_login()
        print()
        if user is None:
            # 未登录: 文件操作全部不可用 (灰掉), 只有登录/注册/配置
            print("=== 请先登录 ===")
            print(f"服务器: {_default_server() or '(未配置)'}")
            print("  1. 登录")
            print("  2. 注册")
            print("  3. 配置服务器")
            print("  q. 退出")
            c = input("> ").strip().lower()
            if c == "1":
                auth_login()
            elif c == "2":
                auth_register()
            elif c == "3":
                config_edit()
            elif c == "q":
                print("再见")
                return 0
            else:
                print("\033[91m[错误] 未登录, 只能进行 登录/注册/配置 操作\033[0m")
            continue

        # 已登录: 完整主菜单
        print(f"主菜单 (当前用户: {user})")
        print("  1. 上传文件")
        print("  2. 任务续传")
        print("  3. 配置")
        print("  4. 文件列表 (ls)")
        print("  5. 下载文件")
        print("  6. 删除文件")
        print("  7. 密钥导出")
        print("  8. 密钥导入")
        print("  9. 注册账号")
        print("  10. 登录")
        print("  11. 注销")
        print("  12. 重新生成密钥")
        print("  q. 退出")
        choice = input("> ").strip().lower()
        if choice == "1":
            if not _require_login():
                continue
            upload()
        elif choice == "2":
            if not _require_login():
                continue
            tasks()
        elif choice == "3":
            config_show()
            edit = input("修改服务器地址? (y/N): ").strip().lower()
            if edit == "y":
                config_edit()
        elif choice == "4":
            if not _require_login():
                continue
            files_list()
        elif choice == "5":
            if not _require_login():
                continue
            download()
        elif choice == "6":
            if not _require_login():
                continue
            delete_file()
        elif choice == "7":
            export_key()
        elif choice == "8":
            import_key()
        elif choice == "9":
            auth_register()
        elif choice == "10":
            auth_login()
        elif choice == "11":
            auth_logout()
            continue  # 注销后回未登录界面
        elif choice == "12":
            regenerate_key()
        elif choice == "q":
            print("再见")
            return 0
        else:
            print("[提示] 请选择 1-12 或 q")


if __name__ == "__main__":
    try:
        sys.exit(main())
    except (KeyboardInterrupt, EOFError):
        print("\n再见")
        sys.exit(0)