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 re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from urllib import request as urlrequest

# 已压缩/不可压格式 (正则): 命中则跳过 7z 直传 (压不了的不浪费时间)
INCOMPRESSIBLE = re.compile(
    r"\.(7z|apk|avi|bz2|flac|gif|gz|jpeg|jpg|m4a|mkv|mov|mp3|mp4|pdf|png|rar|tar|ttf|webm|webp|woff|woff2|xz|zip)$",
    re.IGNORECASE,
)

CLIENT_DIR = os.environ.get("SZ_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 _tui_config() -> Path:
    """配置文件: 源码目录 config/config.json (开发) -> XDG (安装版)"""
    p = Path(f"{CLIENT_DIR}/config/config.json")
    if p.exists():
        return p
    xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
    return Path(xdg) / "7z-encrypt" / "config.json"


def _default_server() -> str:
    """从配置文件读 server.url (无则本地默认)"""
    try:
        cfg = json.loads(_tui_config().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",
        path + ".chunk*",  # 卷压缩模式产物 (逐卷加密文件)
    ):
        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
    if not os.path.isfile(path):
        print(f"[错误] 不是文件 (目录上传请用菜单 13 一键备份): {path}")
        return
    server = _clean(input(f"服务器 [{_default_server()}]: ")) or _default_server()
    if not server.startswith(("http://", "https://")):
        server = "http://" + server  # 只输 IP:端口 自动补协议
    dname = _clean(input("目标目录 (回车=根目录): ")).strip()
    dir_id = ""
    if dname:
        # 按名称查目录 id (没有则创建)
        try:
            from transfer import TransferClient
            cfg = json.loads(_tui_config().read_text(encoding="utf-8"))
            cli = TransferClient(cfg.get("server", {}).get("url", server),
                                 token=cfg.get("server", {}).get("token"))
            dirs = cli.list_dirs()
            hit = next((d for d in dirs if d["name"] == dname), None)
            if hit is None:
                hit = cli.create_dir(dname)
            dir_id = str(hit["dir_id"])
        except Exception as e:
            print(f"[错误] 目录处理失败: {e}")
            return

    # 全部文件统一卷模式: 32MB 钉死切卷; 可压缩格式逐卷 7z, 已压缩格式切卷+加密不 7z
    if not INCOMPRESSIBLE.search(path):
        print("\n[1/2] 卷压缩: 32MB 流式切卷 + 逐卷 7z (mx=9) + 逐卷加密 ...")
        code = run([
            "metadata.py", path, "--chunk-size", "32", "--vol-compress", "32",
            "--file-name", os.path.basename(path),
        ])
    else:
        print("[1/2] 已压缩格式: 32MB 切卷 + 逐卷加密 (不 7z)")
        code = run([
            "metadata.py", path, "--chunk-size", "32", "--vol-compress", "32",
            "--vol-no-7z", "--file-name", os.path.basename(path),
        ])
    if code != 0:
        print("[失败] 切卷/加密失败")
        return

    print("\n[2/2] 上传 ...")
    init_json = path + ".init.json"
    manifest = path + "_manifest.json"
    cmd = ["transfer.py", "--server", server, "--init", init_json, "--manifest", manifest]
    if dir_id:
        cmd += ["--dir", dir_id]
    code = run(cmd)
    if code == 0:
        print("[完成]")
        cleanup_artifacts(path)  # 清 .enc/.part/init/manifest/卷文件
    else:
        print("[失败] 上传出错 (加密产物保留, 可续传)")


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:
    """下载: 先列文件, 再选序号, 本地解密还原"""
    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
    ref = _clean(input("序号 (列表里的编号, 如 3): "))
    if not ref:
        return
    out = _clean(input("保存路径 [当前目录/服务端文件名]: "))
    args = ["transfer.py", "--server", server, "--download", ref, "--decrypt"]
    if out:
        args += ["--out", out]
    code = run(args)
    print("[完成]" if code == 0 else "[失败] 下载出错")


def delete_file() -> None:
    """删除: 先列文件, 再选序号删除"""
    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
    ref = _clean(input("要删除的序号: "))
    if not ref:
        return
    code = run(["transfer.py", "--server", server, "--delete", ref])
    print("[完成]" if code == 0 else "[失败] 删除出错")


def _keyring_path() -> Path:
    """密钥库路径: 配置显式指定用指定, 否则系统默认 (SZ_KEYRING env/XDG)"""
    try:
        cfg = json.loads(_tui_config().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:
    """从配置文件读 server.token (源码目录优先, 安装版走 XDG)"""
    try:
        cfg = json.loads(_tui_config().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("  13. 一键备份 (相册/文档/下载)")
        print("  14. 恢复备份")
        print("  15. 目录管理 (创建/列表/删除)")
        print("  16. 移动文件到目录")
        print("  17. 克隆文件")
        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 == "13":
            backup_now()
        elif choice == "14":
            restore_backup()
        elif choice == "15":
            dirs_menu()
        elif choice == "16":
            move_file_menu()
        elif choice == "17":
            clone_file_menu()
        elif choice == "q":
            print("再见")
            return 0
        else:
            print("[提示] 请选择 1-17 或 q")


def move_file_menu() -> None:
    """菜单 16: 移动文件到目录"""
    print("\n=== 移动文件 ===")
    run(["transfer.py", "--list"])
    ref = _clean(input("文件序号/ID: ")).strip()
    dst = _clean(input("目标目录 (0=根目录, 可先菜单15查看): ")).strip()
    if not ref or not dst:
        print("[失败] 序号和目标目录不能为空")
        return
    code = run(["transfer.py", "--move", ref, dst])
    print("[完成]" if code == 0 else "[失败] 移动出错")


def clone_file_menu() -> None:
    """菜单 17: 克隆文件"""
    print("\n=== 克隆文件 ===")
    run(["transfer.py", "--list"])
    ref = _clean(input("文件序号/ID: ")).strip()
    dst = _clean(input("目标目录 (0=根目录, 可先菜单15查看): ")).strip()
    if not ref or not dst:
        print("[失败] 序号和目标目录不能为空")
        return
    code = run(["transfer.py", "--clone", ref, dst])
    print("[完成]" if code == 0 else "[失败] 克隆出错")


def dirs_menu() -> None:
    """菜单 15: 目录管理 (创建/列表/删除)"""
    while True:
        print("\n=== 目录管理 ===")
        print("  1. 创建目录")
        print("  2. 目录列表")
        print("  3. 删除目录")
        print("  q. 返回主菜单")
        c = input("> ").strip().lower()
        if c == "1":
            name = _clean(input("目录名: ")).strip()
            if not name:
                print("[失败] 目录名不能为空")
                continue
            code = run(["transfer.py", "--dir-create", name])
            print("[完成]" if code == 0 else "[失败] 创建目录出错")
        elif c == "2":
            run(["transfer.py", "--dir-list"])
        elif c == "3":
            did = _clean(input("要删除的目录序号 (dir_id): ")).strip()
            if not did:
                continue
            code = run(["transfer.py", "--dir-delete", did])
            print("[完成]" if code == 0 else "[失败] 删除目录出错")
        elif c == "q":
            return


def backup_now() -> None:
    """菜单 13: 一键备份 (相册/文档/下载/聊天记录/指定目录)"""
    print("--- 一键备份 ---")
    print("  预设项: photos (相册) / documents (文档) / download (下载) / chat (聊天记录)")
    print("  或直接输入目录路径备份指定目录 (如 /home/lou/图片), 回车=全部")
    val = _clean(input("  备份项或目录 (回车=全部): ").strip())
    if not val:
        run(["backup.py"])
    elif os.path.isdir(os.path.expanduser(val)):
        run(["backup.py", "--dir", val])
    else:
        run(["backup.py"] + val.split())


def restore_backup() -> None:
    """菜单 14: 恢复备份"""
    print("--- 恢复备份 ---")
    print("  备份项: photos/documents/download/chat 或自定义目录名 (--dir 备份的)")
    name = _clean(input("  备份项: ").strip())
    if not name:
        print("[提示] 已取消")
        return
    run(["backup.py", "restore", name])


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