#!/usr/bin/env python3 # -*- coding: utf-8 -*- """观澜 · 如东样板 v2 启动器 (跨平台, 只用标准库). 用法: python guanlan.py check 自检: Python/依赖/端口/数据产物/发布件/Ollama python guanlan.py serve 启动全部服务 (网关 + 工作台 + CMS + 仿真×2 + 三维), 写 run/pids.json, 等 healthz python guanlan.py status 各服务端口/进程状态 python guanlan.py stop 停止本启动器起的全部服务 python guanlan.py open 打印入口地址 (并尝试用系统浏览器打开) python guanlan.py qa 离线模型已知答案回放 (需服务已起; 参数见 scripts/llm_known_answer.py) 端口与路径在 configs/serve.json (没有则用内置默认). 所有服务只绑 127.0.0.1.""" from __future__ import annotations import json, os, socket, subprocess, sys, time, urllib.request, warnings, webbrowser, signal from pathlib import Path ROOT = Path(__file__).resolve().parent; RUN = ROOT / "run"; LOGS = ROOT / "logs"; PIDS = RUN / "pids.json" sys.path.insert(0, str(ROOT)) from src import paths as P # 配置/路径唯一取用口 (P.SERVE_JSON / P.config()…, 见 docs §11) WIN = os.name == "nt" def _hermetic(): """摘掉机器全局的 PYTHONPATH: 本项目依赖都装在 .venv 里, 而 PYTHONPATH 会被插到 sys.path 前面, 顶掉 venv 里 pin 住的版本 (实测 requests 2.33.0 顶掉 2.34.2), 还会让 pip 少装传递依赖。 子进程从 env() 继承的是摘干净之后的环境。""" for d in [x for x in os.environ.pop("PYTHONPATH", "").split(os.pathsep) if x]: while d in sys.path: sys.path.remove(d) _hermetic() DEFAULT = { "host": "127.0.0.1", "gateway": 28084, "detail": 18033, "cms": 18020, "sim": 18791, "sim_sys": 18792, "viewer": 64292, "ollama": 11434, "release_dir": "release", "viewer_dir": "release/viewer", "sim_dir": "resources/oem_envision_sc1_rudong2014", "python": "", # 空 = 当前解释器 (安装脚本建的 .venv) } def jload(p: Path): """读 JSON 配置。必须用 utf-8-sig: 记事本和 PowerShell 5.1 的 `Set-Content -Encoding UTF8` 写出的都是 **带 BOM** 的 UTF-8, 裸 utf-8 读会 JSONDecodeError (Unexpected UTF-8 BOM)。""" return json.loads(Path(p).read_text(encoding="utf-8-sig")) def cfg(): c = dict(DEFAULT); p = P.SERVE_JSON # 端口/路径真源 (配置取用口 = src/paths.py) if p.exists(): try: c.update(jload(p)) except Exception as ex: # 配置写坏不该让整个产品起不来: 退回内置默认端口, 但要说清楚 print(f" [X] {p} 读不了 ({ex.__class__.__name__}: {str(ex)[:100]}); 本次用内置默认端口。" f" 重跑 install 脚本可重写该文件") return c def py(c): """解释器: serve.json 的 python (相对路径按**安装根**解析) → 本目录 .venv (按平台探测) → 当前解释器。 2026-09-11: 原来只认绝对路径 —— 安装目录整个拷到别的电脑/别的盘后, 那个绝对路径就失效了。""" v = (c.get("python") or "").strip() if v: p = Path(v) p = p if p.is_absolute() else (ROOT / p) if p.exists(): return str(p) for rel in (".venv/Scripts/python.exe", ".venv/bin/python", ".venv/bin/python3"): p = ROOT / rel if p.exists(): return str(p) return sys.executable def up(host, port, t=0.4): try: with socket.create_connection((host, port), timeout=t): return True except OSError: return False def env(c): e = dict(os.environ); rel = str(ROOT / c["release_dir"]) raw = Path(c.get("raw_dir", "data/raw")); raw = raw if raw.is_absolute() else (ROOT / raw) e.update(GUANLAN_RELEASE_DIR=rel, GUANLAN_PORTAL=str(ROOT / c["release_dir"] / "portal.html"), GUANLAN_STATIC=rel, WINDSCADA_ROOT=str(ROOT), WINDCMS_ROOT=str(ROOT), WINDSCADA_RUDONG_SRC=str(raw), GUANLAN_OLLAMA_PORT=str(c["ollama"]), PYTHONUTF8="1", PYTHONIOENCODING="utf-8"); return e def services(c): P = py(c); h = c["host"] return [ ("detail", c["detail"], [P, "scripts/windscada_serve.py", "--host", h, "--port", str(c["detail"])]), ("cms", c["cms"], [P, "scripts/windcms.py", "serve", "--farm", "rudong", "--port", str(c["cms"])]), ("viewer", c["viewer"], [P, "scripts/static_server.py", "--port", str(c["viewer"]), "--host", h, "--dir", str(ROOT / c["viewer_dir"]), "--comp", "viewer"]), ("sim_sys", c["sim_sys"], [P, "-u", str(ROOT / c["release_dir"] / "sim_sys_server.py")]), ("sim", c["sim"], [P, "scripts/static_server.py", "--port", str(c["sim"]), "--host", h, "--dir", str(ROOT / c["sim_dir"]), "--comp", "sim"]), ("gateway", c["gateway"], [P, "scripts/guanlan_gateway.py", "--host", h, "--port", str(c["gateway"])]), ] def spawn(cmd, log: Path, e): f = open(log, "ab") kw = dict(cwd=str(ROOT), env=e, stdout=f, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL) if WIN: # 2026-09-16: 统一到 CREATE_NO_WINDOW (src/proc.py) —— 原先用 DETACHED_PROCESS, 子进程没有控制台, # 而 .venv 的 python.exe 是**转发器**, 它会再拉一个真解释器; 落地实测那一步会冒出可见黑窗。 # CREATE_NO_WINDOW 给的是"没有窗口的控制台", stdio 照旧重定向到日志, 且与 DETACHED 互斥故只用前者。 from src import proc as _proc kw["creationflags"] = _proc.flags() else: kw["start_new_session"] = True return subprocess.Popen(cmd, **kw).pid def alive(pid): if not pid: return False if WIN: # errors="replace" 必须有: tasklist 的输出是**控制台代码页**(中文 Windows = GBK), 而本进程 # 可能是 PYTHONUTF8=1 起的(默认文本编码变 UTF-8) → 解码失败会让 r.stdout 变成 None, # 下一行 `str(pid) in None` 直接 TypeError。2026-09-12 由运维控制台的动作进程实测逮到。 # ★ 走 src.proc.run_text: 本函数也会被**无控制台**的进程调用 (网关/运维任务), 裸 spawn tasklist # 会给它新建可见控制台 → 反复闪窗 (2026-09-16)。 from src import proc as _proc r = _proc.run_text(["tasklist", "/FI", f"PID eq {pid}"]) return r.stdout is not None and str(pid) in r.stdout try: os.kill(pid, 0); return True except OSError: return False def kill(pid): from src import proc as _proc if WIN: _proc.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True) else: try: os.killpg(os.getpgid(pid), signal.SIGTERM) except OSError: try: os.kill(pid, signal.SIGTERM) except OSError: pass def cmd_check(c): ok = True def row(name, good, note=""): nonlocal ok; ok &= bool(good); print(f" [{'OK' if good else 'FAIL'}] {name}{(' — ' + note) if note else ''}") print(f"观澜·如东样板 v2 自检 @ {ROOT}"); v = sys.version_info; row("Python ≥ 3.11", v >= (3, 11), f"{v.major}.{v.minor}.{v.micro} ({sys.executable})") # 源码可编译性 —— 重点是**非法转义序列**: 普通字符串里写 `\d` `\*` 这类, Python 3.12 只给 # SyntaxWarning, 未来版本会变成 SyntaxError。这类问题在开发机上永远看不见, 换台机器/换个解释器才炸 # (2026-09-11 逮到 src\windcms\serve.py 里 JS 正则的两处, 已按"补成合法转义"修好)。 srcs = sorted(list((ROOT / "src").rglob("*.py")) + list((ROOT / "scripts").rglob("*.py"))) bad = [] with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") for f in srcs: try: compile(f.read_text(encoding="utf-8"), str(f), "exec") except SyntaxError as e: bad.append(f"{f.relative_to(ROOT)}:{e.lineno} {e.msg}") for it in w: if issubclass(it.category, SyntaxWarning): bad.append(f"{Path(str(it.filename)).name}:{it.lineno} {it.message}") row(f"源码可编译 ({len(srcs)} 个文件, 无非法转义/语法错)", not bad, "; ".join(bad[:3])) # 子进程口径自检 (2026-09-16 加): 这三条是**回归守卫** —— 它们各自都真炸过一次, 而炸法都是"静默": # ① capture_output=True 与显式 stdout 互斥 → 异常被 except 吞成 alive=False ⇒ 在跑的任务被 # 误判成"被强杀", 页面显示完成、重算按钮可再点(可能并发重算); # ② CREATE_NO_WINDOW 新建的控制台会把子进程标准句柄吸走 ⇒ 重算日志一个字都没有、任务挂住; # ③ 不弹窗 (NO_WINDOW 位必须真的设上, 否则"点按钮闪黑窗"回来)。 try: from src import proc as _proc _r = _proc.run_text([sys.executable, "-c", "print('proc-ok')"]) row("子进程口径: run_text(capture_output) 可用", _r.returncode == 0 and "proc-ok" in (_r.stdout or ""), "src/proc.py (capture_output 与显式 stdout 互斥, 冲突会让 job_running 误判任务已死)") import tempfile as _tf _lg = Path(_tf.gettempdir()) / "_guanlan_proc_check.log" _p = _proc.spawn([sys.executable, "-c", "print('spawn-ok')"], log=_lg); _p.wait(timeout=60) _txt = _lg.read_text(encoding="utf-8", errors="replace") if _lg.exists() else "" try: _lg.unlink() except OSError: pass row("子进程口径: spawn(log=) 输出进日志", "spawn-ok" in _txt, "CREATE_NO_WINDOW 会吸走句柄, 不显式继承则日志空白 (重算看起来卡住)") row("子进程口径: 无窗口位已设 (CREATE_NO_WINDOW)", bool(_proc.flags() & 0x08000000) or os.name != "nt", f"flags={hex(_proc.flags())}") except Exception as _e: row("子进程口径自检", False, f"{type(_e).__name__}: {_e}") # 入口引用闭合 (2026-09-16 现场实炸后加): 目标机 (D:\产品\app) 双击 start.bat 报 # "无法找到脚本文件 start_hidden.vbs" —— 包少带了一个被引用的文件。这条守卫在**装机后**复查一遍: # start.bat/check.bat/stop.bat/install.* 引用的每个文件都必须在位, 缺一个就 FAIL。 # (那个 .vbs 已按用户令删除, 无窗口启动改用 pythonw.exe + scripts\guanlan_start_hidden.py; # 守卫与实现语言无关, 照样生效 —— 现在它盯的是 start.bat → scripts\guanlan_start_hidden.py 这类引用。) try: from src.entry_refs import encoding_problems, missing_refs, referenced_tree_files _miss = missing_refs(ROOT) _nref = sum(len(v) for v in referenced_tree_files(ROOT).values()) row(f"入口脚本引用闭合 ({_nref} 条: start.bat→guanlan_start_hidden.py 等)", not _miss, "; ".join(f"{e} 缺 {r}" for e, r in _miss[:3]) or "每个入口引用的文件都在位") _enc = encoding_problems(ROOT) row("入口脚本编码守则 (.ps1 = UTF-8 BOM + CRLF; .bat = 纯 ASCII/CRLF/无 BOM/无 `<<`; .sh = LF)", not _enc, "; ".join(_enc[:2]) or "编码与字符都对 (PS 5.1 才不会把中文按 GBK 解错; 批处理里不放中文/重定向符)") except Exception as _e: row("入口脚本引用闭合", False, f"{type(_e).__name__}: {_e}") # 页面归口 (2026-09-17 用户令 1): 门户里哪些页面是"数据派生快照"(属于产物)、它们陈旧没陈旧。 # 用户问过"#findings/#sim/#documents 该不该随输入数据变、算不算产物" —— 答案落在 # configs/portal_pages.yaml 登记表 + scripts/pages_audit.py 的检查里, 这里只报结论。 # 退出码: 0 一致 / 5 缺文件或未登记 / 6 溯源缺失 / 7 **页面陈旧**(数据变了页面没变) / 8 交付件缺版本 / 9 分类错。 try: import importlib.util as _ilu _spec = _ilu.spec_from_file_location('_pages_audit', ROOT / "scripts" / "pages_audit.py") _m = _ilu.module_from_spec(_spec) _spec.loader.exec_module(_m) _rc, _res, _gaps = _m.audit() _bad = [r for r in _res if r[0] not in ("OK", "i")] _known = [r for r in _res if r[0] == "i"] _note = (f"检查 {len(_res)} 项: 不一致 {len(_bad)}" + (f" ({'; '.join(f'{r[1]}: {r[2][:60]}' for r in _bad[:2])})" if _bad else "") + f" · 已知缺口 {len(_known)} (引用悬空/无溯源, 登记表明写理由)" + (f" · 未登记页 {len(_gaps)}" if _gaps else "")) row("页面归口: 页面是否随数据变 / 是否陈旧 (configs/portal_pages.yaml)", _rc == 0, _note) except Exception as _e: row("页面归口审计", False, f"{type(_e).__name__}: {_e}") # 配置与日志的统一口径 (2026-09-17 用户令 2): 配置目录/文件、日志目录/命名/格式。 # 两个检查器都是"看一眼实物 + 代码"的静态检查, 很便宜; 有问题就 FAIL 并指到具体文件。 for _script, _label in (("config_audit.py", "配置统一: 目录约定/命名/内容/引用闭合/不手拼路径"), ("log_audit.py", "日志统一: logs/ 唯一落点/命名/行格式/保留策略")): try: import importlib.util as _ilu2 _sp = _ilu2.spec_from_file_location(f'_{_script[:-3]}', ROOT / "scripts" / _script) _mod = _ilu2.module_from_spec(_sp) _sp.loader.exec_module(_mod) _rc2, _res2, _info2 = _mod.audit() _lvl = {} for _l in _res2: _lvl[_l[0]] = _lvl.get(_l[0], 0) + 1 _bad2 = [r for r in _res2 if r[0] in ("X", "!")] row(_label, _rc2 == 0, (f"rc={_rc2}: " + "; ".join(f"{r[1]}: {r[2][:70]}" for r in _bad2[:2])) if _bad2 else f"无不一致 · 已知缺口/白名单 {_lvl.get('i', 0)} · 提示 {_lvl.get('?', 0)}") except Exception as _e: row(_label, False, f"{type(_e).__name__}: {_e}") # 输入数据放置体检 (2026-09-17 用户令 3): data/raw 的结构/命名是否合约定、源类齐不齐。 # 判 FAIL 的只有**结构类**问题 (放错层/名字不对/压缩包没解压) —— 那类问题会让摄入静默读不到; # "缺某类源件"只作为提示 (新装的机器本来就没数据, 不该因此判失败)。 try: import importlib.util as _ilu3 _sp3 = _ilu3.spec_from_file_location('_raw_check', ROOT / "scripts" / "raw_data_check.py") _rc3 = _ilu3.module_from_spec(_sp3) _sp3.loader.exec_module(_rc3) _r3, _n3, _s3 = _rc3.check_tree(Path(_rc3.station_dir()), deep=False) _struct = [x for x in _r3 if x[3] in (_rc3.RC_STRUCT, _rc3.RC_CLASH)] _gaps = [x for x in _r3 if x[3] == _rc3.RC_GAP] if not _s3.get('exists'): row("输入数据放置体检 (data/raw)", True, "尚未放置输入数据 —— 属预期; 放置规范见 docs/输入数据放置指导_v0.1.md") else: row("输入数据放置体检 (data/raw 结构/命名/源类)", not _struct, (f"{len(_struct)} 项结构问题: " + "; ".join(f"{x[1]}: {x[2][:60]}" for x in _struct[:2])) if _struct else f"{_s3.get('files', 0):,} 件源件 · 结构合规" + (f" · 缺 {len(_gaps)} 类/项: " + "; ".join(f"{x[1]}: {x[2][:50]}" for x in _gaps[:2]) if _gaps else "") + (f" · {len(_n3)} 条提示" if _n3 else "")) except Exception as _e: row("输入数据放置体检", False, f"{type(_e).__name__}: {_e}") for m in ("numpy", "pandas", "pyarrow", "polars", "yaml", "matplotlib", "plotly", "jinja2", "docx"): try: __import__(m); row(f"依赖 {m}", True) except Exception as ex: row(f"依赖 {m}", False, f"未安装: {ex.__class__.__name__} (运行 install 脚本)") # ★2026-09-16 用户令: 打包"不含 输入数据/产物/日志"。产物缺失时, 若包内 dist-manifest.json 就声明了 # no_products/no_data, 这里显示 `--`(待重算) 而**不是 FAIL** —— 否则开箱自检会把"本来就该重算"的 # 状态报成失败, install 脚本收尾的那次 check 也会失败 (把正常状态说成问题)。 _man = {} try: _mp = ROOT / "dist-manifest.json" _man = jload(_mp) if _mp.is_file() else {} except Exception: _man = {} _no_products = bool(_man.get("no_products")) for rel, n in (("outputs/rudong/windscada", "L0/L1 产物 (parquet)"), ("outputs/rudong/ontology/objects.json", "本体对象库"), ("outputs/rudong/sop/findings.json", "findings"), ("outputs/rudong/guanlan/facts_contract_v0.json", "事实契约"), (c["release_dir"] + "/portal.html", "门户"), (c["release_dir"] + "/sim_sys_server.py", "仿真合页服务"), (c["release_dir"] + "/如东SWT40_控制律仿真台_20260906.zip", "仿真合页资料包"), (c["viewer_dir"], "三维 viewer 资产"), (c["sim_dir"], "仿真回放资产"), (c["release_dir"] + "/如东", "治理清单交付件 (可选)")): p = ROOT / rel; exists = p.exists() and (any(p.iterdir()) if p.is_dir() else p.stat().st_size > 0) if "可选" in n: print(f" [{'OK' if exists else '--'}] {n}: {rel}") elif _no_products and rel.startswith("outputs/"): print(f" [--] {n}: {rel} (本包按用户令**未随产物** —— 把现场包放好后跑 " f"scripts/place_raw_data.py --src <现场包目录> --scope full 再 scripts/rebuild_all.py)") else: row(n, exists, rel) raw = Path(c.get("raw_dir", "data/raw")); raw = raw if raw.is_absolute() else (ROOT / raw) print(f" [{'OK' if raw.exists() else '--'}] 原始件目录 (补数据放这里): {raw}{'' if raw.exists() else ' (不存在; 原始件不随包分发, 只影响摄入命令, 不影响页面)'}") for name, port, _ in services(c): print(f" [{'运行中' if up(c['host'], port) else '空闲'}] 端口 {port} ({name})") # 模型闸: 探针 (generate 非 tags) + 档位模型是否都在本机 + 摘要记录; 无模型时问答不可用, 其余页面不受影响 sys.path.insert(0, str(ROOT)) try: from src.ontology import llm_gate; mcfg = jload(P.MODELS_JSON) pr = llm_gate.probe(mcfg["tiers"]["default_qa"]["model"], timeout=90); row("本机模型 Ollama 探针 (默认档)", pr["ok"], f"{pr.get('latency_s')} s, digest {pr.get('digest')}" if pr["ok"] else pr.get("err", "") + " (启动 Ollama 并按 configs/models.json 的 pull_commands 拉模型)") try: with urllib.request.urlopen(f"http://{c['host']}:{c['ollama']}/api/tags", timeout=10) as r: have = {m["name"] for m in json.load(r).get("models", [])} for tier, t in mcfg["tiers"].items(): name = t["model"]; ok_ = name in have or (name + ":latest") in have; print(f" [{'OK' if ok_ else '--'}] 模型档 {tier}: {name}{'' if ok_ else ' (未下载: ' + 'ollama pull ' + name + ')'}") except Exception as ex: print(f" [--] 模型清单不可读: {ex.__class__.__name__}") st = llm_gate.state(); print(f" [{'FAIL' if st['open'] else 'OK'}] 模型可用性检查: {'已熔断 (连续失败后自动停用, 60 秒后重试)' if st['open'] else '正常'}; 审计日志 {st['audit']}") except Exception as ex: row("模型可用性检查", False, f"{ex.__class__.__name__}: {str(ex)[:100]}") print("结论:", "全绿, 可 serve" if ok else "有 FAIL 项, 先按提示处理"); return 0 if ok else 2 def cmd_serve(c): from src import logfile as _lf; _lf.prefix_stdout('serve') # 统一日志格式 (用户令 2) RUN.mkdir(exist_ok=True); LOGS.mkdir(exist_ok=True); pids = jload(PIDS) if PIDS.exists() else {}; e = env(c) for name, port, cmd in services(c): if up(c["host"], port): print(f" {name} {port} 已在运行 (复用)"); continue if name == "gateway": time.sleep(1.5) pid = spawn(cmd, LOGS / f"{name}.log", e); pids[name] = {"pid": pid, "port": port}; print(f" {name} {port} 启动 pid {pid}") PIDS.write_text(json.dumps(pids, ensure_ascii=False, indent=1), encoding="utf-8") url = f"http://{c['host']}:{c['gateway']}/healthz" for _ in range(40): try: with urllib.request.urlopen(url, timeout=20) as r: d = json.load(r); break except Exception: time.sleep(1) else: print("网关 healthz 60 s 内未就绪, 看 logs/gateway.log"); return 2 bad = [m for m in d["modules"] if not m["ok"]]; print(f"入口 http://{c['host']}:{c['gateway']}/ 状态 {d['status']} {d['mode']} 模块 {len(d['modules']) - len(bad)}/{len(d['modules'])} ok") for m in bad: print(f" [DOWN] {m['path']} {m['name']} (看 logs/)") return 0 if not bad else 1 def cmd_status(c): pids = jload(PIDS) if PIDS.exists() else {} for name, port, _ in services(c): p = pids.get(name, {}).get("pid"); print(f" {name:8s} 端口 {port} {'监听中' if up(c['host'], port) else '未监听'} pid {p if p else '-'} {'存活' if p and alive(p) else ''}") return 0 def cmd_stop(c): if not PIDS.exists(): print("没有 pids.json (不是本启动器起的, 或已停)"); return 0 pids = jload(PIDS) for name, v in pids.items(): kill(v["pid"]); print(f" 停 {name} pid {v['pid']}") time.sleep(1.5); left = [n for n, p, _ in services(c) if up(c["host"], p)]; PIDS.unlink() print("全部已停" if not left else f"仍在监听 (可能是别的实例): {left}"); return 0 def main(): a = sys.argv[1] if len(sys.argv) > 1 else "check"; c = cfg() if a == "open": url = f"http://{c['host']}:{c['gateway']}/"; print(url); webbrowser.open(url); return 0 if a == "qa": return subprocess.call([py(c), str(ROOT / "scripts/llm_known_answer.py"), "--port", str(c["detail"])] + sys.argv[2:], cwd=str(ROOT)) return {"check": cmd_check, "serve": cmd_serve, "status": cmd_status, "stop": cmd_stop}.get(a, lambda c: (print(__doc__), 2)[1])(c) if __name__ == "__main__": sys.exit(main())