guanlan-src/scripts/service_ctl.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""服务注册/卸载/起停/状态 (2026-09-17 用户令 1: 观澜安装为服务)。

     Windows : Windows 服务 (SCM)。本器用 `sc.exe` 注册, 服务体是 `scripts\win_service.py`
               (ctypes 直连 SCM, 不依赖 pywin32 —— 保持"单包离线零新依赖")。
     Linux   : systemd 单元 `guanlan.service`, 服务体是 `scripts/service_main.py` (前台守护)。

## 为什么要有 --dry-run 与前置检查

注册服务需要**管理员/root**；没有权限时最该做的是"把将要做的事说清楚"，而不是抛一句 Access is denied。
所以本器的每个动作都先过一遍前置检查（权限 / .venv / 版本记录 / 端口），
`--dry-run` 逐条打印将要执行的命令与实际会写的文件内容（systemd 单元全文），一条都不落。

## 用法

    python scripts/service_ctl.py status                # 装没装 / 在跑没在跑（不需要权限）
    python scripts/service_ctl.py install --dry-run     # 只打印: 前置检查 + 将执行的命令 + 单元文件
    python scripts/service_ctl.py install               # 真注册（Windows 需管理员, Linux 需 root）
    python scripts/service_ctl.py start|stop|restart|uninstall
    python scripts/service_ctl.py install --user-login  # Windows 备选: 不用管理员的"登录自启(Run 键)"

退出码: 0 成功 · 2 前置条件不满足(如缺 .venv) · 3 权限不足 · 4 平台不支持 · 5 服务操作失败
"""
from __future__ import annotations

import argparse
import os
import pathlib
import subprocess
import sys

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

PY = None
UNIT_PATH = pathlib.Path('/etc/systemd/system') / f'{V.SERVICE_NAME}.service'
WIN32 = os.name == 'nt'


def py_exe() -> str:
    global PY
    if PY is None:
        cand = P.venv_python()
        PY = str(cand if cand else sys.executable)
    return PY


# ── 前置条件 ─────────────────────────────────────────────────────────────────────────
def is_admin() -> bool:
    if not WIN32:
        return os.geteuid() == 0
    try:
        import ctypes
        return bool(ctypes.windll.shell32.IsUserAnAdmin())
    except Exception:
        return False


def preflight() -> list[str]:
    """→ 问题清单（空 = 可以做）。每条都写成"人能照着解决"的一句话。"""
    bad = []
    if not (ROOT / 'guanlan.py').is_file():
        bad.append(f'安装根不对：{ROOT} 下没有 guanlan.py')
    if not pathlib.Path(py_exe()).is_file():
        bad.append(f'没找到虚拟环境解释器 {py_exe()} —— 先跑 install.bat / sh install.sh')
    if WIN32 and not (ROOT / 'scripts' / 'win_service.py').is_file():
        bad.append('缺 scripts\\win_service.py（服务体）')
    if not WIN32 and not (ROOT / 'scripts' / 'service_main.py').is_file():
        bad.append('缺 scripts/service_main.py（服务体）')
    return bad


def unit_text() -> str:
    """systemd 单元全文（--dry-run 会原样打印；install 时写进 UNIT_PATH）。

    ★ 路径统一用 POSIX 写法并用引号包住（安装目录可能带空格：`/opt/guanlan app`），
      否则 systemd 会把 `ExecStart=/opt/gu anlan/.venv/bin/python` 拆成两个参数。
    """
    root = pathlib.Path(ROOT).as_posix()
    py = pathlib.Path(py_exe()).as_posix()
    return f'''[Unit]
Description={V.SERVICE_DISPLAY}
Documentation=file://{root}/docs/系统设计说明.md
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
WorkingDirectory="{root}"
ExecStart="{py}" "{root}/scripts/service_main.py"
ExecStop="{py}" "{root}/guanlan.py" stop
Restart=always
RestartSec=10
TimeoutStopSec=120
KillMode=mixed
# 组件只绑 127.0.0.1；要对外请在 nginx 侧做，不要改组件绑定
Environment=PYTHONUTF8=1
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target
'''


def win_commands(action: str) -> list[list[str]]:
    """Windows 侧要执行的 sc.exe 命令（--dry-run 原样打印，install/start/... 真跑）。"""
    binpath = f'"{py_exe()}" "{ROOT / "scripts" / "win_service.py"}"'
    n = V.SERVICE_NAME
    if action == 'install':
        return [
            ['sc.exe', 'create', n, 'binPath=', binpath, 'start=', 'auto', 'DisplayName=', V.SERVICE_DISPLAY],
            ['sc.exe', 'description', n, f'{V.NAME} v{V.VERSION} 本机离线版（观澜风电资产智能）'],
            # 服务崩溃/被结束时的恢复策略: 60 s 后重启, 连续失败继续重启, 一天后重置计数
            ['sc.exe', 'failure', n, 'reset=', '86400', 'actions=', 'restart/60000/restart/60000/restart/60000'],
            ['sc.exe', 'start', n],
        ]
    if action == 'uninstall':
        return [['sc.exe', 'stop', n], ['sc.exe', 'delete', n]]
    return [['sc.exe', action, n]]


def linux_commands(action: str) -> list[list[str]]:
    n = V.SERVICE_NAME
    if action == 'install':
        return [['systemctl', 'daemon-reload'], ['systemctl', 'enable', '--now', n],
                ['systemctl', 'status', n, '--no-pager']]
    if action == 'uninstall':
        return [['systemctl', 'disable', '--now', n], ['rm', '-f', str(UNIT_PATH)], ['systemctl', 'daemon-reload']]
    return [['systemctl', action, n]]


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


# ── 状态（不需要权限） ───────────────────────────────────────────────────────────────
def win_status() -> dict:
    r = subprocess.run(['sc.exe', 'query', V.SERVICE_NAME], capture_output=True, text=True, errors='replace')
    txt = (r.stdout or '') + (r.stderr or '')
    installed = r.returncode == 0
    running = installed and ('RUNNING' in txt.upper())
    return dict(platform='windows', installed=installed, running=running, detail=txt.strip().splitlines()[:6])


def linux_status() -> dict:
    if not UNIT_PATH.is_file():
        return dict(platform='linux', installed=False, running=False, detail=[f'没有 {UNIT_PATH}'])
    r = subprocess.run(['systemctl', 'is-active', V.SERVICE_NAME], capture_output=True, text=True, errors='replace')
    return dict(platform='linux', installed=True, running=(r.stdout or '').strip() == 'active',
                detail=[f'{UNIT_PATH} 存在', f'is-active: {(r.stdout or "").strip() or (r.stderr or "").strip()[:60]}'])


def user_login_entry(action: str, dry: bool) -> int:
    """Windows 备选（**不需要管理员**）：把启动器写进当前用户的开机自启（HKCU\\...\\Run）。

    真服务要管理员权限；很多现场账号没有。这条给一个"登录即起"的等价退路，
    并在文档里写明它与真服务的区别（服务=开机即起、无需登录、可被 SCM 恢复；本项=用户登录后起）。
    """
    import winreg  # noqa: PLC0415
    key_path = r'Software\Microsoft\Windows\CurrentVersion\Run'
    cmd = f'"{py_exe()}" "{ROOT / "scripts" / "guanlan_start_hidden.py"}"'
    if dry:
        print(f'   $ reg add HKCU\\{key_path} /v {V.SERVICE_NAME} /d "{cmd}" /f    (= {action})')
        return 0
    with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_SET_VALUE) as k:
        if action == 'install':
            winreg.SetValueEx(k, V.SERVICE_NAME, 0, winreg.REG_SZ, cmd)
            print(f'   已写入 HKCU\\{key_path}\\{V.SERVICE_NAME}')
        else:
            try:
                winreg.DeleteValue(k, V.SERVICE_NAME)
                print('   已移除登录自启项')
            except FileNotFoundError:
                print('   （本来就没有这项）')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('action', choices=('status', 'install', 'uninstall', 'start', 'stop', 'restart', 'selftest'))
    ap.add_argument('--dry-run', action='store_true', help='只打印要做的事与将执行的命令, 不落任何改动')
    ap.add_argument('--user-login', action='store_true',
                    help='Windows 备选: 不注册服务, 改成"当前用户登录时自启"(不需要管理员)')
    a = ap.parse_args()
    print(f'== {V.NAME} v{V.VERSION} 服务化 · {a.action} ({"" if WIN32 else "linux/macos"}) ==')
    print(f'   安装根: {ROOT}')
    print(f'   解释器: {py_exe()}')
    print(f'   管理员/root: {is_admin()}')

    if a.action == 'selftest':
        import service_worker as SW
        return SW.selftest()

    st = win_status() if WIN32 else linux_status()
    print(f'   当前状态: 已注册={st["installed"]} 运行中={st["running"]}')
    for ln in st['detail']:
        print(f'     {ln}')

    if a.action == 'status':
        return 0

    if a.action == 'install' and a.dry_run:
        bad = preflight()
        if bad:
            print('   前置条件不满足:')
            for b in bad:
                print(f'     - {b}')
            return 2
        print('   前置检查: 通过')
    elif a.action == 'install':
        bad = preflight()
        if bad:
            print('   [X] 前置条件不满足:')
            for b in bad:
                print(f'     - {b}')
            return 2

    if a.action in ('install', 'uninstall', 'start', 'stop', 'restart') and not is_admin() and not a.dry_run \
            and not (WIN32 and a.action == 'install' and a.user_login):
        print('   [X] 权限不足: 注册/卸载/起停服务需要管理员权限。')
        print('       ① 用管理员身份重开终端再执行本命令; 或')
        print('       ② 先看它会做什么: python scripts/service_ctl.py install --dry-run')
        if WIN32:
            print('       ③ 不想提权: python scripts/service_ctl.py install --user-login'
                  '（当前用户登录时自启, 不是真服务）')
        return 3

    if WIN32 and a.action == 'install' and a.user_login:
        return user_login_entry('install', a.dry_run)
    if WIN32 and a.action == 'uninstall' and a.user_login:
        return user_login_entry('uninstall', a.dry_run)

    if not WIN32 and a.action == 'install':
        print(f'   将写入单元文件 {UNIT_PATH}：')
        print('   ' + '-' * 66)
        for ln in unit_text().rstrip().splitlines():
            print('   | ' + ln)
        print('   ' + '-' * 66)
        if a.dry_run:
            run_cmds(linux_commands(a.action), True)
            print('   (dry-run: 以上命令未执行)')
            return 0
        UNIT_PATH.parent.mkdir(parents=True, exist_ok=True)
        UNIT_PATH.write_text(unit_text(), encoding='utf-8')
        print(f'   已写 {UNIT_PATH}')

    cmds = win_commands(a.action) if WIN32 else linux_commands(a.action)
    rc = run_cmds(cmds, a.dry_run)
    if a.dry_run:
        print('   (dry-run: 以上命令未执行；去掉 --dry-run 才真做)')
    # 装完写安装记录里的运行方式 (install-info.json 的 mode)
    if a.action == 'install' and not a.dry_run and rc == 0:
        mode = 'windows-service' if WIN32 else 'systemd'
        V.write_installed(ROOT, python=py_exe(), mode=mode)
        print(f'   已在 {V.info_path(ROOT)} 记录运行方式 = {mode}')
    return rc


if __name__ == '__main__':
    sys.exit(main())