guanlan-src/scripts/_ops_run.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""运维控制台的动作执行器 (2026-09-12) —— 跑一条命令, 并把**真实退出码**写回 run/ops_job.json。

为什么不让网关直接 Popen 就完事: 那样只知道"进程还在不在", 进程一结束就只能猜它成功没成功 ——
实测就这么吃过一次(停止服务脚本崩在 TypeError, 页面却显示"已停组件服务")。由本执行器包一层,
它在子进程结束后把 `status/rc/finished` 落盘, 页面就能显示"完成(退出码 0)"或"失败(退出码 1)"。

日志: 本进程的 stdout/stderr 已被调用方重定向到 logs/ops_<tag>_<时间>.log, 子命令**继承**它,
所以命令自己的输出原样进日志(页面显示的就是这些尾巴)。

用法: python scripts/_ops_run.py --tag rebuild -- <命令与参数...>
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import subprocess
import sys
import threading
import time

ROOT = pathlib.Path(__file__).resolve().parents[1]
JOB = ROOT / 'run' / 'ops_job.json'
HEARTBEAT_S = 15


def _write(j: dict):
    JOB.parent.mkdir(exist_ok=True)
    JOB.write_text(json.dumps(j, ensure_ascii=False, indent=1), encoding='utf-8')


def _heartbeat(stop_evt: threading.Event):
    """每 15 s 更新 job 文件的 mtime —— 页面据此区分"还在跑"与"被强杀"(pid 会被复用, 只看 pid 会误判)。"""
    while not stop_evt.wait(HEARTBEAT_S):
        try:
            os.utime(JOB, None)
        except OSError:
            pass


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--tag', required=True)
    ap.add_argument('cmd', nargs=argparse.REMAINDER, help='-- 之后是真正的命令')
    a = ap.parse_args()
    cmd = [c for c in a.cmd if c != '--']
    if not cmd:
        print('[X] 没给命令'); return 2

    job = {}
    try:
        job = json.loads(JOB.read_text(encoding='utf-8'))
    except Exception:
        pass
    job.update(kind=a.tag, cmd=' '.join(cmd), pid=os.getpid(), status='running',
               started=job.get('started') or time.strftime('%Y-%m-%d %H:%M:%S'))
    _write(job)

    print(f'$ {" ".join([sys.executable] + cmd)}\n', flush=True)
    stop_evt = threading.Event()
    threading.Thread(target=_heartbeat, args=(stop_evt,), daemon=True).start()
    t0 = time.time()
    try:
        rc = subprocess.run([sys.executable] + cmd, cwd=str(ROOT), env=dict(os.environ)).returncode
    except Exception as e:
        print(f'[X] 命令起不来: {type(e).__name__}: {e}', flush=True)
        rc = 99
    stop_evt.set()
    dt = time.time() - t0
    job.update(status='done', rc=rc, seconds=round(dt, 1), finished=time.strftime('%Y-%m-%d %H:%M:%S'))
    _write(job)
    print(f'\n[ops] {a.tag} 结束: 退出码 {rc}, 耗时 {dt:.1f}s', flush=True)
    return rc


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