guanlan-src/scripts/service_worker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""服务化共用工作体 (2026-09-17 用户令 1: 观澜安装为服务)。

Windows 服务 (`scripts/win_service.py`) 与 Linux systemd (`scripts/service_main.py`) 都调这里，
**同一套起停/守护逻辑，只有外壳不同** —— 免得两平台各写一份、行为慢慢分叉。

## 它在做什么

    起   : `guanlan.py serve`（幂等：已在跑的复用，缺的补起，等 healthz）
    守护 : 每 `INTERVAL` 秒跑一次同样的 `guanlan.py serve` —— 哪个组件掉了就被拉起来；
           同时看网关 `/healthz`，连续失败则记 ERROR（服务管理器据此判活）
    停   : `guanlan.py stop`（连网关一起停；服务模式下整机停服就该全停）

为什么用**子进程 + 现成 CLI** 而不是把服务代码直接写进各组件：
服务管理器的语义是"一个前台进程代表整套系统"，而观澜本来就是 6 个独立进程；
复用 `guanlan.py` 的两条命令，起停口径与手工运维**完全一致**，不会出现"服务起的和手工起的不一样"。
"""
from __future__ import annotations

import datetime as dt
import os
import pathlib
import socket
import subprocess
import sys
import time

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

INTERVAL = 15          # 守护巡检间隔（秒）
LOG = None             # 由外壳设置为 logs/service.log


def log(msg: str, level: str = 'INFO') -> None:
    """服务日志 —— 与 §11.2 同一行格式，落 `logs/<组件>.log` 的 service 一份。"""
    from src import logfile as _lf
    line = _lf.line(level, 'service', msg)
    print(line, flush=True)
    try:
        with open(_lf.component_log('service'), 'a', encoding='utf-8', newline='\n') as f:
            f.write(line + '\n')
    except Exception:
        pass


def _py() -> str:
    return str(P.venv_python() or sys.executable)


def run(cmd: list, timeout: int = 300) -> int:
    """跑一条 guanlan.py 子命令（捕获输出写进服务日志）。"""
    p = subprocess.run([_py(), 'guanlan.py'] + cmd, cwd=str(ROOT), capture_output=True, text=True,
                       errors='replace', timeout=timeout)
    for ln in (p.stdout or '').splitlines():
        log('  ' + ln)
    for ln in (p.stderr or '').splitlines():
        log('  ' + ln, 'WARN')
    return p.returncode


def gateway_up(host: str = '127.0.0.1') -> bool:
    try:
        with socket.create_connection((host, 28084), timeout=1.5):
            return True
    except OSError:
        return False


def start() -> None:
    log(f'启动 {V.NAME} v{V.VERSION}（{V.EDITION}）')
    rc = run(['serve'])
    log(f'起组件完成 (rc={rc}; 1 = 有模块未就绪, 属正常)')


def ensure() -> bool:
    """一次巡检：把缺的组件补起来。→ 网关是否在听。"""
    run(['serve'])
    return gateway_up()


def stop() -> None:
    log('停止全部组件（含网关）')
    rc = run(['stop'])
    log(f'停止完成 (rc={rc})')


def raw_watch_cfg() -> dict:
    """输入数据看门狗配置（`configs/serve.json` 的 `raw_watch`）。**默认关** —— 自动重算会占机器约一小时，
    该由现场决定开不开。读不到/写坏了按"关"处理（不因为配置问题去动数据）。"""
    d = dict(enabled=False, minutes=30, auto_rebuild=False)
    try:
        import json
        c = json.loads((ROOT / 'configs' / 'serve.json').read_text(encoding='utf-8-sig'))
        w = c.get('raw_watch') or {}
        if isinstance(w, dict):
            d.update({k: w[k] for k in ('enabled', 'minutes', 'auto_rebuild') if k in w})
    except Exception:
        pass
    return d


_watch = {'last': 0.0}


def raw_watch(stop_evt) -> None:
    """输入数据看门狗（用户令 2026-09-19「增加自动扫描识别机制，以能发现新增数据，并纳入重算」）。

    每 `minutes` 分钟扫一次 `data/raw`：**发现新增/变化就在服务日志里写明"该跑哪几步"**；
    `auto_rebuild=true` 时顺手发起一次重算（走 `_ops_run.py`，与运维控制台「执行重算」同一条路）。
    ★两道保险：① 重算已在跑（`run/ops_job.json` status=running）就不再发起；② 本函数自己吞异常，
    绝不因为看门狗把组件守护循环带崩。
    """
    c = raw_watch_cfg()
    if not c['enabled']:
        return
    now = time.time()
    if now - _watch['last'] < max(1, int(c['minutes'])) * 60:
        return
    _watch['last'] = now
    try:
        env = dict(os.environ, PYTHONUTF8='1', PYTHONIOENCODING='utf-8')
        env['WINDSCADA_ROOT'] = str(ROOT)
        p = subprocess.run([_py(), 'scripts/raw_scan.py', '--json', '--check'], cwd=str(ROOT), env=env,
                           capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=1800)
        if p.returncode == 5:
            log('输入数据扫描: 还没有基线快照 —— 记一份（下次起就能比出新增）')
            subprocess.run([_py(), 'scripts/raw_scan.py', '--write'], cwd=str(ROOT), env=env,
                           capture_output=True, timeout=1800)
            return
        if p.returncode != 4:
            return
        import json as _json
        d = _json.loads(p.stdout or '{}')
        ch = d.get('changes') or []
        log(f'输入数据有 {len(ch)} 处新增/变化: ' + '；'.join(f'{x["family"]}({x["kind"]})' for x in ch[:6]))
        if d.get('steps'):
            log('  该跑: ' + ' / '.join(d['steps']))
        job = ROOT / 'run' / 'ops_job.json'
        running = False
        try:
            running = (_json.loads(job.read_text(encoding='utf-8')).get('status') == 'running')
        except Exception:
            pass
        if not c['auto_rebuild']:
            log('  （raw_watch.auto_rebuild=false：只报不跑；要自动重算把它设 true）')
            return
        if running:
            log('  已有重算在跑 —— 本次不重复发起（新数据会在下一轮或那次重算里被扫到）')
            return
        log('  按配置发起一次重算（raw_watch.auto_rebuild=true）')
        subprocess.Popen([_py(), 'scripts/_ops_run.py', '--tag', 'rawwatch', '--', 'scripts/rebuild_all.py',
                          '--auto'], cwd=str(ROOT), env=env, stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL)
    except Exception as e:
        log(f'输入数据看门狗异常: {type(e).__name__}: {e}', 'WARN')


def supervise(stop_evt, once: bool = False) -> None:
    """守护循环：直到 stop_evt 被置位（或 once=True 只跑一轮）。"""
    fails = 0
    while not stop_evt.is_set():
        try:
            ok = ensure()
            if not ok:
                fails += 1
                log(f'网关未就绪（连续 {fails} 次）', 'ERROR' if fails >= 3 else 'WARN')
            else:
                if fails:
                    log('网关已恢复')
                fails = 0
            raw_watch(stop_evt)               # 输入数据看门狗（默认关; 见 raw_watch_cfg）
        except Exception as e:
            log(f'巡检异常: {type(e).__name__}: {e}', 'ERROR')
        if once:
            return
        stop_evt.wait(INTERVAL)


def selftest() -> int:
    """不需要任何权限的自证：依赖、端口、命令行是否齐备（服务注册前先看这个）。"""
    ok = True
    log(f'{V.NAME} v{V.VERSION} 服务化自检 @ {ROOT}')
    py = _py()
    log(f'  解释器: {py}  (存在: {pathlib.Path(py).is_file()})')
    ok &= pathlib.Path(py).is_file()
    log(f'  guanlan.py: {(ROOT / "guanlan.py").is_file()}')
    ok &= (ROOT / 'guanlan.py').is_file()
    log(f'  网关端口 28084 当前: ' + ('在听' if gateway_up() else '未监听'))
    log(f'  版本记录: {V.info_path(ROOT)}' + ('（已存在）' if V.info_path(ROOT).is_file() else '（无 —— 全新安装）'))
    # 组件端口占用检查（服务接管前这些端口应为空闲或已由本系统占用）
    for name, port in (('detail', 18033), ('cms', 18020), ('sim', 18791), ('sim_sys', 18792), ('viewer', 64292)):
        with socket.socket() as s:
            s.settimeout(0.3)
            busy = s.connect_ex(('127.0.0.1', port)) == 0
        log(f'  端口 {port} ({name}): ' + ('在听（多半是本系统已起）' if busy else '空闲'))
    log('自检结论: ' + ('服务注册前置条件齐备' if ok else '有缺项，见上'), 'INFO' if ok else 'ERROR')
    return 0 if ok else 1