81 lines
5.0 KiB
Python
81 lines
5.0 KiB
Python
# ============================================================
|
||
# 可重生模块注册表（moduleId: core-registry, 可重生 ✅）
|
||
# plan.md §9.4/§6.10.3：每个源文件头部的 "moduleId: xxx, 可重生" 声明
|
||
# 即注册表数据源——本模块扫描源码目录，产出重生中心的模块清单。
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import os # 目录遍历
|
||
import re # 模块头解析
|
||
from typing import Any # 类型标注
|
||
|
||
# 扫描根（相对仓库根）：后端 / 前端源码 / 契约
|
||
_SCAN_ROOTS = ["server", "apps/web/src", "shared"]
|
||
# 参与扫描的扩展名
|
||
_EXTS = {".py", ".ts", ".tsx", ".css", ".json", ".html"}
|
||
# 模块头声明的匹配模式：moduleId: xxx（可选"可重生"标记）
|
||
_HEAD_RE = re.compile(r"moduleId:\s*([\w-]+)(.{0,20}?可重生)?")
|
||
|
||
|
||
def _layer_of(path: str) -> str:
|
||
"""按路径猜测模块所属层（注册表的分组维度）。"""
|
||
p = path.replace("\\", "/") # 统一分隔符
|
||
if p.startswith("server/engines"): return "engine" # 引擎层
|
||
if p.startswith("server/agent_core"): return "agent-core" # 智能体核心
|
||
if p.startswith("server/aps_domain"): return "domain" # 领域层
|
||
if p.startswith("server/state"): return "state" # 数据层
|
||
if p.startswith("server/gateway") or p.endswith("server/main.py"): return "gateway" # 接入层
|
||
if p.startswith("apps/web"): return "ui" # 表现层
|
||
if p.startswith("shared"): return "contract" # 契约层
|
||
return "other" # 兜底
|
||
|
||
|
||
def scan_modules(repo_root: str = ".") -> list[dict[str, Any]]:
|
||
"""扫描源码，收集全部带 moduleId 声明的模块（P0 只读；重生中心数据源）。"""
|
||
modules: list[dict[str, Any]] = [] # 结果集
|
||
for root in _SCAN_ROOTS: # 逐扫描根
|
||
base = os.path.join(repo_root, root) # 绝对化
|
||
if not os.path.isdir(base): # 目录缺失跳过
|
||
continue
|
||
for dirpath, dirnames, filenames in os.walk(base): # 递归遍历
|
||
dirnames[:] = [d for d in dirnames if d not in ("node_modules", "dist", "__pycache__", "data")] # 剪枝
|
||
for fn in filenames: # 逐文件
|
||
if os.path.splitext(fn)[1] not in _EXTS: # 非目标扩展名跳过
|
||
continue
|
||
path = os.path.join(dirpath, fn) # 文件路径
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f: # 只读文件头
|
||
head = f.read(2048) # 模块头声明约定在前 2KB 内
|
||
except OSError: # 读失败跳过（不致命）
|
||
continue
|
||
m = _HEAD_RE.search(head) # 匹配声明
|
||
if not m: # 无声明 → 非注册模块
|
||
continue
|
||
rel = os.path.relpath(path, repo_root).replace("\\", "/") # 相对路径
|
||
modules.append({ # 注册表条目（ModuleManifest 的扫描态子集）
|
||
"moduleId": m.group(1), # 模块 ID
|
||
"path": rel, # 源文件路径
|
||
"layer": _layer_of(rel), # 所属层
|
||
"regenerable": bool(m.group(2)), # 是否声明可重生
|
||
})
|
||
modules.sort(key=lambda x: (x["layer"], x["moduleId"])) # 稳定排序（分组展示友好）
|
||
return modules # 返回清单
|
||
|
||
|
||
def verify_audit_chain(events: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""校验审计哈希链完整性（门禁管理台"防线状态"的校验按钮，§6.10.2）。
|
||
|
||
重算每个事件的哈希并核对前驱衔接；返回 {ok, checked, brokenAt}。
|
||
"""
|
||
import hashlib, json # 局部导入（仅校验用）
|
||
prev = "GENESIS" # 链头哨兵（与 write_audit 一致）
|
||
for i, ev in enumerate(events): # 顺序遍历
|
||
if ev.get("prevHash") != prev: # 前驱衔接断裂
|
||
return {"ok": False, "checked": i, "brokenAt": ev.get("id")}
|
||
body = {k: v for k, v in ev.items() if k != "hash"} # 剥离哈希字段还原签名体
|
||
payload = prev + json.dumps(body, ensure_ascii=False, sort_keys=True) # 重建签名载荷
|
||
if hashlib.sha256(payload.encode("utf-8")).hexdigest() != ev.get("hash"): # 哈希不符 → 被篡改
|
||
return {"ok": False, "checked": i, "brokenAt": ev.get("id")}
|
||
prev = ev["hash"] # 推进链头
|
||
return {"ok": True, "checked": len(events), "brokenAt": None} # 全链完好
|