guanlan-src/scripts/guanlan_uninstall.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""观澜卸载器 (2026-09-17 用户令: 增加卸载脚本 `uninstall.bat` / `uninstall.sh`)。

入口就是这两个壳脚本 + 本文件；三处共用**同一份**实现（与 install.bat/install.sh→install.ps1 同构），
免得"Windows 一套、Linux 一套"各写一遍逻辑、过两个月行为不一致。

## 卸载拆成哪几步（默认口径：只拆"本程序装上去的东西"，绝不动用户数据）

    ① 停服务      先停服务化常驻体（Windows 服务 / systemd 单元），再停手工启动的那套
    ② 注销服务    `sc.exe delete guanlan` / `systemctl disable --now` + 删单元文件（需管理员/root）
    ③ 撤开机自启  HKCU\...\Run 的 guanlan 项 / systemd 用户单元（不需要管理员）
    ④ 删快捷方式  `启动观澜.lnk`、`start_guanlan.lnk`（安装根 + 桌面 + applications）
    ⑤ 清运行态    `run/pids.json`、残留 `*.pid`
    ⑥ 删安装记录  `install-info.json`（"已装的是哪一版"的唯一凭据 —— 删掉它 = 这台机器回到"没装过"）

    **保留**：`data/`（输入数据）· `outputs/`（产物）· `logs/`（日志）· `.venv/`（虚拟环境）· 整棵程序目录

## 为什么默认既不删数据、也不删程序目录

· `data/` 与 `outputs/` 是**现场资产**（原始件十几 GB + 重算产物）；"卸载程序"不等于"销毁数据"。
  要连数据一起清，显式给 `--purge`（并且**再问一次**，默认答否；`--keep-logs` 可留下日志）。
· 程序目录**不自己删**：Windows 上正在跑的 `python.exe` 就在 `.venv` 里，删自己必然部分失败，
  留下"删了一半的目录"比留着整个目录更难排查。要清干净给 `--all`：本器尽力删，**删不掉的逐条列出**，
  并打印一条收尾命令（`rmdir /s /q <根>` 或 `rm -rf <根>`）让你关掉本窗口后执行。

## 用法

    uninstall.bat                    交互：先打印计划 → 问一次 → 执行
    uninstall.bat --dry-run          只打印计划，一个字节都不动（★自证用）
    uninstall.bat --yes              不问（脚本/无人值守）
    uninstall.bat --purge --yes      连 data/ outputs/ logs/ .venv/ 一起删
    uninstall.bat --all --yes        连程序文件一起删（最后给收尾命令）
    sh uninstall.sh [同样的开关]      Linux / macOS

    退出码: 0 完成 · 2 需要确认(非交互又没给 --yes) · 3 权限不足(②需管理员/root, 其余照做) · 5 有步骤失败
"""
from __future__ import annotations

import argparse
import os
import pathlib
import shutil
import subprocess
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from src import version as V                                            # noqa: E402

WIN32 = os.name == 'nt'
UNIT_PATH = pathlib.Path('/etc/systemd/system') / f'{V.SERVICE_NAME}.service'
USER_UNIT = pathlib.Path.home() / '.config' / 'systemd' / 'user' / f'{V.SERVICE_NAME}.service'
AUTOSTART = pathlib.Path.home() / '.config' / 'autostart' / f'{V.SERVICE_NAME}.desktop'
SHORTCUTS = ('启动观澜.lnk', 'start_guanlan.lnk')

# 卸载时要清掉的可重建件 / 用户数据（--purge 才动）
PURGE_DIRS = ('data', 'outputs', 'logs', 'run', 'runtime')
# --all 才动的"程序本体"
PROGRAM_DIRS = ('src', 'scripts', 'configs', 'release', 'resources', 'reference', 'docs', 'wheels', 'vendor')
PROGRAM_FILES = ('guanlan.py', 'requirements.txt', 'README_先读我.MD', '测试须知.txt',
                 'install.bat', 'install.ps1', 'install.sh', 'uninstall.bat', 'uninstall.sh',
                 'check.bat', 'start.bat', 'stop.bat', 'pack.bat', 'pack.sh', 'dist-manifest.json')

RESULTS: list[tuple[str, int, str]] = []        # (步骤, rc, 说明) —— 结尾统一汇总


def say(msg: str = '') -> None:
    print(msg)


def py_exe() -> str:
    """优先用本安装的 .venv 解释器；它没了（比如已经被删）就退回当前解释器。"""
    cand = ROOT / '.venv' / ('Scripts/python.exe' if WIN32 else 'bin/python')
    return str(cand) if cand.is_file() else sys.executable


def is_admin() -> bool:
    if not WIN32:
        try:
            return os.geteuid() == 0
        except AttributeError:
            return False
    try:
        import ctypes
        return bool(ctypes.windll.shell32.IsUserAnAdmin())
    except Exception:
        return False


def run(cmd: list[str], dry: bool, note: str = '') -> int:
    printable = ' '.join(f'"{x}"' if ' ' in x else x for x in cmd)
    say(f'   $ {printable}' + (f'   ({note})' if note else ''))
    if dry:
        return 0
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, errors='replace')
    except FileNotFoundError:
        say(f'   [X] 找不到命令 {cmd[0]}')
        return 5
    for ln in (r.stdout or '').strip().splitlines():
        say('     ' + ln)
    for ln in (r.stderr or '').strip().splitlines():
        say('     [stderr] ' + ln)
    return 0 if r.returncode == 0 else 5


def rm(path: pathlib.Path, dry: bool) -> int:
    """真删（目录递归 / 文件单件）；不存在算成功（卸载必须幂等，跑第二遍不该报错）。"""
    if not path.exists():
        say(f'   - 本来就没有: {rel(path)}')
        return 0
    size, n = size_of(path)
    say(f'   $ 删除 {rel(path)}' + (f'   ({n} 件, {size / 1e6:.1f} MB)' if path.is_dir() else f'   ({size} B)'))
    if dry:
        return 0
    try:
        if path.is_dir():
            shutil.rmtree(path)
        else:
            path.unlink()
        return 0
    except OSError as e:
        say(f'   [X] 删不掉: {e}')
        return 5


def rel(p: pathlib.Path) -> str:
    try:
        return p.relative_to(ROOT).as_posix() or '.'
    except ValueError:
        return str(p)


def size_of(p: pathlib.Path) -> tuple[int, int]:
    """→ (字节数, 件数)。"""
    if p.is_file():
        try:
            return p.stat().st_size, 1
        except OSError:
            return 0, 0
    tot = n = 0
    for dp, _dn, fns in os.walk(p):
        for fn in fns:
            try:
                tot += (pathlib.Path(dp) / fn).stat().st_size
                n += 1
            except OSError:
                pass
    return tot, n


# ── 各步 ────────────────────────────────────────────────────────────────────────────
def service_registered() -> bool:
    if WIN32:
        try:
            r = subprocess.run(['sc.exe', 'query', V.SERVICE_NAME], capture_output=True, text=True, errors='replace')
            return r.returncode == 0
        except FileNotFoundError:
            return False
    return UNIT_PATH.is_file()


def step_stop(dry: bool) -> None:
    say('① 停服务')
    rc = 0
    note = []
    if service_registered():
        rc = run([py_exe(), str(ROOT / 'scripts' / 'service_ctl.py'), 'stop'], dry, '停服务化常驻体')
        note.append('服务化常驻体已停' if rc == 0 else '常驻体那步没成功(多为权限)')
    else:
        say(f'   - 没注册过 {V.SERVICE_NAME} 服务（跳过常驻体）')
        note.append('无服务化常驻体')
    r2 = run([py_exe(), str(ROOT / 'guanlan.py'), 'stop'], dry, '停启动器起的组件')
    rc = rc or r2
    note.append('启动器组件已停' if r2 == 0 else '启动器组件停止失败')
    RESULTS.append(('停止服务', rc, '；'.join(note)))


def step_unregister(dry: bool) -> None:
    say('② 注销服务（需管理员 / root）')
    if not service_registered():
        say(f'   - 本机没注册过 {V.SERVICE_NAME} 服务（跳过）')
        RESULTS.append(('注销服务', 0, '本来就没注册'))
        return
    if not is_admin() and not dry:
        say('   [!] 权限不足：注销服务需要管理员/root —— 其余步骤照做，这一步留给你手工执行：')
        if WIN32:
            say(f'       sc.exe stop {V.SERVICE_NAME} && sc.exe delete {V.SERVICE_NAME}')
        else:
            say(f'       sudo systemctl disable --now {V.SERVICE_NAME} && sudo rm -f {UNIT_PATH}')
        RESULTS.append(('注销服务', 3, '权限不足（已给出手工命令）'))
        return
    rc = run([py_exe(), str(ROOT / 'scripts' / 'service_ctl.py'), 'uninstall'], dry, '停 + 删注册')
    RESULTS.append(('注销服务', rc, '服务已注销' if rc == 0 else '注销失败'))


def step_autostart(dry: bool) -> None:
    say('③ 撤开机自启（不需要管理员）')
    rc = 0
    if WIN32:
        key = r'Software\Microsoft\Windows\CurrentVersion\Run'
        say(f'   $ reg delete HKCU\\{key} /v {V.SERVICE_NAME} /f')
        if not dry:
            try:
                import winreg
                with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key, 0, winreg.KEY_SET_VALUE) as k:
                    try:
                        winreg.DeleteValue(k, V.SERVICE_NAME)
                        say('   已移除登录自启项')
                    except FileNotFoundError:
                        say('   - 本来就没有这项')
            except Exception as e:                                     # 注册表异常不该中断整场卸载
                say(f'   [X] {type(e).__name__}: {e}')
                rc = 5
    elif AUTOSTART.exists() or USER_UNIT.exists():
        rc = run(['systemctl', '--user', 'disable', '--now', V.SERVICE_NAME], dry, '用户级自启(可能没装)')
        rc = rc or rm(AUTOSTART, dry)
        rc = rc or rm(USER_UNIT, dry)
    else:
        say('   - 没有用户级自启项（跳过）')
    RESULTS.append(('撤开机自启', rc, '登录/开机自启项已清' if rc == 0 else '清理失败'))


def step_shortcuts(dry: bool) -> None:
    say('④ 删快捷方式（安装根 + 桌面 + applications）')
    targets: list[pathlib.Path] = [ROOT / n for n in SHORTCUTS]
    if WIN32:
        up = os.environ.get('USERPROFILE')
        if up:
            targets += [pathlib.Path(up) / d / n for d in ('Desktop', '桌面') for n in SHORTCUTS]
    else:
        targets += [pathlib.Path.home() / d / f'{n[:-4]}.desktop'
                    for d in ('Desktop', '.local/share/applications') for n in SHORTCUTS]
    hit = [p for p in targets if p.exists()]
    rc = 0
    for p in hit:
        rc = rc or rm(p, dry)
    if not hit:
        say('   - 没有安装时建的快捷方式（跳过）')
    RESULTS.append(('删快捷方式', rc, f'删了 {len(hit)} 个' if hit else '本来就没有'))


def step_runtime(dry: bool) -> None:
    say('⑤ 清运行态（run/pids.json、残留 *.pid）')
    rc = rm(ROOT / 'run', dry)
    for p in sorted(ROOT.glob('*.pid')):
        rc = rc or rm(p, dry)
    RESULTS.append(('清运行态', rc, 'PID 记录已清' if rc == 0 else '清理失败'))


def step_install_record(dry: bool) -> None:
    say('⑥ 删安装记录 install-info.json')
    ins = V.read_installed(ROOT)
    if ins:
        say(f"   - 记录的是: {ins.get('name', '?')} v{ins.get('version', '?')}"
            f"（装于 {ins.get('installed_at', '?')}，运行方式 {ins.get('mode', '?')}）")
    else:
        say('   - 没有安装记录（本来就没装过 / 已被删）')
    rc = rm(V.info_path(ROOT), dry)
    RESULTS.append(('删安装记录', rc, '这台机器回到"没装过"状态' if rc == 0 else '删除失败'))


def step_purge(dry: bool, yes: bool, skip: tuple[str, ...] = ()) -> None:
    say('⑦ --purge: 删数据与可重建件（输入数据 / 产物 / 日志 / 虚拟环境）')
    dirs = tuple(d for d in PURGE_DIRS if d not in skip)
    targets = [ROOT / d for d in dirs] + [ROOT / '.venv']
    present = [p for p in targets if p.exists()]
    if skip:
        say(f'   (--keep-logs: 保留 {", ".join(skip)})')
    if not present:
        say('   - 没什么可清的（都不在）')
        RESULTS.append(('--purge 清数据', 0, '无对象'))
        return
    tot = sum(size_of(p)[0] for p in present)
    say(f'   将删除: {", ".join(rel(p) for p in present)}  —— 合计 {tot / 1e9:.2f} GB')
    if not yes:
        if not sys.stdin or not sys.stdin.isatty():
            say('   [!] 非交互环境：--purge 需要显式 --yes 确认 —— 本步跳过（数据原样保留）')
            RESULTS.append(('--purge 清数据', 2, '非交互未确认，已跳过'))
            return
        ans = input('   这会永久删除现场数据与产物，确认? 输入 yes 继续，其它键跳过: ').strip().lower()
        if ans not in ('yes', 'y'):
            say('   - 已跳过（数据原样保留）')
            RESULTS.append(('--purge 清数据', 0, '用户选择跳过'))
            return
    rc = 0
    for p in present:
        rc = rc or rm(p, dry)
    RESULTS.append(('--purge 清数据', rc, f'删除 {len(present)} 项' if rc == 0 else '有删不掉的'))


def step_all(dry: bool) -> None:
    say('⑧ --all: 删程序本体（尽力而为，删不掉的逐条列出）')
    # ★ 先把自己的工作目录移出安装根：Windows 不允许删除"当前目录"，也不允许删正在跑的 python.exe
    if not dry:
        try:
            os.chdir(os.environ.get('TEMP') or os.environ.get('TMP') or '/tmp')
        except OSError:
            pass
    failed: list[str] = []
    for name in PROGRAM_DIRS + PROGRAM_FILES:
        p = ROOT / name
        if p.exists() and rm(p, dry) != 0:
            failed.append(rel(p))
    if not dry and ROOT.exists():
        for leftover in sorted(ROOT.iterdir()):                        # 兜底：清单之外的零碎也清
            if rm(leftover, dry) != 0:
                failed.append(rel(leftover))
        try:
            ROOT.rmdir()
            say(f'   已删除安装根 {ROOT}')
        except OSError as e:
            say(f'   [!] 安装根没删掉（{e.__class__.__name__}）—— 正常：本程序正在跑, 自己的目录被占用')
            say('       关掉本窗口后手工执行: '
                + (f'rmdir /s /q "{ROOT}"' if WIN32 else f'rm -rf "{ROOT}"'))
    if failed:
        say(f'   删不掉的 {len(failed)} 项: {", ".join(failed[:8])}' + (' …' if len(failed) > 8 else ''))
    RESULTS.append(('--all 删程序', 0 if not failed else 5,
                    '程序文件已清' if not failed else f'{len(failed)} 项留待手工删'))


# ── 计划与自检 ──────────────────────────────────────────────────────────────────────
def plan_lines() -> list[str]:
    """把"将要做什么"列出来（--dry-run 打印它；也是自检的断言对象）。"""
    ins = V.read_installed(ROOT)
    return [
        f'观澜卸载计划 @ {ROOT}',
        f'  本包版本: {V.NAME} v{V.VERSION}'
        + (f"；安装记录写的: v{ins.get('version', '?')}（装于 {ins.get('installed_at', '?')}）"
           if ins else '；没有安装记录（本来就没装过 / 已被删）'),
        '  ① 停服务: service_ctl stop → guanlan.py stop',
        '  ② 注销服务: '
        + (f'sc.exe stop/delete {V.SERVICE_NAME}' if WIN32 else f'systemctl disable --now {V.SERVICE_NAME} + 删 {UNIT_PATH}')
        + f'  (已注册={service_registered()}, 有权限={is_admin()})',
        '  ③ 撤开机自启: ' + (r'HKCU\Software\Microsoft\Windows\CurrentVersion\Run' if WIN32 else str(USER_UNIT)),
        '  ④ 删快捷方式: ' + ' / '.join(SHORTCUTS) + '（安装根 + 桌面 + applications）',
        '  ⑤ 清运行态: run/ 与 *.pid',
        f'  ⑥ 删安装记录: {V.INSTALL_INFO}',
        f'  ⑦/⑧ 只有给了 --purge / --all 才动数据与程序本体',
        '  保留(默认): ' + ' '.join(f'{d}/' + ('' if (ROOT / d).exists() else '(没有)')
                                    for d in ('data', 'outputs', 'logs', '.venv')) + ' 与整棵程序目录',
    ]


def selftest() -> int:
    r"""自检（--selftest）：证明"计划齐全"且"dry-run 一个字节都不动"。

    一个是**内容断言**（六个步骤一个不缺），一个是**副作用断言**（跑完 --dry-run 后关键路径的
    存在性与大小完全不变）—— 后者才是"不会误删"的证据。
    """
    ok = True
    need = ('① 停服务', '② 注销服务', '③ 撤开机自启', '④ 删快捷方式', '⑤ 清运行态', '⑥ 删安装记录', '保留')
    txt = '\n'.join(plan_lines())
    miss = [k for k in need if k not in txt]
    print(f'  [{"OK" if not miss else "FAIL"}] 计划齐全（六个步骤 + 保留清单）' + ('' if not miss else f' — 缺 {miss}'))
    ok &= not miss
    watch = ['install-info.json', 'run', 'data', 'outputs', 'logs', '.venv']
    before = {w: (size_of(ROOT / w) if (ROOT / w).exists() else None) for w in watch}
    step_stop(True)
    step_unregister(True)
    step_autostart(True)
    step_shortcuts(True)
    step_runtime(True)
    step_install_record(True)
    after = {w: (size_of(ROOT / w) if (ROOT / w).exists() else None) for w in watch}
    same = before == after
    print(f'  [{"OK" if same else "FAIL"}] dry-run 无副作用（{len(watch)} 个关键路径的存在性与大小都没变）')
    ok &= same
    print('  [OK] 六步全部构建成功: ' + '; '.join(f'{n}={rc}' for n, rc, _ in RESULTS))
    RESULTS.clear()
    print('  结论: ' + ('自检通过' if ok else '自检未通过'))
    return 0 if ok else 1


def main() -> int:
    ap = argparse.ArgumentParser(description=f'{V.NAME} v{V.VERSION} 卸载器')
    ap.add_argument('--dry-run', action='store_true', help='只打印计划, 一个字节都不动')
    ap.add_argument('--yes', '-y', action='store_true', help='不问, 直接执行（脚本/无人值守用）')
    ap.add_argument('--purge', action='store_true', help='连 data/ outputs/ logs/ .venv/ 一起删（会再问一次）')
    ap.add_argument('--all', action='store_true', help='连程序本体一起删（尽力而为, 最后给收尾命令）')
    ap.add_argument('--keep-logs', action='store_true', help='--purge 时保留 logs/（排查用）')
    ap.add_argument('--selftest', action='store_true', help='自检: 计划齐全 + dry-run 无副作用')
    a = ap.parse_args()

    say(f'== {V.NAME} v{V.VERSION} 卸载 · {ROOT} ==')
    if a.selftest:
        return selftest()
    for ln in plan_lines():
        say(ln)
    say('')

    if not a.dry_run:
        if not a.yes:
            if not sys.stdin or not sys.stdin.isatty():
                say('   [!] 非交互环境：卸载需要确认。请显式加 --yes（或先在终端里手工跑一次看计划）')
                say('       本次未做任何改动。')
                return 2
            ans = input('   确认卸载? 输入 y 继续, 其它键退出: ').strip().lower()
            if ans not in ('y', 'yes'):
                say('   已取消（未改动任何文件）')
                return 0
        say('')

    step_stop(a.dry_run)
    step_unregister(a.dry_run)
    step_autostart(a.dry_run)
    step_shortcuts(a.dry_run)
    step_runtime(a.dry_run)
    step_install_record(a.dry_run)
    if a.purge:
        step_purge(a.dry_run, a.yes, skip=('logs',) if a.keep_logs else ())
    if a.all:
        step_all(a.dry_run)

    say('')
    say('== 结果 ==')
    for name, rc, note in RESULTS:
        say(f'   [{"OK" if rc == 0 else ("!" if rc == 3 else "X")}] {name}: {note}')
    worst = max((rc for _n, rc, _s in RESULTS), default=0)
    if a.dry_run:
        say('   (dry-run: 以上一个字节都没动 — 去掉 --dry-run 才真做)')
        return 0
    say('   保留: ' + ('（--all 已连程序本体一起处理）' if a.all else 'data/ outputs/ logs/ .venv/ 与整棵程序目录'))
    if worst == 0:
        say('   卸载完成。')
        if not a.all:
            say('   想连目录一起清掉: ' + (f'rmdir /s /q "{ROOT}"' if WIN32 else f'rm -rf "{ROOT}"'))
    elif worst == 3:
        say('   卸载基本完成，但**服务注销那一步缺权限** —— 用管理员/root 重跑一次即可（命令见上）。')
    else:
        say('   有步骤失败，见上面的 [X]。')
    # 机器审计流水（卸载也要留痕；logs/ 被 --purge 删掉时这一步自然就没了，属预期）
    try:
        from src import logfile as LF
        LF.append_jsonl('uninstall', dict(root=str(ROOT), version=V.VERSION, purge=bool(a.purge),
                                          all=bool(a.all), worst_rc=worst,
                                          steps=[dict(name=n, rc=rc, note=s) for n, rc, s in RESULTS]))
    except Exception:
        pass
    return 0 if worst == 0 else worst


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())