lou
53e7d1e5f2
文档/ (2026-09-16; 老板定调: 这是 agent 底座, 所以必须扎实, 现在越扎实以后开发越简单)
00-索引 文档地图 + 30 秒概念速查 + 三条命令跑起来 + 事实源优先级
01-快速上手 体检 -> 建库 -> 起内核 -> 起驱动 -> 收工, 全带实测输出; 第一次最易踩的四个坑
02-写一个驱动 五分钟最小驱动 / 形态选择 / 能碰哪些表 / 汇报与调用两份模板 / 交付检查表
03-命令手册 两层每条命令 + 日志选项 + 退出码约定 + --json 样例 + 日常十条
04-契约与调用 一次调用的完整生命周期 / 六条仲裁 / 锁与按需拉起 / 排障表
05-日志与排障 三条道怎么读 + "症状->判据->处置"总表 + 断电收尸语义
06-架构与不变量 分层 / 14 条硬不变量 / 主流程表 / 双真相 / 为什么故意不做 / 已知薄弱点
07-模块与接口 逐模块职责与公开接口 + "想改 X -> 动哪几处"连带清单
08-数据模型 8 张表逐字段 (谁写谁读) + events.kind 字典 + 状态机 + 快照 + 排查 SQL
09-扩展指南 六个配方 (加子命令/加字段/加表/加日志来源/加自测/改判定) + 同步清单
10-验收与质量门 四道门 + 五份自测明细 + pyright 严格档 + 26 条已知坑总表 + 发布 checklist
规矩: 不重复设计文档 / 每条命令实测过再写 (含 jq 表达式) / 代码>设计>文档 的事实源优先级 /
改代码必须同步文档 (清单在 09 末尾) / 暂时没做到的事写成"已知边界"不含糊过去
修复: 同锁串行化原来是死的 (实测抓到的真缺陷)
旧行为: db.领调用 只领 pending (waiting 没人再碰) + db.同锁在跑 把 waiting 也算"占着锁"
-> 同一把锁上两条请求互相排队, 双双停在 waiting 谁也不跑 (实测 id 16/17);
而 收权超时 只收 running -> 排队连超时都没有 = 死锁
修法: ① db.领调用 的 SQL 改 state IN ('pending','waiting') -- 每轮把排队的领回来重判, 锁一空就推进
② db.同锁在跑 只认 state='running' (排队的还没拿到锁, 不挡人)
③ 内核.转发调用 waiting 分支补 deadline (排队也立期限); 内核.收权超时 遍历 running + waiting
④ 抽出 内核.期限文本() 统一算 deadline
实测: 两条同锁调用串行跑完 (19.started_at == 18.finished_at); 排队者超时被收权 (events 有记录)
回归: 自测db.py 调用组 +4 条断言 (waiting 不算占着锁 / waiting 会被重新领 / ...);
去掉一条依赖生产库全局计数的脆弱断言
其它: 内核 与 引导器 的 用法() 末尾加文档指引
验收: uvx pyright 0 errors / 0 warnings; 五份自测全过 (进程/内核 58/配置/db/日志 86);
试跑引导器.py PASS 11 / FAIL 0 / 残留无; 残留进程 0
1594 lines
68 KiB
Python
1594 lines
68 KiB
Python
#!/usr/bin/env python3
|
|
"""UEFI.boot.py -- 引导器 (内核的管家).
|
|
|
|
[它干什么]
|
|
三件事: 管内核环境 / 管包状态 / 管内核进程.
|
|
驱动不归它管 -- 那是内核的活.分层是一层管一层: 引导器管内核, 内核管驱动.
|
|
它只保证"内核活着,跑得起来": 拉起 / 判活 / 重启 / 收日志, 不干预内核怎么调度驱动.
|
|
|
|
[铁律](改这个文件前先读一遍)
|
|
1. 零第三方依赖 -- 引导器自己要是依赖某个包, 那包坏了引导器也起不来, 就**没人来报错了**
|
|
(鸡生蛋问题: 所以它只用 stdlib, 连读写 PG 都是 try import psycopg2 的降级写法).
|
|
2. 引导器自己的命令 (自检 / 环境 / 包) PG 连不上**只 WARN** + 照旧写 json 快照, 不算失败.
|
|
透传给内核的命令**不降级** -- 内核的内存就是 PG, 它自己会报错退出.
|
|
一句话分界: "引导器自己的命令 PG 降级, 内核的命令不降级".
|
|
3. 不自动修复 / 不自动装包 / 不自动升级 -- 只有 `环境 重建`,`包 安装` 这两个**显式敲**的命令才动手.
|
|
(老板口径: "一搞就炸", 手动控制.)
|
|
4. 顶层只有 4 个词归引导器 (自检 / 环境 / 包 / 内核), 其余 argv **原样透传**内核:
|
|
所以 `启动 <驱动>` 归内核,`内核 启动` 归引导器, 重名不冲突.
|
|
|
|
[文件与外部依赖]
|
|
环境.efi.json 项目级唯一配置 (引导器是它唯一的写者, 内核只读)
|
|
环境状态.efi.json 每次体检的快照 (原子替换; .gitignore 盖住)
|
|
内核/内核.py 被移交的内核入口 (引导器不实现它的子命令)
|
|
内核/进程.py 通用进程库 (管内核进程用它, 内核管驱动也用它 -- 只此一份实现)
|
|
内核/db.py 唯一碰 SQL 的文件 (这里只调函数, 不写 SQL)
|
|
内核/文本.py CJK 宽度对齐的表格输出
|
|
|
|
[命令一览](完整帮助见 用法())
|
|
体检类: (无参数) / --check | 自检 / 环境 [--json] / 环境 重建
|
|
包类: 包 [--json] / 包 安装 [名...]
|
|
内核类: 内核 状态 | 启动 [--守护] | 停止 | 重启 | 日志 [-n N] [-f]
|
|
其余: 原样透传内核 (列表 / 扫描 / 启动 <驱动> / 停止 <驱动> / 重启 <驱动> /
|
|
状态 / 日志 <驱动> / 事件 / 清单)
|
|
|
|
[退出码约定]
|
|
0 正常
|
|
1 体检有阻塞项 / 内核文件缺失 / 停不干净
|
|
2 命令不认识
|
|
其他 内核自己的退出码 (原样返回)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
# ─────────────────────────────── 路径与常量 ───────────────────────────────
|
|
|
|
# 项目根 = 本文件所在目录 (所以从哪敲都能跑, 不依赖 cwd)
|
|
项目根: Path = Path(__file__).resolve().parent
|
|
内核目录: Path = 项目根 / "内核"
|
|
# 把 内核/ 加进模块搜索路径, 这样能直接 `import 进程` 这种中文模块名
|
|
# (目录名 内核 是中文, 当不了包名, 所以不写 `import 内核.进程`).
|
|
sys.path.insert(0, str(内核目录))
|
|
|
|
import db # noqa: E402 (内核/ 下的公用模块)
|
|
import 进程 # noqa: E402
|
|
import 文本 # noqa: E402
|
|
import 日志 # noqa: E402
|
|
|
|
环境文件: Path = 项目根 / "环境.efi.json" # 唯一一份配置 (引导器写, 内核只读)
|
|
状态文件: Path = 项目根 / "环境状态.efi.json" # 体检快照 (落盘副本, PG 台账才是活真相)
|
|
内核入口: Path = 内核目录 / "内核.py" # 被移交的内核入口
|
|
内核日志路径: Path = 日志.内核日志路径(项目根) # 内核的结构化日志行 (内核自己写, 不靠 fd 重定向)
|
|
内核输出路径: Path = 日志.内核输出路径(项目根) # 内核 stdout/stderr 原始流 (命令输出 + 崩溃原文)
|
|
引导器日志路径: Path = 日志.引导器日志路径(项目根) # 引导器自己的动作 (体检/包/移交/内核启停)
|
|
# 注意: 这几个常量**不能**叫"内核日志" -- 那会被下面 `def 内核日志()` 同名遮蔽, 打印出来是 <function> (踩过).
|
|
|
|
引导器版本: str = "引导器 v0.1"
|
|
内核版本: str = "内核 v0.1"
|
|
|
|
左: str = 文本.左 # 表格对齐用的字面量, 转发一份少写几个字
|
|
右: str = 文本.右
|
|
|
|
# 环境.efi.json 缺失时用的默认模板 (引导器是配置的唯一写者, 所以它能生成一份).
|
|
# 这几段是"第一次跑起来"的最低要求: 解释器要求 / venv 路径 / 国内源 / 内核必需包 / PG 连接 / 驱动目录.
|
|
# db 段的默认值优先吃 PG* 环境变量, 再兜底本机常见位置 -- 换机器不用改代码.
|
|
默认环境: dict[str, Any] = {
|
|
"efi": 1,
|
|
"python": ">=3.11",
|
|
"venv": ".venv",
|
|
"index": "https://pypi.tuna.tsinghua.edu.cn/simple", # 校网必须走国内源 (GFW)
|
|
"packages": [
|
|
{
|
|
"name": "psycopg2-binary",
|
|
"version": ">=2.9.9",
|
|
"import": "psycopg2",
|
|
"required": True,
|
|
}
|
|
],
|
|
"db": {
|
|
"name": "efi_kernel",
|
|
"host": os.environ.get("PGHOST") or str(Path.home() / "pgdata" / "socket"),
|
|
"port": int(os.environ.get("PGPORT") or 5432),
|
|
"user": os.environ.get("PGUSER") or os.environ.get("USER") or "lou",
|
|
},
|
|
"driver_root": "驱动",
|
|
"stop_timeout": 10, # 停进程: SIGTERM 之后等几秒再 SIGKILL
|
|
"log_lines": 200, # 日志命令默认显示多少行
|
|
"log_level": "INFO", # 日志门槛: DEBUG / INFO / WARN / ERROR (低于它的不写)
|
|
"log_max_mb": 5, # 单份日志超过它就轮转 (改成 <文件>.1, 旧的往后挤)
|
|
"log_keep": 3, # 轮转后留几份历史
|
|
}
|
|
|
|
|
|
# ─────────────────────────────── 小工具 ───────────────────────────────
|
|
|
|
|
|
def 现在() -> str:
|
|
"""当前时间, ISO8601 带时区 (+08:00).给人看,写进 json 都用这个格式."""
|
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def 本地时间(时间戳: float) -> str:
|
|
"""unix 时间戳 -> 本机时区的 ISO8601 (进程启动时刻显示用)."""
|
|
return datetime.fromtimestamp(时间戳).astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def 原子写json(路径: Path, 内容: dict[str, Any]) -> None:
|
|
"""原子写 json: 先写 `<名字>.tmp` 再 os.replace 顶上.
|
|
|
|
参数:
|
|
路径: 目标文件 (父目录不存在会自动建).
|
|
内容: 要序列化的 dict (ensure_ascii=False, 中文不转义, 老板要能直接看).
|
|
|
|
为什么必须原子:
|
|
os.replace 在同文件系统里是原子的, 所以读的人永远看到"完整的旧版"或"完整的新版",
|
|
不会看到写了一半的残缺 json (快照文件被读到半截 = 下游把无快照当"没有状态"处理).
|
|
"""
|
|
路径.parent.mkdir(parents=True, exist_ok=True)
|
|
临时 = 路径.with_name(路径.name + ".tmp")
|
|
临时.write_text(json.dumps(内容, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
os.replace(临时, 路径)
|
|
|
|
|
|
def 说(行: list[str]) -> None:
|
|
"""整块打印 (转发给 文本.打印, 一次写完不逐行挤)."""
|
|
文本.打印(行)
|
|
|
|
|
|
def 日志上限(配置: dict[str, Any]) -> int:
|
|
"""单份日志的轮转阈值, 单位字节 (log_max_mb, 默认 5 MB; 0 = 不轮转)."""
|
|
原始 = 配置.get("log_max_mb")
|
|
if 原始 is None:
|
|
return 日志.默认上限字节
|
|
try:
|
|
return max(int(float(str(原始)) * 1024 * 1024), 0)
|
|
except (TypeError, ValueError):
|
|
return 日志.默认上限字节
|
|
|
|
|
|
def 日志保留(配置: dict[str, Any]) -> int:
|
|
"""轮转后留几份历史 (log_keep, 默认 3; 0 = 不轮转)."""
|
|
try:
|
|
return max(int(配置.get("log_keep") or 日志.默认保留份数), 0)
|
|
except (TypeError, ValueError):
|
|
return 日志.默认保留份数
|
|
|
|
|
|
def 记日志(级别: str, 消息: str, 安静: bool = False) -> None:
|
|
"""写一条引导器日志: 落 内核/logs/引导器.log, 默认同时打 stderr (带 [级别] 前缀).
|
|
|
|
为什么要有它: 引导器以前只打 stderr, 跑完就没了 -- "上次体检几点,谁改了配置,
|
|
内核是不是它起的"查不到历史. 现在 `日志` 就能翻出来 (引导器自己的动作也是审计对象).
|
|
|
|
参数:
|
|
级别: INFO / WARN / ERROR.
|
|
消息: 内容.
|
|
安静: True 只落文件不打 stderr (给"每次跑都有的流水账"用, 如开始/收工两条;
|
|
这样 stderr 只在真有警告时说话, 不吵着 stdout 的输出).
|
|
|
|
引导器是短命进程 (敲一次跑一次), 所以不设门槛 (全记) + 每次跑前轮转一次 (起日志).
|
|
"""
|
|
日志.记(引导器日志路径, 级别, "引导器", 消息, 门槛="", 控制台=False)
|
|
if not 安静:
|
|
print(f"[{级别}] {消息}", file=sys.stderr, flush=True)
|
|
|
|
|
|
def 起日志(配置: dict[str, Any]) -> None:
|
|
"""引导器开跑第一件事: 轮转自己的日志 (超上限就改名 .1, 别让它无限长)."""
|
|
日志.轮转(引导器日志路径, 日志上限(配置), 日志保留(配置))
|
|
|
|
|
|
def db段(配置: dict[str, Any]) -> dict[str, Any]:
|
|
"""取配置里的 db 段 (不是对象就给空字典, 让 db.从配置() 去兜默认值, 不在这崩)."""
|
|
return 取对象(配置.get("db"))
|
|
|
|
|
|
def 取对象(值: Any) -> dict[str, Any]:
|
|
"""把 json 里取出来的"对象"统一成 dict[str, Any]; 不是字典就给空字典.
|
|
|
|
为什么需要这么一个函数:
|
|
json.loads 出来的是 Any (允许), 但 isinstance(x, dict) 窄化出来的只是 dict[Unknown, Unknown] --
|
|
对窄化后的 dict 取成员, 严格模式会报"类型部分未知". 在这里 cast 一次, 后面所有 .get() 都干净.
|
|
顺便把 None / 字符串这类脏值挡成空字典, 调用方少写一个 if.
|
|
"""
|
|
if isinstance(值, dict):
|
|
return cast(dict[str, Any], 值)
|
|
return {}
|
|
|
|
|
|
def 取清单(值: Any) -> list[Any]:
|
|
"""把 json 里取出来的"数组"统一成 list[Any]; 不是数组就给空表 (同上, 为了不留 Unknown)."""
|
|
if isinstance(值, list):
|
|
return cast(list[Any], 值)
|
|
return []
|
|
|
|
|
|
def 驱动目录(配置: dict[str, Any]) -> Path:
|
|
"""驱动根目录的绝对路径 (配置里写的是相对项目根的路径, 如 "驱动")."""
|
|
return (项目根 / str(配置.get("driver_root", "驱动") or "驱动")).resolve()
|
|
|
|
|
|
def venv路径(配置: dict[str, Any]) -> Path:
|
|
"""内核 venv 的绝对路径 (配置里的 venv 相对项目根, 默认 .venv)."""
|
|
return (项目根 / str(配置.get("venv", ".venv") or ".venv")).resolve()
|
|
|
|
|
|
# ─────────────────────────────── 配置 ───────────────────────────────
|
|
|
|
|
|
def 读环境() -> tuple[dict[str, Any], list[str]]:
|
|
"""读项目根的 环境.efi.json.
|
|
|
|
返回:
|
|
(配置, 警告列表).配置 = 默认模板 被文件内容覆盖后的字典; 警告是要打印给老板的提示.
|
|
|
|
容错策略 (都是"能继续就继续, 但要说清楚"):
|
|
* 文件不存在 -> 按默认模板生成一份 (引导器是它唯一写者), 并 WARN 提醒检查 db 段
|
|
* 读不了 / 不是合法 JSON / 顶层不是对象 -> **不覆盖用户文件**, 用默认值继续 + WARN
|
|
顶层 merge 而不是全量替换: 用户少写了某几个键也能跑 (缺的吃默认值), 不会 KeyError.
|
|
"""
|
|
警告: list[str] = []
|
|
if not 环境文件.exists():
|
|
原子写json(环境文件, 默认环境)
|
|
警告.append(f"环境.efi.json 不存在, 已按默认模板生成 (请检查 db 段): {环境文件}")
|
|
return dict(默认环境), 警告
|
|
try:
|
|
原文 = 环境文件.read_text(encoding="utf-8")
|
|
except OSError as 错:
|
|
警告.append(f"环境.efi.json 读不了 ({错}), 按默认模板继续, 没有覆盖你的文件")
|
|
return dict(默认环境), 警告
|
|
try:
|
|
配置 = json.loads(原文)
|
|
except json.JSONDecodeError as 错:
|
|
警告.append(f"环境.efi.json 不是合法 JSON ({错}), 按默认模板继续, 没有覆盖你的文件")
|
|
return dict(默认环境), 警告
|
|
if not isinstance(配置, dict):
|
|
警告.append("环境.efi.json 顶层不是对象, 按默认模板继续")
|
|
return dict(默认环境), 警告
|
|
合并: dict[str, Any] = dict(默认环境)
|
|
合并.update(取对象(配置))
|
|
return 合并, 警告
|
|
|
|
|
|
def 打印警告(警告: list[str]) -> None:
|
|
"""把配置层的警告打到 stderr 并记进引导器日志 (不污染 stdout 的表格输出, 重定向时能分清)."""
|
|
for 条 in 警告:
|
|
记日志("WARN", 条)
|
|
|
|
|
|
# ─────────────────────────────── 版本比较 / 包 ───────────────────────────────
|
|
# 这一段是"自己撸一个够用的版本比较", 不引 packaging 包 -- 引导器零第三方依赖是硬要求.
|
|
|
|
|
|
def 解析版本(文本: str) -> tuple[int, ...]:
|
|
"""把版本串拆成数字段: "2.9.11" -> (2, 9, 11).
|
|
|
|
参数:
|
|
文本: 版本串, 可以带前缀/后缀 (如 "v1.2.3-rc1").
|
|
返回:
|
|
数字元组; 遇到第一个不是数字开头的段就停.
|
|
|
|
够用不超配:
|
|
只比数字段, 不吃 alpha/beta/rc 的语义 (本项目依赖只用 >= 一种约束).
|
|
这样实现 20 行搞定, 也不用为它引一个第三方库.
|
|
"""
|
|
出: list[int] = []
|
|
for 段 in re.split(r"[._\-+]", 文本.strip()):
|
|
命中 = re.match(r"^(\d+)", 段)
|
|
if 命中 is None:
|
|
break
|
|
出.append(int(命中.group(1)))
|
|
return tuple(出)
|
|
|
|
|
|
def 比版本(左值: tuple[int, ...], 右值: tuple[int, ...]) -> int:
|
|
"""比两个已拆好的版本元组 (短的补 0 再比).返回 -1 / 0 / 1."""
|
|
长 = max(len(左值), len(右值))
|
|
左补 = 左值 + (0,) * (长 - len(左值))
|
|
右补 = 右值 + (0,) * (长 - len(右值))
|
|
if 左补 < 右补:
|
|
return -1
|
|
if 左补 > 右补:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def 满足要求(实装: str, 要求: str) -> bool:
|
|
"""判断实装版本是否满足要求串.
|
|
|
|
参数:
|
|
实装: 实际装上的版本 (如 "2.9.13"); 空串 = 没装 = 不满足.
|
|
要求: 约束串, 支持 > >= < <= == != ~= 与逗号分隔的多条件 (如 ">=2.9.9,<3"); 空 = 不校验.
|
|
返回:
|
|
满足 True / 不满足 False.
|
|
|
|
说明:
|
|
~= 按"大于等于 + 主版本相同"处理 (PEP 440 的完全语义没实现, 本项目用不到).
|
|
要求串里没有可识别的比较符时按"不校验"处理 (返回 True), 避免把写错的配置判成缺包.
|
|
"""
|
|
if not 要求.strip():
|
|
return True
|
|
if not 实装.strip():
|
|
return False
|
|
实 = 解析版本(实装)
|
|
命中过 = False
|
|
for 条 in 要求.split(","):
|
|
条 = 条.strip()
|
|
if not 条:
|
|
continue
|
|
命中 = re.match(r"^(>=|<=|==|!=|~=|>|<)\s*(.+)$", 条)
|
|
if 命中 is None:
|
|
continue
|
|
命中过 = True
|
|
符 = 命中.group(1)
|
|
期 = 解析版本(命中.group(2))
|
|
关系 = 比版本(实, 期)
|
|
if 符 == ">=" and 关系 < 0:
|
|
return False
|
|
if 符 == ">" and 关系 <= 0:
|
|
return False
|
|
if 符 == "<=" and 关系 > 0:
|
|
return False
|
|
if 符 == "<" and 关系 >= 0:
|
|
return False
|
|
if 符 == "==" and 关系 != 0:
|
|
return False
|
|
if 符 == "!=" and 关系 == 0:
|
|
return False
|
|
if 符 == "~=":
|
|
if 关系 < 0:
|
|
return False
|
|
if 实 and 期 and 实[0] != 期[0]:
|
|
return False
|
|
return True if 命中过 else True
|
|
|
|
|
|
def 规范包名(名: str) -> str:
|
|
"""包名归一化 (PEP 503): 小写 + 把 -/_/. 都折成 -.
|
|
|
|
为什么需要:
|
|
`uv pip list` 报 `psycopg2-binary`, 配置里可能写 `psycopg2_binary`, 大小写也可能不同.
|
|
两边都归一化再对表, 不然会出现"明明装了却说缺失".
|
|
"""
|
|
return re.sub(r"[-_.]+", "-", 名).strip().lower()
|
|
|
|
|
|
def 读解释器版本(解释器: Path) -> str:
|
|
"""问一个解释器它是几版 (跑 `<解释器> -c ...`).
|
|
|
|
返回:
|
|
"3.11.15" 这样的串; 跑不起来/读不到返回空串 (调用方按"版本不明"处理, 不在这抛).
|
|
"""
|
|
结果 = 进程.跑命令(
|
|
[str(解释器), "-c", "import sys;print('%d.%d.%d' % sys.version_info[:3])"],
|
|
超时=15,
|
|
)
|
|
if 结果.code != 0 or not 结果.lines:
|
|
return ""
|
|
return 结果.lines[-1].strip()
|
|
|
|
|
|
def 首选python版本(要求: str) -> str:
|
|
"""从要求串里抠出重建 venv 要用的版本号: ">=3.11" -> "3.11"; 抠不到给 "python3"."""
|
|
命中 = re.search(r"(\d+)\.(\d+)", 要求)
|
|
if 命中 is None:
|
|
return "python3"
|
|
return f"{命中.group(1)}.{命中.group(2)}"
|
|
|
|
|
|
@dataclass
|
|
class 包要求:
|
|
"""环境.efi.json 里 packages 的一项 (声明侧)."""
|
|
|
|
名: str # pip 安装名 (如 psycopg2-binary)
|
|
要求: str # 版本约束串; 空 = 不校验
|
|
导入名: str # import 时用的名字 (给人看/将来校验用)
|
|
必需: bool # True: 缺了就阻塞; False: 只 WARN (可选依赖)
|
|
|
|
|
|
@dataclass
|
|
class 包状态:
|
|
"""一项包的核对结果 (台账一行)."""
|
|
|
|
名: str
|
|
要求: str
|
|
实装: str # 实际装上的版本; 空 = 没装
|
|
状态: str # 一致 / 缺失 / 版本不符
|
|
必需: bool
|
|
|
|
@property
|
|
def 一致(self) -> bool:
|
|
"""装上了且满足约束."""
|
|
return self.状态 == "一致"
|
|
|
|
|
|
def 读包要求(配置: dict[str, Any]) -> list[包要求]:
|
|
"""把配置里的 packages 段解析成 包要求 列表 (跳过没写 name 的脏项)."""
|
|
出: list[包要求] = []
|
|
原始 = 配置.get("packages")
|
|
for 项 in 取清单(原始):
|
|
条目 = 取对象(项)
|
|
名 = str(条目.get("name", "") or "").strip()
|
|
if not 名:
|
|
continue
|
|
出.append(
|
|
包要求(
|
|
名=名,
|
|
要求=str(条目.get("version", "") or ""),
|
|
导入名=str(条目.get("import", "") or 名),
|
|
必需=bool(条目.get("required", False)),
|
|
)
|
|
)
|
|
return 出
|
|
|
|
|
|
def 读实装包(解释器: Path) -> tuple[dict[str, str], str]:
|
|
"""读 venv 里实际装了哪些包.
|
|
|
|
参数:
|
|
解释器: venv 的 python 路径.
|
|
返回:
|
|
(规范包名 -> 版本, 数据来源说明).读不到就返回 ({}, 原因).
|
|
|
|
两条路:
|
|
1. 有 uv 就 `uv pip list --python <venv> --format json` (快, 还能读没装 pip 的 venv)
|
|
2. 没 uv 就借解释器自己的 importlib.metadata (纯 stdlib, 慢一点但一定能用)
|
|
为什么两条都要:
|
|
校网/新机器上 uv 不一定在; 而"读包状态"是引导器的核心职责, 不能因为缺 uv 就废掉.
|
|
"""
|
|
uv = shutil.which("uv")
|
|
if uv is not None:
|
|
结果 = 进程.跑命令(
|
|
[uv, "pip", "list", "--python", str(解释器), "--format", "json"], 超时=90
|
|
)
|
|
if 结果.code == 0:
|
|
表 = _解析包表("\n".join(结果.lines))
|
|
if 表 is not None:
|
|
return 表, f"uv pip list ({uv})"
|
|
记日志("WARN", f"uv pip list 失败 ({结果.detail or 结果.code}), 回落 importlib.metadata")
|
|
代码 = (
|
|
"import importlib.metadata as m, json;"
|
|
"print(json.dumps([[d.metadata['Name'] or '', d.version or ''] for d in m.distributions()]))"
|
|
)
|
|
结果 = 进程.跑命令([str(解释器), "-c", 代码], 超时=90)
|
|
if 结果.code == 0:
|
|
表 = _解析包表("\n".join(结果.lines))
|
|
if 表 is not None:
|
|
return 表, f"importlib.metadata ({解释器})"
|
|
return {}, f"读不到包列表: {结果.detail or f'退出码 {结果.code}'}"
|
|
|
|
|
|
def _解析包表(原文: str) -> dict[str, str] | None:
|
|
"""把上面两条路拿到的 json 数组统一成 {规范包名: 版本}.解析失败返回 None.
|
|
|
|
两种格式都吃: [{"name":..., "version":...}] (uv) 和 [[名字, 版本]] (importlib 那条).
|
|
"""
|
|
try:
|
|
表 = json.loads(原文.strip() or "null")
|
|
except json.JSONDecodeError:
|
|
return None
|
|
if not isinstance(表, list):
|
|
return None
|
|
出: dict[str, str] = {}
|
|
for 项 in cast(list[Any], 表):
|
|
条目 = 取对象(项)
|
|
名 = str(条目.get("name", "") or "")
|
|
if not 名:
|
|
continue
|
|
出[规范包名(名)] = str(条目.get("version", "") or "")
|
|
return 出
|
|
|
|
|
|
def 核包(要求表: list[包要求], 实装表: dict[str, str]) -> list[包状态]:
|
|
"""逐条核对: 清单里的包, 实际装上没有,版本对不对.
|
|
|
|
参数:
|
|
要求表: 配置里的 packages.
|
|
实装表: 读实装包() 的结果 (规范包名 -> 版本).
|
|
返回:
|
|
包状态 列表 (顺序跟清单一致), 状态取值: 一致 / 缺失 / 版本不符.
|
|
"""
|
|
出: list[包状态] = []
|
|
for 要 in 要求表:
|
|
实 = 实装表.get(规范包名(要.名), "")
|
|
if not 实:
|
|
情况 = "缺失"
|
|
elif 满足要求(实, 要.要求):
|
|
情况 = "一致"
|
|
else:
|
|
情况 = "版本不符"
|
|
出.append(包状态(名=要.名, 要求=要.要求, 实装=实, 状态=情况, 必需=要.必需))
|
|
return 出
|
|
|
|
|
|
# ─────────────────────────────── 体检 ───────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class 检查项:
|
|
"""体检里的一项 (表格一行)."""
|
|
|
|
序号: int
|
|
名称: str
|
|
通过: bool
|
|
阻塞: bool # 不过的时候要不要拦住"移交内核"; False = WARN 级, 只提示
|
|
详情: str # 说清楚"现在是什么情况"
|
|
修复: str = "" # 给一条能直接敲的命令; 空串 = 不用修
|
|
|
|
|
|
@dataclass
|
|
class 体检报告:
|
|
"""一次体检的全部结果.既用来打印, 也用来落 环境状态.efi.json 和 PG 台账."""
|
|
|
|
时间: str
|
|
解释器: Path # 将要跑内核的那个解释器
|
|
解释器版本: str
|
|
python要求: str
|
|
venv路径: Path
|
|
venv健康: bool
|
|
包: list[包状态]
|
|
驱动目录: Path
|
|
驱动目录通: bool
|
|
pg库: db.数据库
|
|
pg通: bool
|
|
pg详情: str
|
|
项: list[检查项]
|
|
|
|
@property
|
|
def 阻塞项(self) -> list[检查项]:
|
|
"""不过的阻塞项 (它们决定"能不能移交内核")."""
|
|
return [条 for 条 in self.项 if 条.阻塞 and not 条.通过]
|
|
|
|
@property
|
|
def 警告项(self) -> list[检查项]:
|
|
"""不过的非阻塞项 (PG 这类, 只 WARN)."""
|
|
return [条 for 条 in self.项 if not 条.阻塞 and not 条.通过]
|
|
|
|
@property
|
|
def 通过(self) -> bool:
|
|
"""可以移交内核 (阻塞项全过即可, PG 不通不拦)."""
|
|
return not self.阻塞项
|
|
|
|
@property
|
|
def 全通(self) -> bool:
|
|
"""六项全绿 (用来决定打一行摘要还是打整张表)."""
|
|
return all(条.通过 for 条 in self.项)
|
|
|
|
|
|
def 检查venv健康(venv: Path, 解释器: Path) -> tuple[bool, str, str]:
|
|
"""体检第 3 项: venv 是不是**真能用** (不只是"目录在").
|
|
|
|
参数:
|
|
venv: venv 根目录.
|
|
解释器: venv 里的 python.
|
|
返回:
|
|
(健康?, 详情, 修复命令).
|
|
|
|
查四件事:
|
|
1. `python -c "import sys;print(sys.prefix)"` 的输出必须等于 venv 绝对路径
|
|
-- 治"目录搬走后解释器还指旧路径"这个病
|
|
2. 有 pyvenv.cfg (标准 venv 的标志)
|
|
3. cfg 里的 home 指向存在的解释器 (基础解释器被删了就废了)
|
|
4. cfg 记的版本跟实际解释器版本对得上
|
|
-- 治"venv 被换过/被 uv 重建过, 元数据和实际解释器不是一套"这个病
|
|
(本项目真踩过: cfg 还写着老路径建的 3.14.4, 实际被换成 3.11.15;
|
|
只看 sys.prefix 会误判成健康)
|
|
"""
|
|
修复 = "环境 重建"
|
|
if not 解释器.exists():
|
|
return False, f"没有 {解释器}", 修复
|
|
结果 = 进程.跑命令([str(解释器), "-c", "import sys;print(sys.prefix)"], 超时=20)
|
|
if 结果.code != 0:
|
|
尾巴 = " | ".join(结果.lines[-2:]) or 结果.detail
|
|
return False, f"跑不起来 (退出码 {结果.code}): {尾巴}", 修复
|
|
前缀 = 结果.lines[-1].strip() if 结果.lines else ""
|
|
期望 = str(venv)
|
|
if 前缀 != 期望:
|
|
return False, f"sys.prefix={前缀} != {期望} (目录搬走过的典型症状: 解释器还指旧路径)", 修复
|
|
cfg = 读pyvenvcfg(venv / "pyvenv.cfg")
|
|
if not cfg:
|
|
return False, "没有 pyvenv.cfg (不是标准 venv)", 修复
|
|
家 = cfg.get("home", "")
|
|
if not 家 or not Path(家).exists():
|
|
return False, f"pyvenv.cfg 的 home={家 or '(空)'} 不存在", 修复
|
|
实际 = 读解释器版本(解释器)
|
|
cfg版本 = cfg.get("version", "") or cfg.get("version_info", "")
|
|
规整cfg = 解析版本(cfg版本)
|
|
规整实 = 解析版本(实际)
|
|
if 规整cfg and 规整实 and 规整实[: len(规整cfg)] != 规整cfg:
|
|
return (
|
|
False,
|
|
f"pyvenv.cfg 记的版本 {cfg版本} 与实际解释器 {实际} 对不上 (venv 被换过/被搬过)",
|
|
修复,
|
|
)
|
|
return True, f"sys.prefix 正确, cfg home={家}, 版本 {实际}", ""
|
|
|
|
|
|
def 读pyvenvcfg(路径: Path) -> dict[str, str]:
|
|
"""读 venv 的 pyvenv.cfg (key = value 的纯文本, 一行一条).读不到给空字典."""
|
|
出: dict[str, str] = {}
|
|
try:
|
|
原文 = 路径.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return 出
|
|
for 行 in 原文.splitlines():
|
|
if "=" not in 行:
|
|
continue
|
|
键, _, 值 = 行.partition("=")
|
|
出[键.strip()] = 值.strip()
|
|
return 出
|
|
|
|
|
|
def 检查驱动目录(路径: Path) -> tuple[bool, str]:
|
|
"""体检第 5 项: 驱动根目录在不在,能不能写.
|
|
|
|
返回:
|
|
(通过?, 详情).
|
|
|
|
为什么真写一个探针文件:
|
|
"目录存在"和"能写"是两回事 (权限/只读挂载).真建一个再删, 才敢说可读写.
|
|
为什么不自动建空目录:
|
|
驱动目录不存在 = 项目没初始化好, 这是要人看的事.自动建个空的会掩盖问题 (静默假成功).
|
|
"""
|
|
if not 路径.exists():
|
|
return False, f"{路径} 不存在 (不自动建空目录)"
|
|
if not 路径.is_dir():
|
|
return False, f"{路径} 不是目录"
|
|
探针 = 路径 / f".efi-写测试-{os.getpid()}.tmp"
|
|
try:
|
|
探针.write_text("probe", encoding="utf-8")
|
|
探针.unlink()
|
|
except OSError as 错:
|
|
return False, f"{路径} 不可写: {错}"
|
|
return True, f"{路径} 可读写"
|
|
|
|
|
|
def 检查PG(库: db.数据库) -> tuple[bool, str, str]:
|
|
"""体检第 6 项: PG 能连吗 (实例在 + 库存在 + 真连一次).
|
|
|
|
参数:
|
|
库: 从 环境.efi.json 的 db 段来的连接信息.
|
|
返回:
|
|
(通?, 详情, 修复命令).
|
|
|
|
三级降级检查 (引导器必须永远能报出个结论):
|
|
1. 有 psycopg2 -> 连 postgres 库查 pg_database (库不存在时目标库根本连不上去),
|
|
再对目标库真连一次 (光"库在"不代表权限对)
|
|
2. 没 psycopg2 但系统有 psql -> 用 psql 子进程试连, 靠报错里有没有 "does not exist" 区分
|
|
"实例没起"和"库没建" (两者修复命令不同)
|
|
3. 都没有 -> 只报 socket 文件在不在, 明确说"没法真连" (不假装通过)
|
|
"""
|
|
起库 = f"pg_ctl -D {Path(库.host).parent} start"
|
|
建库 = f"createdb -h {库.host} -p {库.port} -U {库.user} {库.name}"
|
|
if not db.有psycopg2:
|
|
psql = shutil.which("psql")
|
|
if psql is not None:
|
|
结果 = 进程.跑命令(
|
|
[psql, "-h", 库.host, "-p", str(库.port), "-U", 库.user, "-d", 库.name, "-tAc", "select 1"],
|
|
超时=15,
|
|
)
|
|
if 结果.code == 0:
|
|
return True, f"{库.描述()} 真连通过 (psql)", ""
|
|
if "does not exist" in " ".join(结果.lines):
|
|
return False, f"实例通了, 但库 {库.name} 不存在", 建库
|
|
return False, f"psql 连不上: {' | '.join(结果.lines[-2:]) or 结果.detail}", 起库
|
|
套接字 = Path(库.host) / f".s.PGSQL.{库.port}"
|
|
if 套接字.exists():
|
|
return False, f"缺 psycopg2 / psql, 只看到 socket 在 ({套接字}), 没法真连", "包 安装"
|
|
return False, f"没有 socket ({套接字}), 也没 psycopg2", 起库
|
|
try:
|
|
连接 = db.连(库, 超时=5.0, 库名="postgres")
|
|
except Exception as 错:
|
|
return False, f"连不上实例: {错}", 起库
|
|
try:
|
|
存在 = db.库存在(连接, 库.name)
|
|
except Exception as 错:
|
|
return False, f"查库存在失败: {错}", ""
|
|
finally:
|
|
连接.close()
|
|
if not 存在:
|
|
return False, f"实例通了, 但库 {库.name} 不存在", 建库
|
|
try:
|
|
连接 = db.连(库, 超时=5.0)
|
|
except Exception as 错:
|
|
return False, f"库在, 但连不上: {错}", ""
|
|
finally:
|
|
连接.close()
|
|
return True, f"{库.描述()} 真连通过", ""
|
|
|
|
|
|
def 体检(配置: dict[str, Any]) -> 体检报告:
|
|
"""跑完六项检查, 汇总成一份报告 (只读不写, 落盘由调用方决定).
|
|
|
|
参数:
|
|
配置: 读环境() 的结果.
|
|
返回:
|
|
体检报告.
|
|
|
|
六项的阻塞级别 (设计 03 §5 + §10 的分界):
|
|
1 解释器版本 阻塞 -- 解释器不对, 内核跑不起来
|
|
2 venv 存在 阻塞
|
|
3 venv 健康 阻塞
|
|
4 包状态 阻塞 (只有**必需**包缺失才阻塞; 可选包缺失只写进详情)
|
|
5 驱动目录可读写 阻塞
|
|
6 PG 能连 不阻塞 -- 引导器自己的命令要能在 PG 挂了时报出原因, 不能自己也起不来;
|
|
真需要 PG 的是内核, 它自己会报错退出
|
|
解释器取哪个:
|
|
venv 在就用 venv 的 (内核实际跑在它下面); venv 不在就用系统 python3 先判版本对不对.
|
|
"""
|
|
venv = venv路径(配置)
|
|
venv解释器 = venv / "bin" / "python"
|
|
系统python = shutil.which("python3")
|
|
系统路径 = Path(系统python) if 系统python is not None else Path(sys.executable)
|
|
用解释器 = venv解释器 if venv解释器.exists() else 系统路径
|
|
实际版本 = 读解释器版本(用解释器)
|
|
要求 = str(配置.get("python", "") or "")
|
|
项: list[检查项] = []
|
|
|
|
# 1 解释器版本
|
|
过1 = bool(实际版本) and 满足要求(实际版本, 要求)
|
|
详情1 = (
|
|
f"{用解释器} = {实际版本} (要求 {要求 or '不限'})"
|
|
if 过1
|
|
else f"需要 {要求 or '不限'}, 当前 {用解释器} = {实际版本 or '读不到版本'}"
|
|
)
|
|
项.append(检查项(1, "解释器版本", 过1, True, 详情1, "" if 过1 else f"uv python install {首选python版本(要求)}"))
|
|
|
|
# 2 venv 存在
|
|
存在 = venv解释器.exists()
|
|
项.append(
|
|
检查项(2, "venv 存在", 存在, True, str(venv解释器) if 存在 else f"没有 {venv解释器}", "" if 存在 else "环境 重建")
|
|
)
|
|
|
|
# 3 venv 健康
|
|
健康, 详情3, 修复3 = 检查venv健康(venv, venv解释器)
|
|
项.append(检查项(3, "venv 健康", 健康, True, 详情3, 修复3))
|
|
|
|
# 4 包状态 (venv 不在就没法读清单, 直接判不过)
|
|
要求表 = 读包要求(配置)
|
|
实装表: dict[str, str] = {}
|
|
来源 = "venv 不在, 没法读包"
|
|
if 存在:
|
|
实装表, 来源 = 读实装包(venv解释器)
|
|
包表 = 核包(要求表, 实装表)
|
|
缺必需 = [条 for 条 in 包表 if 条.必需 and not 条.一致]
|
|
缺可选 = [条 for 条 in 包表 if not 条.必需 and not 条.一致]
|
|
if not 要求表:
|
|
过4 = True
|
|
详情4 = "清单为空 (环境.efi.json 的 packages)"
|
|
elif not 存在:
|
|
过4 = False
|
|
详情4 = f"venv 不在, {len(要求表)} 项没法核"
|
|
elif 缺必需:
|
|
过4 = False
|
|
详情4 = "必需包不满足: " + ", ".join(
|
|
f"{条.名}(要 {条.要求 or '任意'}, 实装 {条.实装 or '无'})" for 条 in 缺必需
|
|
)
|
|
else:
|
|
过4 = True
|
|
详情4 = f"{len(包表)} 项全部一致 ({来源})"
|
|
if 过4 and 缺可选:
|
|
详情4 += "; 可选缺失: " + ", ".join(f"{条.名}({条.状态})" for 条 in 缺可选)
|
|
项.append(检查项(4, "包状态", 过4, True, 详情4, "" if 过4 else "包 安装"))
|
|
|
|
# 5 driver_root 可读写
|
|
驱动 = 驱动目录(配置)
|
|
驱动通, 详情5 = 检查驱动目录(驱动)
|
|
项.append(检查项(5, "驱动目录可读写", 驱动通, True, 详情5, "" if 驱动通 else f"mkdir -p {驱动}"))
|
|
|
|
# 6 PG 能连 (不阻塞, 见函数头的说明)
|
|
库 = db.从配置(db段(配置))
|
|
pg通, 详情6, 修复6 = 检查PG(库)
|
|
项.append(检查项(6, "PG 能连", pg通, False, 详情6, 修复6))
|
|
|
|
return 体检报告(
|
|
时间=现在(),
|
|
解释器=用解释器,
|
|
解释器版本=实际版本,
|
|
python要求=要求,
|
|
venv路径=venv,
|
|
venv健康=健康,
|
|
包=包表,
|
|
驱动目录=驱动,
|
|
驱动目录通=驱动通,
|
|
pg库=库,
|
|
pg通=pg通,
|
|
pg详情=详情6,
|
|
项=项,
|
|
)
|
|
|
|
|
|
# ─────────────────────────────── 渲染 ───────────────────────────────
|
|
|
|
|
|
def 打印体检(报告: 体检报告) -> list[str]:
|
|
"""把体检报告渲染成表格行 (不打印, 由调用方决定跟别的输出怎么拼).
|
|
|
|
结果列取值: 通过 / 不过 (阻塞项) / WARN (非阻塞项).
|
|
末尾一定会打印"判定"行 + 阻塞项清单 (带修复命令), 让老板知道下一步敲什么.
|
|
"""
|
|
表: list[list[str]] = []
|
|
for 条 in 报告.项:
|
|
结果 = "通过" if 条.通过 else ("不过" if 条.阻塞 else "WARN")
|
|
表.append([str(条.序号), 条.名称, 结果, 条.详情, 条.修复 or "—"])
|
|
行: list[str] = [f"环境体检 {报告.时间} ({引导器版本} / {内核版本})"]
|
|
行.extend(文本.表格(["#", "检查", "结果", "说明", "修复命令"], 表, [右, 左, 左, 左, 左]))
|
|
行.append("")
|
|
判定 = "通过" if 报告.通过 else "不过"
|
|
行.append(
|
|
f"判定: {判定} 阻塞项 {len(报告.阻塞项)} / 警告 {len(报告.警告项)} "
|
|
f"解释器 {报告.解释器版本 or '?'} venv {'健康' if 报告.venv健康 else '不健康'} "
|
|
f"PG {'通' if 报告.pg通 else '不通'}"
|
|
)
|
|
if 报告.阻塞项:
|
|
行.append("阻塞项不拦引导器自己的命令, 但拦移交内核 (修好再透传):")
|
|
for 条 in 报告.阻塞项:
|
|
行.append(f" #{条.序号} {条.名称}: {条.详情}" + (f" -> {条.修复}" if 条.修复 else ""))
|
|
return 行
|
|
|
|
|
|
def 摘要行(报告: 体检报告) -> str:
|
|
"""全绿时的一行摘要 (透传前打这行, 不刷整张表 -- 老板要能一眼看到"能跑")."""
|
|
包好 = sum(1 for 条 in 报告.包 if 条.一致)
|
|
return (
|
|
f"体检通过: 解释器 {报告.解释器版本} | venv 健康 | 包 {包好}/{len(报告.包)} | "
|
|
f"驱动目录 可写 | PG {'通' if 报告.pg通 else '不通(WARN)'}"
|
|
)
|
|
|
|
|
|
def 报告快照(报告: 体检报告) -> dict[str, Any]:
|
|
"""报告 -> 落盘的 json 结构 (键一律英文给机器读, 内容中文给人读).
|
|
|
|
这份结构同时写进: 环境状态.efi.json (文件副本) 和 PG 的 kernel_env 表 (台账).
|
|
一份数据两处落, 所以字段名在 db.记体检() 里也按这套键取.
|
|
"""
|
|
return {
|
|
"efi": 1,
|
|
"ts": 报告.时间,
|
|
"generator": 引导器版本,
|
|
"kernel": 内核版本,
|
|
"blocking_ok": 报告.通过, # 阻塞项全过 = 可以移交内核
|
|
"all_ok": 报告.全通, # 六项全绿
|
|
"python": {"path": str(报告.解释器), "version": 报告.解释器版本, "require": 报告.python要求},
|
|
"venv": {"path": str(报告.venv路径), "healthy": 报告.venv健康},
|
|
"packages": [
|
|
{"name": 条.名, "want": 条.要求, "got": 条.实装, "ok": 条.一致, "required": 条.必需}
|
|
for 条 in 报告.包
|
|
],
|
|
"driver_root": {"path": str(报告.驱动目录), "ok": 报告.驱动目录通},
|
|
"pg": {
|
|
"name": 报告.pg库.name,
|
|
"host": 报告.pg库.host,
|
|
"port": 报告.pg库.port,
|
|
"user": 报告.pg库.user,
|
|
"ok": 报告.pg通,
|
|
"detail": 报告.pg详情,
|
|
},
|
|
"checks": [
|
|
{
|
|
"no": 条.序号,
|
|
"name": 条.名称,
|
|
"ok": 条.通过,
|
|
"blocking": 条.阻塞,
|
|
"detail": 条.详情,
|
|
"fix": 条.修复,
|
|
}
|
|
for 条 in 报告.项
|
|
],
|
|
}
|
|
|
|
|
|
# ─────────────────────────────── 记账 (PG) ───────────────────────────────
|
|
|
|
# 进程内缓存一条 PG 连接: 一次引导器调用里要写两次台账 (体检 + 运行), 没必要连两遍.
|
|
_PG连接: dict[str, Any] = {}
|
|
|
|
|
|
def PG连接(配置: dict[str, Any]) -> tuple[Any, str]:
|
|
"""拿一条 PG 连接 (尽力而为, 不给就返回 None + 原因).
|
|
|
|
参数:
|
|
配置: 读环境() 的结果.
|
|
返回:
|
|
(连接 或 None, 人读说明).说明不管成功失败都拿来说给老板听.
|
|
|
|
为什么"失败不抛":
|
|
引导器自己的命令 (自检/环境/包) 在 PG 挂了的时候也必须能跑,能报错,能给出修复命令.
|
|
连上后顺手把引导器自己那两张台账表建好 (db.建引导器表, 幂等, 不碰驱动程序的地盘).
|
|
"""
|
|
if "连接" in _PG连接:
|
|
return _PG连接["连接"], _PG连接["说明"]
|
|
if not db.有psycopg2:
|
|
说明 = f"缺 psycopg2 (装: 包 安装) -- 引导器降级, 只写 json 快照"
|
|
_PG连接["连接"], _PG连接["说明"] = None, 说明
|
|
return None, 说明
|
|
库 = db.从配置(db段(配置))
|
|
try:
|
|
连接 = db.连(库, 超时=5.0)
|
|
except Exception as 错:
|
|
说明 = f"连不上 PG ({库.描述()}): {错}"
|
|
_PG连接["连接"], _PG连接["说明"] = None, 说明
|
|
return None, 说明
|
|
try:
|
|
db.建引导器表(连接)
|
|
说明 = f"已连 {库.描述()}"
|
|
except Exception as 错:
|
|
说明 = f"连上了 {库.描述()}, 但建引导器表失败: {错}"
|
|
_PG连接["连接"], _PG连接["说明"] = 连接, 说明
|
|
return 连接, 说明
|
|
|
|
|
|
def 写快照(报告: 体检报告) -> None:
|
|
"""写 环境状态.efi.json (必做, 原子替换; 失败会抛, 由调用方决定怎么办)."""
|
|
原子写json(状态文件, 报告快照(报告))
|
|
|
|
|
|
def PG记体检(配置: dict[str, Any], 报告: 体检报告) -> None:
|
|
"""写 kernel_env 台账一行 (尽力而为: 连不上/写不进只 WARN, 不算失败).
|
|
|
|
这条就是"引导器自己的命令 PG 降级"的落地: 体检结论永远落到 json 快照, PG 有就再记一份台账.
|
|
"""
|
|
连接, 说明 = PG连接(配置)
|
|
if 连接 is None:
|
|
记日志("WARN", f"{说明}; 只写了 {状态文件.name}")
|
|
return
|
|
try:
|
|
db.记体检(连接, 报告快照(报告))
|
|
except Exception as 错:
|
|
记日志("WARN", f"写 kernel_env 失败: {错}")
|
|
|
|
|
|
# ─────────────────────────────── 内核解释器 / 例行体检 ───────────────────────────────
|
|
|
|
|
|
def 内核解释器(配置: dict[str, Any]) -> tuple[Path, str]:
|
|
"""决定用哪个解释器跑内核.
|
|
|
|
返回:
|
|
(解释器路径, 警告文本).警告非空时调用方会打出来.
|
|
|
|
顺序: venv 的 python -> 系统 python3 -> 当前解释器 (一路降级但都明说).
|
|
为什么宁可降级也不要直接失败:
|
|
内核缺依赖会自己报错 (它连不上 PG 就退出), 引导器的职责是"把它拉起来 + 如实报告".
|
|
"""
|
|
venv解释器 = venv路径(配置) / "bin" / "python"
|
|
if venv解释器.exists():
|
|
return venv解释器, ""
|
|
系统python = shutil.which("python3")
|
|
if 系统python is not None:
|
|
return Path(系统python), f"venv 不在, 回落系统 python3 ({系统python}) -- 先 `环境 重建`"
|
|
return Path(sys.executable), f"连 python3 都没有, 回落当前解释器 ({sys.executable})"
|
|
|
|
|
|
def 例行体检(配置: dict[str, Any], 详细: bool) -> 体检报告:
|
|
"""标准三件套: 体检 -> 写快照 -> 记 PG 台账, 然后按需打印.
|
|
|
|
参数:
|
|
配置: 读环境() 的结果.
|
|
详细: True 一定打整张表; False 只在"有问题"时打表, 全绿就打一行摘要.
|
|
返回:
|
|
体检报告 (调用方拿 报告.通过 决定要不要往下走).
|
|
"""
|
|
报告 = 体检(配置)
|
|
写快照(报告)
|
|
PG记体检(配置, 报告)
|
|
if 详细 or not 报告.全通:
|
|
说(打印体检(报告))
|
|
else:
|
|
print(摘要行(报告))
|
|
return 报告
|
|
|
|
|
|
# ─────────────────────────────── 命令: 自检 / 环境 / 包 ───────────────────────────────
|
|
|
|
|
|
def 命令自检(配置: dict[str, Any]) -> int:
|
|
"""`--check` / `自检`: 只体检, 不启内核.退出码 = 有没有阻塞项 (方便脚本用)."""
|
|
报告 = 例行体检(配置, 详细=True)
|
|
return 0 if 报告.通过 else 1
|
|
|
|
|
|
def 命令环境(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`环境 [--json]` 体检; `环境 重建` 重建 venv (显式命令).
|
|
|
|
--json 走单独分支: 机器读的输出里不能混人读的表格 (否则 `环境 --json > x.json` 解析不了).
|
|
两条路都落快照 + 记台账 (体检结论永远留痕).
|
|
"""
|
|
if 参数 and 参数[0] == "重建":
|
|
return 环境重建(配置)
|
|
if "--json" in 参数:
|
|
报告 = 体检(配置)
|
|
写快照(报告)
|
|
PG记体检(配置, 报告)
|
|
print(json.dumps(报告快照(报告), ensure_ascii=False, indent=2))
|
|
return 0 if 报告.通过 else 1
|
|
报告 = 例行体检(配置, 详细=True)
|
|
return 0 if 报告.通过 else 1
|
|
|
|
|
|
def 环境重建(配置: dict[str, Any]) -> int:
|
|
"""`环境 重建`: 按 python 要求重建 venv (只在这个显式命令里动手).
|
|
|
|
步骤:
|
|
1. 旧 venv **改名留退路** (`.venv.bak-<时间戳>`), 不直接删 -- 万一新的更糟还能退回去
|
|
2. uv 在就 `uv venv --python <版本>`, 不在就回落 `python -m venv` (并 WARN)
|
|
3. 输出实时转发 (老板要看得见进度), 跑完再体检一遍
|
|
4. 有包没装 -> 只提示 `包 安装`, **不自动装** (手动控制口径)
|
|
|
|
返回:
|
|
退出码: 改名失败 1 / 建 venv 失败 1 / 体检有阻塞项 1 / 否则 0.
|
|
"""
|
|
venv = venv路径(配置)
|
|
uv = shutil.which("uv")
|
|
要求 = str(配置.get("python", "") or "")
|
|
首选 = 首选python版本(要求)
|
|
if venv.exists():
|
|
备份 = venv.with_name(f"{venv.name}.bak-{datetime.now().strftime('%Y%m%dT%H%M%S')}")
|
|
try:
|
|
os.replace(venv, 备份)
|
|
except OSError as 错:
|
|
print(f"[错误] 旧 venv 改名失败 ({错}), 不动它")
|
|
return 1
|
|
print(f"[信息] 旧 venv 已改名留退路: {备份}")
|
|
if uv is not None:
|
|
命令 = [uv, "venv", "--python", 首选, str(venv)]
|
|
else:
|
|
print(f"[WARN] 没有 uv, 回落 {sys.executable} -m venv")
|
|
命令 = [sys.executable, "-m", "venv", str(venv)]
|
|
print("$ " + " ".join(命令))
|
|
结果 = 进程.跑并转发(命令, cwd=项目根, 尾部行数=40)
|
|
print(f"退出码 {结果.code} 耗时 {结果.seconds:.1f}s")
|
|
if 结果.code != 0:
|
|
print(f"[错误] 重建失败: {' | '.join(结果.lines[-3:])}")
|
|
return 1
|
|
报告 = 例行体检(配置, 详细=True)
|
|
缺 = [条 for 条 in 报告.包 if not 条.一致]
|
|
if 缺:
|
|
print(f"还有 {len(缺)} 个包没装, 显式敲: 包 安装")
|
|
return 0 if 报告.通过 else 1
|
|
|
|
|
|
def 命令包(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`包 [--json]` 打台账表; `包 安装 [名...]` 装包 (显式命令).
|
|
|
|
返回:
|
|
退出码: 有**必需**包不满足 1 (可选包缺失不算失败) / 否则 0.
|
|
"""
|
|
if 参数 and 参数[0] == "安装":
|
|
return 包安装(参数[1:], 配置)
|
|
报告 = 体检(配置) # 这里只要包状态, 不落快照/台账 (那是 环境/自检 的活)
|
|
if "--json" in 参数:
|
|
数据 = [
|
|
{"name": 条.名, "want": 条.要求, "got": 条.实装, "ok": 条.一致, "required": 条.必需}
|
|
for 条 in 报告.包
|
|
]
|
|
print(json.dumps(数据, ensure_ascii=False, indent=2))
|
|
return 0
|
|
表 = [
|
|
[条.名, 条.要求 or "—", 条.实装 or "—", 条.状态, "必需" if 条.必需 else "可选"]
|
|
for 条 in 报告.包
|
|
]
|
|
行 = [f"包状态 venv {报告.venv路径}"]
|
|
行.extend(文本.表格(["包名", "要求", "实装", "状态", "必需性"], 表, [左, 左, 左, 左, 左]))
|
|
未满足 = [条 for 条 in 报告.包 if not 条.一致]
|
|
行.append("")
|
|
行.append(
|
|
f"共 {len(报告.包)} 项, 不满足 {len(未满足)} 项"
|
|
+ (" 显式敲: 包 安装 (引导器不自动装)" if 未满足 else "")
|
|
)
|
|
说(行)
|
|
return 1 if [条 for 条 in 报告.包 if 条.必需 and not 条.一致] else 0
|
|
|
|
|
|
def 包安装(名字: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`包 安装 [名...]`: 装包 (去掉名字 = 按配置清单全装).
|
|
|
|
命令怎么拼:
|
|
uv 在: `uv pip install --python <venv>/bin/python --index-url <国内源> <名字...>`
|
|
uv 不在: `<venv>/bin/python -m pip install -i <国内源> <名字...>`
|
|
为什么带 --index-url:
|
|
校网 GFW 下直连 PyPI 基本不通, 必须走 环境.efi.json 里配的清华源.
|
|
输出实时转发:
|
|
装包要几十秒到几分钟, 老板明确要求能看进度, 不接受黑盒等待.
|
|
|
|
返回:
|
|
退出码 = 装包命令自己的退出码 (0 成功).
|
|
"""
|
|
venv解释器 = venv路径(配置) / "bin" / "python"
|
|
if not venv解释器.exists():
|
|
print(f"[错误] venv 不在 ({venv解释器}) -- 先 `环境 重建`")
|
|
return 1
|
|
要求表 = 读包要求(配置)
|
|
if not 名字:
|
|
名字 = [条.名 for 条 in 要求表]
|
|
if not 名字:
|
|
print("[错误] 没给包名, 环境.efi.json 的 packages 也是空的")
|
|
return 1
|
|
索引 = str(配置.get("index", "") or "")
|
|
uv = shutil.which("uv")
|
|
命令: list[str]
|
|
if uv is not None:
|
|
命令 = [uv, "pip", "install", "--python", str(venv解释器)]
|
|
if 索引:
|
|
命令 += ["--index-url", 索引]
|
|
命令 += 名字
|
|
else:
|
|
print("[WARN] 没有 uv, 回落 venv 的 pip")
|
|
命令 = [str(venv解释器), "-m", "pip", "install"]
|
|
if 索引:
|
|
命令 += ["-i", 索引]
|
|
命令 += 名字
|
|
print("$ " + " ".join(命令))
|
|
结果 = 进程.跑并转发(命令, cwd=项目根, 尾部行数=60)
|
|
print(f"退出码 {结果.code} 耗时 {结果.seconds:.1f}s")
|
|
if 结果.code != 0:
|
|
print(f"[错误] 装包失败: {' | '.join(结果.lines[-3:])}")
|
|
return 结果.code
|
|
|
|
|
|
# ─────────────────────────────── 命令: 内核进程 ───────────────────────────────
|
|
|
|
|
|
def 表行(键: str, 值: str) -> list[str]:
|
|
"""两列表格的一行 (键值对输出用)."""
|
|
return [键, 值]
|
|
|
|
|
|
def 内核状态(配置: dict[str, Any]) -> int:
|
|
"""`内核 状态`: 内核进程在不在 + 上次运行 + 今日次数.
|
|
|
|
两个信息源交叉着用:
|
|
/proc 扫入口 (进程.按入口找进程) -- 判"现在在不在跑", 不依赖 PG
|
|
PG kernel_runs -- 判"上次跑了什么/多久/退出码", 以及断电残留的没收尾记录
|
|
只读不写 (收尸是 `内核 停止` 的事), 所以退出码恒 0.
|
|
"""
|
|
找 = 进程.按入口找进程(内核入口)
|
|
pid = 找[0] if 找 else None
|
|
信息 = 进程.读进程信息(pid) if pid is not None else None
|
|
连接, 连接说明 = PG连接(配置)
|
|
上次: dict[str, Any] | None = None
|
|
未结束: list[dict[str, Any]] = []
|
|
今日 = 0
|
|
if 连接 is not None:
|
|
try:
|
|
上次 = db.最近运行(连接)
|
|
未结束 = db.未结束运行(连接)
|
|
今日 = db.今日运行次数(连接)
|
|
except Exception as 错:
|
|
连接说明 += f", 查台账失败: {错}"
|
|
表: list[list[str]] = []
|
|
if pid is not None and 信息 is not None:
|
|
表.append(表行("状态", f"运行中 (pid {pid}, 进程组 {信息.pgid})"))
|
|
表.append(表行("启动时间", 本地时间(信息.启动时刻)))
|
|
表.append(表行("运行时长", 信息.存活文本))
|
|
表.append(表行("内存 / CPU", f"{信息.内存文本} / {信息.cpu秒:.1f}s"))
|
|
表.append(表行("命令行", 信息.cmdline文本))
|
|
elif pid is not None:
|
|
表.append(表行("状态", f"pid {pid} 在, 但读不到 /proc 详情"))
|
|
else:
|
|
表.append(表行("状态", "已停止"))
|
|
表.append(表行("入口", str(内核入口)))
|
|
if 上次 is not None:
|
|
起 = str(上次.get("started_at") or "")[:19]
|
|
止 = str(上次.get("finished_at") or "")[:19] or "(没结束)"
|
|
秒 = 上次.get("seconds")
|
|
秒文本 = f"{float(秒):.1f}s" if isinstance(秒, (int, float)) else "—"
|
|
表.append(
|
|
表行(
|
|
"上次运行",
|
|
f"{起} -> {止} 共 {秒文本} 退出码 {上次.get('exit_code')} "
|
|
f"{上次.get('mode') or '?'} {'ok' if 上次.get('ok') else '未标 ok'}",
|
|
)
|
|
)
|
|
详情 = str(上次.get("detail") or "")
|
|
if 详情:
|
|
表.append(表行("上次详情", 详情[:160]))
|
|
表.append(表行("今日运行", f"{今日} 次"))
|
|
if 未结束:
|
|
# 只有"记着在跑但从 /proc 看不到对应 pid"的才算残留 (正在跑的那条本来就还没收尾)
|
|
pid集 = {int(条.get("pid") or 0) for 条 in 未结束}
|
|
if pid is None or pid not in pid集:
|
|
表.append(
|
|
表行(
|
|
"收尸提示",
|
|
f"PG 里 {len(未结束)} 条运行记录没写 finished_at (断电/被杀留下) -- `内核 停止` 会收尾",
|
|
)
|
|
)
|
|
行 = ["内核进程状态"]
|
|
行.extend(文本.表格(["项", "值"], 表, [左, 左]))
|
|
行.append("")
|
|
行.append(f"台账来源: {连接说明}")
|
|
说(行)
|
|
return 0
|
|
|
|
|
|
def 跑内核(argv: list[str], 配置: dict[str, Any]) -> int:
|
|
"""移交: fork 内核进程 (argv 原样透传) -> 实时转发输出 -> 记 kernel_runs -> 返回它的退出码.
|
|
|
|
参数:
|
|
argv: 要原样交给内核的参数 (如 ["列表"] / ["扫描"] / []).
|
|
配置: 读环境() 的结果.
|
|
返回:
|
|
内核的退出码 (原样返回, 引导器不解释).
|
|
|
|
几个细节:
|
|
* 空文件检测: 内核入口如果还是 0 字节 (还没实现), 明确 WARN, 免得让人以为"跑成功了"
|
|
* PYTHONUNBUFFERED=1: 内核往管道写, 不加这个它的输出会憋到退出才一起出来 (实时进度没了)
|
|
* 记账失败只 WARN: 台账是尽力而为, 不能因为 PG 抖动就让内核跑不起来
|
|
"""
|
|
if not 内核入口.exists():
|
|
记日志("ERROR", f"内核入口不存在: {内核入口}")
|
|
return 1
|
|
if 内核入口.stat().st_size == 0:
|
|
记日志("WARN", "内核入口是空文件 (0 字节) -- 本次只验证移交通路")
|
|
解释器, 警告 = 内核解释器(配置)
|
|
if 警告:
|
|
记日志("WARN", 警告)
|
|
env = dict(os.environ)
|
|
env["PYTHONUNBUFFERED"] = "1"
|
|
命令 = [str(解释器), str(内核入口)] + argv
|
|
连接, _ = PG连接(配置)
|
|
运行id = 0
|
|
if 连接 is not None:
|
|
try:
|
|
运行id = db.记运行开始(连接, " ".join(命令), "oneshot", None)
|
|
except Exception as 错:
|
|
记日志("WARN", f"kernel_runs 写不进去: {错}")
|
|
print(f"移交内核: {' '.join(命令)}")
|
|
结果 = 进程.跑并转发(命令, cwd=项目根, env=env)
|
|
说明 = " | ".join(结果.lines[-3:]) # 失败时台账里有日志尾巴, 不吞错
|
|
if 连接 is not None and 运行id:
|
|
try:
|
|
db.记运行结束(连接, 运行id, 结果.code, 结果.seconds, 结果.code == 0, 说明)
|
|
except Exception as 错:
|
|
记日志("WARN", f"kernel_runs 写不进去: {错}")
|
|
print(f"内核退出码 {结果.code} 耗时 {结果.seconds:.1f}s")
|
|
记日志(
|
|
"INFO" if 结果.code == 0 else "WARN",
|
|
f"移交内核 退出码 {结果.code} 耗时 {结果.seconds:.1f}s 命令={' '.join(命令)}",
|
|
)
|
|
return 结果.code
|
|
|
|
|
|
def 内核启动(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`内核 启动 [--守护] [参数...]`.
|
|
|
|
两种形态:
|
|
不带 --守护 -> **前台跑一次**: 走 移交() 的路子, 输出实时可见, 内核退出就返回它的码
|
|
(内核默认动作通常是"列表", 所以这条适合"起一下看看")
|
|
带 --守护 -> **后台常驻**: 进程库 启动() detached 拉起, stdout/stderr 进
|
|
内核/logs/内核.out.log (命令输出 + 崩溃原文; 结构化日志内核自己写
|
|
内核/logs/内核.log, 见 设计/04),
|
|
记账 mode=daemon, pid 回填; 引导器不等待, 保证它活着就是全部职责
|
|
|
|
`--守护` 是从参数里**摘掉**的 (它是引导器的开关, 不转给内核), 其余参数原样透传.
|
|
幂等: 已经在跑就不重复起 (同一时间只允许一份).
|
|
"""
|
|
守护 = "--守护" in 参数
|
|
透传参数 = [条 for 条 in 参数 if 条 != "--守护"]
|
|
现有 = 进程.按入口找进程(内核入口)
|
|
if 现有:
|
|
print(f"内核已经在跑 (pid {现有[0]}) -- 幂等, 不重复起; 看状态用 `内核 状态`")
|
|
return 0
|
|
if not 守护:
|
|
return 跑内核(透传参数, 配置)
|
|
if not 内核入口.exists():
|
|
记日志("ERROR", f"内核入口不存在: {内核入口}")
|
|
return 1
|
|
解释器, 警告 = 内核解释器(配置)
|
|
if 警告:
|
|
记日志("WARN", 警告)
|
|
env = dict(os.environ)
|
|
env["PYTHONUNBUFFERED"] = "1"
|
|
# 守护模式关掉日志行的 stderr 副本: stdio 都重定向进 内核.out.log, 日志行再从 stderr 走一遍
|
|
# 就成了同一份文件里的重复内容 (内核.log 才是日志行该待的地方, 内核对这个开关有解释)
|
|
env["EFI_LOG_CONSOLE"] = "0"
|
|
命令 = [str(解释器), str(内核入口)] + 透传参数
|
|
连接, _ = PG连接(配置)
|
|
运行id = 0
|
|
if 连接 is not None:
|
|
try:
|
|
运行id = db.记运行开始(连接, " ".join(命令), "daemon", None)
|
|
except Exception as 错:
|
|
记日志("WARN", f"kernel_runs 写不进去: {错}")
|
|
print(f"[守护] {' '.join(命令)}")
|
|
print(f"[守护] 日志: {内核日志路径}")
|
|
日志.轮转(内核输出路径, 日志上限(配置), 日志保留(配置)) # 起内核前轮转它的输出文件
|
|
结果 = 进程.启动(
|
|
命令,
|
|
cwd=项目根,
|
|
日志=内核输出路径,
|
|
env=env,
|
|
入口=内核入口,
|
|
分隔=f"启动内核 {' '.join(命令)}",
|
|
)
|
|
print(f"[守护] {结果.detail}")
|
|
if 结果.ok:
|
|
记日志("INFO", f"内核 启动 --守护 pid={结果.pid} 命令={' '.join(命令)}")
|
|
else:
|
|
记日志("ERROR", f"内核 起不来: {结果.detail}")
|
|
if 连接 is not None and 运行id:
|
|
try:
|
|
if 结果.ok:
|
|
db.记内核pid(连接, 运行id, 结果.pid) # 常驻成功: 补 pid, 不收尾
|
|
else:
|
|
db.记运行结束(连接, 运行id, 结果.exit_code, 结果.seconds, False, 结果.detail)
|
|
except Exception as 错:
|
|
记日志("WARN", f"kernel_runs 写不进去: {错}")
|
|
return 0 if 结果.ok else 1
|
|
|
|
|
|
def 内核停止(配置: dict[str, Any]) -> int:
|
|
"""`内核 停止`: 按 pid 逐个停 (进程库 停止() 会校验 cmdline + 收子树), 再统一收尾台账.
|
|
|
|
为什么逐个 pid 而不是一把梭:
|
|
一次只动一个目标, 每步都有判据可查; 停完再扫一遍 /proc 确认干净 (复查才算数).
|
|
为什么要收尾台账:
|
|
断电/被 SIGKILL 的内核留下的 kernel_runs 行没有 finished_at, 会在 状态 里显示成
|
|
"还有 N 条没收尾", 这里统一补上并写明是谁收的尾.
|
|
|
|
返回:
|
|
0 (没在跑也返回 0, 幂等) / 1 (有残留或有残留进程).
|
|
"""
|
|
连接, _ = PG连接(配置)
|
|
找 = 进程.按入口找进程(内核入口)
|
|
if not 找:
|
|
print("内核没在跑 (幂等)")
|
|
if 连接 is not None:
|
|
try:
|
|
收 = db.收尾未结束(连接, "引导器停止内核时收尾")
|
|
if 收:
|
|
print(f"[收尸] 补了 {收} 条没写 finished_at 的运行记录")
|
|
except Exception as 错:
|
|
记日志("WARN", f"收尾失败: {错}")
|
|
return 0
|
|
超时 = float(配置.get("stop_timeout", 10) or 10)
|
|
失败 = 0
|
|
for pid in 找:
|
|
结果 = 进程.停止(pid, 入口=内核入口, 超时=超时)
|
|
print(f"pid {pid}: {结果.detail}")
|
|
if not 结果.ok:
|
|
失败 += 1
|
|
剩 = 进程.按入口找进程(内核入口) # 复查: 必须为空才算干净
|
|
if 剩:
|
|
print(f"[错误] 还有 {len(剩)} 个没停掉: {剩}")
|
|
失败 += 1
|
|
if 连接 is not None:
|
|
try:
|
|
收 = db.收尾未结束(连接, f"引导器停止内核 (退出: {'有残留' if 失败 else '干净'})")
|
|
if 收:
|
|
print(f"[收尸] 补了 {收} 条没写 finished_at 的运行记录")
|
|
except Exception as 错:
|
|
记日志("WARN", f"收尾失败: {错}")
|
|
结论 = "干净" if not 失败 else f"{失败} 处没停干净"
|
|
记日志("INFO" if not 失败 else "WARN", f"内核 停止: {结论}")
|
|
return 1 if 失败 else 0
|
|
|
|
|
|
def 内核重启(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`内核 重启`: 先停干净, 再启动.没停干净就**不启** (一次只动一件事, 免得留孤儿互抢)."""
|
|
停 = 内核停止(配置)
|
|
if 停 != 0:
|
|
print("[错误] 没停干净, 不接着启 (一次只动一件事)")
|
|
return 停
|
|
return 内核启动(参数, 配置)
|
|
|
|
|
|
def 打日志(路径: Path, 选: 日志.选项, 称呼: str) -> int:
|
|
"""打一份日志 (带 -n/-f/--级别/-g/--json); 引导器看的三种来源共用这一份实现."""
|
|
if 选.跟随:
|
|
print(f"[跟 {称呼}] {路径} (Ctrl+C 停)", flush=True)
|
|
return 日志.跟(路径, 选.级别, 选.关键词)
|
|
行表 = 日志.尾(路径, 选.行数, 选.级别, 选.关键词)
|
|
if not 行表:
|
|
if not 路径.exists():
|
|
print(f"还没有 {称呼} 日志: {路径}")
|
|
else:
|
|
print(f"{路径} 里没有符合条件的行 (级别 {选.级别 or '全部'} / 关键词 {选.关键词 or '无'})")
|
|
return 0
|
|
日志.打印(行表, 选.json输出)
|
|
return 0
|
|
|
|
|
|
def 内核日志(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`内核 日志 [-n N] [-f] [--级别 X] [-g 关键词] [--json] [--输出]`: 看内核的日志.
|
|
|
|
默认看**结构化日志** (内核/logs/内核.log, 一行一条固定字段); 加 `--输出` 改看
|
|
内核/logs/内核.out.log (内核进程的 stdout/stderr 原始流: 命令输出 + 崩溃原文).
|
|
|
|
参数:
|
|
参数: 命令行剩下的部分 (解析交给 日志.解析选项, 与内核那边的日志命令同一套开关).
|
|
配置: 提供 log_lines 默认行数.
|
|
返回:
|
|
0 (没有日志文件也算正常: 内核还没跑过) / 1 (参数写错).
|
|
"""
|
|
选, 问题 = 日志.解析选项(参数, int(配置.get("log_lines", 200) or 200))
|
|
if 问题:
|
|
print(f"日志参数有问题: {问题}")
|
|
print("用法: 内核 日志 [-n 200] [-f] [--级别 X] [-g 关键词] [--json] [--输出]")
|
|
return 1
|
|
return 打日志(内核输出路径 if 选.输出 else 内核日志路径, 选, "内核输出" if 选.输出 else "内核")
|
|
|
|
|
|
def 命令日志(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`日志 [选项]`: 引导器自己的动作日志 (体检 / 包 / 移交 / 内核启停).
|
|
|
|
引导器以前只打 stderr, 跑完就查不到历史; 现在每次动作都落 内核/logs/引导器.log,
|
|
这条命令就是它的查看入口 (默认看引导器自己的, --内核/--输出 转看内核那两份).
|
|
"""
|
|
选, 问题 = 日志.解析选项(参数, int(配置.get("log_lines", 200) or 200))
|
|
if 问题:
|
|
print(f"日志参数有问题: {问题}")
|
|
print(用法日志())
|
|
return 1
|
|
if 选.跟随 and (选.内核 or 选.输出):
|
|
print("-f 一次只能跟一个来源 (引导器 / --内核 / --输出)")
|
|
return 1
|
|
if 选.输出:
|
|
return 打日志(内核输出路径, 选, "内核输出")
|
|
if 选.内核:
|
|
return 打日志(内核日志路径, 选, "内核")
|
|
return 打日志(引导器日志路径, 选, "引导器")
|
|
|
|
|
|
def 用法日志() -> str:
|
|
"""`日志` 命令的用法一行 (参数写错时打)."""
|
|
return ("用法: 日志 [-n 200] [-f] [--级别 DEBUG|INFO|WARN|ERROR] [-g 关键词] "
|
|
"[--json] [--内核] [--输出]")
|
|
|
|
|
|
def 命令内核(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""`内核 ...` 的分发 (默认子命令 = 状态).
|
|
|
|
启动/重启这类"会动进程"的动作先做例行体检, 有阻塞项就拦住不让动 (跟透传一个门槛).
|
|
状态/日志/停止不做门槛: 那是"看/收尾", PG 或 venv 坏了也得能停能看.
|
|
"""
|
|
子 = 参数[0] if 参数 else "状态"
|
|
剩 = 参数[1:]
|
|
if 子 == "状态":
|
|
return 内核状态(配置)
|
|
if 子 == "日志":
|
|
return 内核日志(剩, 配置)
|
|
if 子 == "停止":
|
|
return 内核停止(配置)
|
|
if 子 in ("启动", "重启"):
|
|
报告 = 例行体检(配置, 详细=False)
|
|
if not 报告.通过:
|
|
说(打印体检(报告))
|
|
print("体检不过 -- 先修好再动内核 (引导器不自动改)")
|
|
return 1
|
|
return 内核启动(剩, 配置) if 子 == "启动" else 内核重启(剩, 配置)
|
|
print(f"[错误] 内核子命令不认识: {子}\n")
|
|
print(用法())
|
|
return 2
|
|
|
|
|
|
# ─────────────────────────────── 透传 ───────────────────────────────
|
|
|
|
|
|
def 透传(参数: list[str], 配置: dict[str, Any]) -> int:
|
|
"""把 argv 原样交给内核 (引导器只做体检门槛和记账, 不实现内核的子命令).
|
|
|
|
流程 (设计 03 §9 的启动时序):
|
|
① 例行体检 (全绿打一行摘要, 有问题打整张表)
|
|
② 阻塞项不过 -> 打印原因 + 修复命令, 退出 1 (**不自动改**)
|
|
③ 过了 -> 跑内核(参数), 内核的退出码原样返回
|
|
这里不判断 PG 通不通: PG 是内核的内存, 它自己会报错退出 (引导器不替它降级).
|
|
"""
|
|
报告 = 例行体检(配置, 详细=False)
|
|
if not 报告.通过:
|
|
说(打印体检(报告))
|
|
print("体检不过 -- 修好再移交 (引导器不自动改,不自动装,不自动重建)")
|
|
return 1
|
|
return 跑内核(参数, 配置)
|
|
|
|
|
|
# ─────────────────────────────── 用法 ───────────────────────────────
|
|
|
|
|
|
def 用法() -> str:
|
|
"""帮助文本 (-h / 子命令认不出来时打)."""
|
|
return "\n".join(
|
|
[
|
|
"用法: python3 UEFI.boot.py <命令>",
|
|
"",
|
|
" 引导器自己的命令 (顶层固定这 5 个词):",
|
|
" (无参数) 体检 -> 移交内核的默认动作 (列表)",
|
|
" --check | 自检 只体检, 不启内核 (有阻塞项退出 1)",
|
|
" 环境 [--json] 体检 6 项, 逐项给结论 + 修复命令",
|
|
" 环境 重建 按 python 要求重建 venv (显式, 旧 venv 改名留退路)",
|
|
" 包 [--json] 包状态台账 (读 uv pip list / importlib.metadata)",
|
|
" 包 安装 [名...] 装包 (显式, 走 环境.efi.json 的国内源)",
|
|
" 日志 [选项] 引导器自己的动作日志 (--内核 / --输出 转看内核那两份)",
|
|
"",
|
|
" 内核进程管理 (内核通常常驻, 生命周期归引导器):",
|
|
" 内核 状态 有没有在跑 + 上次运行 + 今日次数 (PG + /proc 交叉验证)",
|
|
" 内核 启动 [参数...] 前台跑一次, 输出实时可见",
|
|
" 内核 启动 --守护 后台常驻 (结构化日志 内核/logs/内核.log,",
|
|
" 命令输出与崩溃原文 内核/logs/内核.out.log)",
|
|
" 内核 停止 先校验 cmdline 再 SIGTERM 进程组, 不收则 SIGKILL 收残",
|
|
" 内核 重启 停干净了才启 (一次只动一件事)",
|
|
" 内核 日志 [选项] 默认看结构化日志; --输出 看命令输出/崩溃原文",
|
|
"",
|
|
" 日志选项 (引导器 日志 / 内核 日志 / 透传的 日志 都认同一套):",
|
|
" -n 200 尾部行数 -f 实时跟 (Ctrl+C 停)",
|
|
" --级别 WARN 只看这级别及以上 -g 关键词 只看含它的行",
|
|
" --json 输出 JSON Lines --全部/--内核/--引导器/--输出 换来源",
|
|
"",
|
|
" 其余命令原样透传内核 (内核自己认, 见 设计/02):",
|
|
" 列表 | 扫描 | 启动 <驱动> | 停止 <驱动> | 重启 <驱动> | 状态 | 日志 <驱动> | 事件 | 清单",
|
|
"",
|
|
" 分界: 引导器自己的命令 PG 不通只 WARN (降级到 json 快照); 透传内核的命令不降级.",
|
|
"",
|
|
" 文档: 文档/00-索引.md (用起来 01-05 / 改底座 06-10; 设计档案 设计/, 日志系统 设计/04)",
|
|
]
|
|
)
|
|
|
|
|
|
# ─────────────────────────────── 入口 ───────────────────────────────
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
"""总入口: 读配置 -> 轮转日志 -> 分发 -> 落一条收工记录.
|
|
|
|
分发规则 (顺序就是优先级, 越具体的越靠前):
|
|
-h/--help/帮助 -> 用法
|
|
版本 -> 打版本号
|
|
--check/自检 -> 只体检 (不启内核)
|
|
环境/包/内核/日志 -> 引导器自己 5 个词
|
|
其余 -> 原样透传内核
|
|
argv 空 (直接跑 UEFI.boot.py) 等于"体检 + 内核默认动作".
|
|
|
|
为什么要记"开始/收工"两条: 引导器是短命进程, 不记就等于没有历史 --
|
|
这两条是**审计对角线** (跑了什么命令 / 退出码几), 只落文件不打 stderr (不吵 stdout 的输出).
|
|
"""
|
|
配置, 警告 = 读环境()
|
|
起日志(配置) # 第一件事: 轮转引导器日志 (超上限改名 .1, 不让它无限长)
|
|
记日志("INFO", f"{引导器版本} 开始: {' '.join(argv) if argv else '(无参数)'}", 安静=True)
|
|
打印警告(警告)
|
|
码 = 分发(argv, 配置)
|
|
记日志("INFO", f"收工: 退出码 {码}", 安静=True)
|
|
return 码
|
|
|
|
|
|
def 分发(argv: list[str], 配置: dict[str, Any]) -> int:
|
|
"""命令分发 (main 只管开始/收工两条日志, 具体分支都在这)."""
|
|
if not argv:
|
|
return 透传([], 配置)
|
|
头 = argv[0]
|
|
if 头 in ("-h", "--help", "帮助", "help"):
|
|
print(用法())
|
|
return 0
|
|
if 头 == "版本":
|
|
print(f"{引导器版本} / {内核版本}")
|
|
return 0
|
|
if 头 in ("--check", "自检", "检查"):
|
|
return 命令自检(配置)
|
|
if 头 == "环境":
|
|
return 命令环境(argv[1:], 配置)
|
|
if 头 == "包":
|
|
return 命令包(argv[1:], 配置)
|
|
if 头 == "内核":
|
|
return 命令内核(argv[1:], 配置)
|
|
if 头 == "日志":
|
|
return 命令日志(argv[1:], 配置)
|
|
return 透传(argv, 配置)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main(sys.argv[1:]))
|
|
except KeyboardInterrupt:
|
|
print("\n[中断]")
|
|
sys.exit(130)
|