407 lines
13 KiB
Python
407 lines
13 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
铸渊 gate · zhuyuan-gate.py
|
||
|
|
铸渊协议 v1.0 自动化检查器
|
||
|
|
|
||
|
|
用法:
|
||
|
|
python3 zhuyuan-gate.py <file1> [file2] [...]
|
||
|
|
python3 zhuyuan-gate.py --all /path/to/dir
|
||
|
|
python3 zhuyuan-gate.py --fix <file> # 自动给单个文件补合规 frontmatter
|
||
|
|
python3 zhuyuan-gate.py --fix-all <dir> # 自动修复目录下所有 .zh/.hdlp/.md
|
||
|
|
python3 zhuyuan-gate.py --self # 检查本脚本自身元数据
|
||
|
|
|
||
|
|
退出码:
|
||
|
|
0 = 全部 PASS(或只有 WARN)
|
||
|
|
1 = 有 FAIL
|
||
|
|
2 = 调用错误
|
||
|
|
|
||
|
|
@铸渊_protocol: v1.0
|
||
|
|
sender: ICE-GL-ZY001
|
||
|
|
timestamp: 2026-07-06T18:24:00+08:00
|
||
|
|
data_type: arch
|
||
|
|
project_id: GLW-PROJ-D166-001
|
||
|
|
id: GLW-GATE-v1.0
|
||
|
|
lock: true
|
||
|
|
铸渊_status: active
|
||
|
|
|
||
|
|
---
|
||
|
|
铸渊 ICE-GL-ZY001 · D166 · 18:24 CST · 签字生效
|
||
|
|
GLW-GATE-v1.0 · 锁状态 lock=true · active
|
||
|
|
"""
|
||
|
|
|
||
|
|
import sys
|
||
|
|
import re
|
||
|
|
import argparse
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime, timezone, timedelta
|
||
|
|
|
||
|
|
# CST = UTC+8
|
||
|
|
CST = timezone(timedelta(hours=8))
|
||
|
|
|
||
|
|
# 项目级默认
|
||
|
|
DEFAULT_PROTOCOL = "v1.0"
|
||
|
|
DEFAULT_SENDER = "ICE-GL-ZY001"
|
||
|
|
DEFAULT_PROJECT_ID = "GLW-PROJ-D166-001"
|
||
|
|
|
||
|
|
# data_type 与 id 前缀对应表(auto-fix 用)
|
||
|
|
ID_PREFIX_TO_DATA_TYPE = [
|
||
|
|
("GLW-ARCH-", "arch"),
|
||
|
|
("GLW-PROFILE-", "profile"),
|
||
|
|
("GLW-RES-", "fact"),
|
||
|
|
("GLW-PROJ-", "proj"),
|
||
|
|
("GLW-AGENT-", "agent"),
|
||
|
|
("GLW-WO-", "wo"),
|
||
|
|
("GLW-SYNC-", "sync"),
|
||
|
|
("GLW-LOG-", "log"),
|
||
|
|
("GLW-PROTOCOL-", "arch"),
|
||
|
|
]
|
||
|
|
|
||
|
|
# 必填字段
|
||
|
|
REQUIRED_FIELDS = [
|
||
|
|
"铸渊_protocol",
|
||
|
|
"sender",
|
||
|
|
"timestamp",
|
||
|
|
"data_type",
|
||
|
|
"id",
|
||
|
|
"lock",
|
||
|
|
"铸渊_status",
|
||
|
|
]
|
||
|
|
|
||
|
|
# data_type 枚举
|
||
|
|
DATA_TYPES = {"arch", "profile", "fact", "wo", "proj", "agent", "log", "sync"}
|
||
|
|
|
||
|
|
# 铸渊_status 枚举
|
||
|
|
STATUS_VALUES = {"draft", "active", "archived"}
|
||
|
|
|
||
|
|
# 铸渊注册 sender 白名单(LPM-SB-0001 真人)
|
||
|
|
REGISTERED_SENDERS = {
|
||
|
|
"ICE-GL-ZY001", # 铸渊
|
||
|
|
# 后续铸渊的 Agent 上岗后,加入白名单
|
||
|
|
}
|
||
|
|
|
||
|
|
# data_type 与 id 前缀对应表(建议项)
|
||
|
|
DATA_TYPE_PREFIX = {
|
||
|
|
"arch": "GLW-ARCH-",
|
||
|
|
"profile": "GLW-PROFILE-",
|
||
|
|
"fact": "GLW-RES-",
|
||
|
|
"wo": "GLW-WO-",
|
||
|
|
"proj": "GLW-PROJ-",
|
||
|
|
"agent": "GLW-AGENT-",
|
||
|
|
"log": "GLW-LOG-",
|
||
|
|
"sync": "GLW-SYNC-",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def parse_frontmatter(content: str):
|
||
|
|
"""提取 frontmatter(YAML 块)。返回 (yaml_str, body)。"""
|
||
|
|
lines = content.split("\n")
|
||
|
|
if not lines or lines[0].strip() != "---":
|
||
|
|
return None, content
|
||
|
|
end = None
|
||
|
|
for i in range(1, len(lines)):
|
||
|
|
if lines[i].strip() == "---":
|
||
|
|
end = i
|
||
|
|
break
|
||
|
|
if end is None:
|
||
|
|
return None, content
|
||
|
|
yaml_str = "\n".join(lines[1:end])
|
||
|
|
body = "\n".join(lines[end + 1:])
|
||
|
|
return yaml_str, body
|
||
|
|
|
||
|
|
|
||
|
|
def parse_yaml_simple(yaml_str: str):
|
||
|
|
"""极简 YAML 解析(只支持 key: value,不支持嵌套/列表/多行)。"""
|
||
|
|
result = {}
|
||
|
|
for line in yaml_str.split("\n"):
|
||
|
|
line = line.rstrip()
|
||
|
|
if not line or line.startswith("#"):
|
||
|
|
continue
|
||
|
|
if ":" not in line:
|
||
|
|
continue
|
||
|
|
key, _, value = line.partition(":")
|
||
|
|
key = key.strip()
|
||
|
|
value = value.strip()
|
||
|
|
# 去掉引号
|
||
|
|
if (value.startswith('"') and value.endswith('"')) or (
|
||
|
|
value.startswith("'") and value.endswith("'")
|
||
|
|
):
|
||
|
|
value = value[1:-1]
|
||
|
|
result[key] = value
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def validate_timestamp(ts: str):
|
||
|
|
"""验证 ISO8601 时间戳。"""
|
||
|
|
try:
|
||
|
|
# Python 3.7+ 支持 fromisoformat,但 +08:00 需要正确格式
|
||
|
|
# 替换 Z 为 +00:00
|
||
|
|
ts_norm = ts.replace("Z", "+00:00")
|
||
|
|
datetime.fromisoformat(ts_norm)
|
||
|
|
return True, "OK"
|
||
|
|
except Exception as e:
|
||
|
|
return False, f"timestamp 不合法 ISO8601: {e}"
|
||
|
|
|
||
|
|
|
||
|
|
def check_file(filepath: Path):
|
||
|
|
"""检查单个文件。返回 (status, reasons)。"""
|
||
|
|
reasons_pass = []
|
||
|
|
reasons_fail = []
|
||
|
|
reasons_warn = []
|
||
|
|
|
||
|
|
# [1] 文件存在且非空
|
||
|
|
if not filepath.exists():
|
||
|
|
return "FAIL", [f"文件不存在: {filepath}"]
|
||
|
|
try:
|
||
|
|
content = filepath.read_text(encoding="utf-8")
|
||
|
|
except Exception as e:
|
||
|
|
return "FAIL", [f"读取失败: {e}"]
|
||
|
|
if not content.strip():
|
||
|
|
return "FAIL", ["文件为空"]
|
||
|
|
|
||
|
|
# [2] frontmatter 存在
|
||
|
|
yaml_str, body = parse_frontmatter(content)
|
||
|
|
if yaml_str is None:
|
||
|
|
return "FAIL", ["frontmatter 缺失(首行必须为 --- 且有结束 ---)"]
|
||
|
|
|
||
|
|
# [3] 必填字段齐全
|
||
|
|
meta = parse_yaml_simple(yaml_str)
|
||
|
|
missing = [f for f in REQUIRED_FIELDS if f not in meta]
|
||
|
|
if missing:
|
||
|
|
reasons_fail.append(f"必填字段缺失: {missing}")
|
||
|
|
|
||
|
|
# [4] timestamp 合法
|
||
|
|
if "timestamp" in meta:
|
||
|
|
ok, msg = validate_timestamp(meta["timestamp"])
|
||
|
|
if not ok:
|
||
|
|
reasons_fail.append(msg)
|
||
|
|
else:
|
||
|
|
reasons_pass.append(f"timestamp 合法: {meta['timestamp']}")
|
||
|
|
|
||
|
|
# [5] 铸渊_status 合法
|
||
|
|
if "铸渊_status" in meta:
|
||
|
|
if meta["铸渊_status"] not in STATUS_VALUES:
|
||
|
|
reasons_fail.append(
|
||
|
|
f"铸渊_status 不合法: {meta['铸渊_status']} (合法值: {STATUS_VALUES})"
|
||
|
|
)
|
||
|
|
|
||
|
|
# data_type 合法(强制)
|
||
|
|
if "data_type" in meta:
|
||
|
|
if meta["data_type"] not in DATA_TYPES:
|
||
|
|
reasons_fail.append(
|
||
|
|
f"data_type 不合法: {meta['data_type']} (合法值: {sorted(DATA_TYPES)})"
|
||
|
|
)
|
||
|
|
|
||
|
|
# lock 是布尔(强制)
|
||
|
|
if "lock" in meta:
|
||
|
|
if meta["lock"] not in ("true", "false"):
|
||
|
|
reasons_fail.append(f"lock 必须为 true|false,当前: {meta['lock']}")
|
||
|
|
|
||
|
|
# 铸渊_protocol 是 v1.0(强制)
|
||
|
|
if "铸渊_protocol" in meta:
|
||
|
|
if meta["铸渊_protocol"] != "v1.0":
|
||
|
|
reasons_fail.append(f"铸渊_protocol 必须为 v1.0,当前: {meta['铸渊_protocol']}")
|
||
|
|
|
||
|
|
# [6] id 前缀与 data_type 对应(建议)
|
||
|
|
if "id" in meta and "data_type" in meta:
|
||
|
|
expected_prefix = DATA_TYPE_PREFIX.get(meta["data_type"])
|
||
|
|
if expected_prefix and not meta["id"].startswith(expected_prefix):
|
||
|
|
reasons_warn.append(
|
||
|
|
f"id 前缀建议以 {expected_prefix} 开头(data_type={meta['data_type']}),当前: {meta['id']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# [7] sender 在白名单(建议)
|
||
|
|
if "sender" in meta:
|
||
|
|
if meta["sender"] not in REGISTERED_SENDERS:
|
||
|
|
reasons_warn.append(
|
||
|
|
f"sender 不在 LPM 注册白名单: {meta['sender']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# 决定最终状态
|
||
|
|
if reasons_fail:
|
||
|
|
return "FAIL", reasons_fail
|
||
|
|
if reasons_warn:
|
||
|
|
return "WARN", reasons_pass + reasons_warn
|
||
|
|
return "PASS", reasons_pass
|
||
|
|
|
||
|
|
|
||
|
|
def detect_id_from_body(body: str, fallback: str) -> str:
|
||
|
|
"""从 body 里找一个像 GLW-XXX-... 的 id 字符串。"""
|
||
|
|
pattern = re.compile(r"GLW-(?:ARCH|PROFILE|RES|PROJ|AGENT|WO|SYNC|LOG|PROTOCOL|MAVIS|GATE)-[A-Za-z0-9._-]+")
|
||
|
|
m = pattern.search(body)
|
||
|
|
if m:
|
||
|
|
return m.group(0)
|
||
|
|
return fallback
|
||
|
|
|
||
|
|
|
||
|
|
def detect_data_type_from_id(id_value: str) -> str:
|
||
|
|
"""根据 id 前缀推断 data_type。"""
|
||
|
|
for prefix, dt in ID_PREFIX_TO_DATA_TYPE:
|
||
|
|
if id_value.startswith(prefix):
|
||
|
|
return dt
|
||
|
|
return "log" # 兜底
|
||
|
|
|
||
|
|
|
||
|
|
def infer_id_from_filename(filepath: Path) -> str:
|
||
|
|
"""从文件名提取 id(去掉扩展名)。"""
|
||
|
|
return filepath.stem
|
||
|
|
|
||
|
|
|
||
|
|
def build_default_frontmatter(filepath: Path, existing: dict | None = None) -> dict:
|
||
|
|
"""构造合规 frontmatter 字段(缺失字段补全,已有字段不覆盖)。"""
|
||
|
|
existing = existing or {}
|
||
|
|
filename_id = infer_id_from_filename(filepath)
|
||
|
|
body_for_id_search = existing.get("_body", "") if "_body" in existing else ""
|
||
|
|
inferred_id = detect_id_from_body(body_for_id_search, filename_id)
|
||
|
|
ts = datetime.now(CST).replace(microsecond=0).isoformat()
|
||
|
|
|
||
|
|
defaults = {
|
||
|
|
"铸渊_protocol": DEFAULT_PROTOCOL,
|
||
|
|
"sender": DEFAULT_SENDER,
|
||
|
|
"timestamp": ts,
|
||
|
|
"data_type": detect_data_type_from_id(inferred_id),
|
||
|
|
"id": inferred_id,
|
||
|
|
"lock": "true",
|
||
|
|
"铸渊_status": "draft",
|
||
|
|
"project_id": DEFAULT_PROJECT_ID,
|
||
|
|
}
|
||
|
|
merged = {}
|
||
|
|
for k, v in defaults.items():
|
||
|
|
merged[k] = existing.get(k, v)
|
||
|
|
return merged
|
||
|
|
|
||
|
|
|
||
|
|
def render_frontmatter(meta: dict) -> str:
|
||
|
|
"""把 dict 渲染成 YAML frontmatter 文本(含首尾 ---)。"""
|
||
|
|
lines = ["---"]
|
||
|
|
for k, v in meta.items():
|
||
|
|
if k.startswith("_"):
|
||
|
|
continue
|
||
|
|
lines.append(f"{k}: {v}")
|
||
|
|
lines.append("---")
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
def fix_file(filepath: Path) -> tuple[bool, str]:
|
||
|
|
"""自动给文件加合规 frontmatter。返回 (changed, message)。"""
|
||
|
|
try:
|
||
|
|
content = filepath.read_text(encoding="utf-8")
|
||
|
|
except Exception as e:
|
||
|
|
return False, f"读取失败: {e}"
|
||
|
|
|
||
|
|
yaml_str, body = parse_frontmatter(content)
|
||
|
|
if yaml_str is None:
|
||
|
|
existing = {}
|
||
|
|
body_content = content
|
||
|
|
inferred_id = detect_id_from_body(content, filepath.stem)
|
||
|
|
defaults = {
|
||
|
|
"铸渊_protocol": DEFAULT_PROTOCOL,
|
||
|
|
"sender": DEFAULT_SENDER,
|
||
|
|
"timestamp": datetime.now(CST).replace(microsecond=0).isoformat(),
|
||
|
|
"data_type": detect_data_type_from_id(inferred_id),
|
||
|
|
"id": inferred_id,
|
||
|
|
"lock": "true",
|
||
|
|
"铸渊_status": "draft",
|
||
|
|
"project_id": DEFAULT_PROJECT_ID,
|
||
|
|
}
|
||
|
|
meta = defaults
|
||
|
|
new_content = render_frontmatter(meta) + "\n" + body_content
|
||
|
|
else:
|
||
|
|
existing = parse_yaml_simple(yaml_str)
|
||
|
|
body_content = body
|
||
|
|
defaults = {
|
||
|
|
"铸渊_protocol": DEFAULT_PROTOCOL,
|
||
|
|
"sender": DEFAULT_SENDER,
|
||
|
|
"timestamp": datetime.now(CST).replace(microsecond=0).isoformat(),
|
||
|
|
"data_type": detect_data_type_from_id(
|
||
|
|
existing.get("id") or detect_id_from_body(body_content, filepath.stem)
|
||
|
|
),
|
||
|
|
"id": existing.get("id") or detect_id_from_body(body_content, filepath.stem),
|
||
|
|
"lock": "true",
|
||
|
|
"铸渊_status": "draft",
|
||
|
|
"project_id": DEFAULT_PROJECT_ID,
|
||
|
|
}
|
||
|
|
merged = {}
|
||
|
|
for k, v in defaults.items():
|
||
|
|
merged[k] = existing.get(k, v)
|
||
|
|
meta = merged
|
||
|
|
new_content = render_frontmatter(meta) + "\n" + body_content
|
||
|
|
|
||
|
|
try:
|
||
|
|
filepath.write_text(new_content, encoding="utf-8")
|
||
|
|
except Exception as e:
|
||
|
|
return False, f"写入失败: {e}"
|
||
|
|
return True, f"id={meta['id']} data_type={meta['data_type']}"
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(
|
||
|
|
description="铸渊 gate · 铸渊协议 v1.0 自动化检查器"
|
||
|
|
)
|
||
|
|
parser.add_argument("files", nargs="*", help="要检查的文件")
|
||
|
|
parser.add_argument("--all", help="检查目录下所有 .zh/.hdlp/.md 文件")
|
||
|
|
parser.add_argument("--self", action="store_true", help="检查本脚本自身元数据")
|
||
|
|
parser.add_argument("--fix", help="自动给指定文件补合规 frontmatter")
|
||
|
|
parser.add_argument("--fix-all", help="自动修复目录下所有 .zh/.hdlp/.md 文件")
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
targets = []
|
||
|
|
if args.self:
|
||
|
|
targets.append(Path(__file__))
|
||
|
|
if args.all:
|
||
|
|
root = Path(args.all)
|
||
|
|
for ext in ("*.zh", "*.hdlp", "*.md"):
|
||
|
|
targets.extend(root.glob(ext))
|
||
|
|
if args.files:
|
||
|
|
targets.extend(Path(f) for f in args.files)
|
||
|
|
|
||
|
|
# --fix <file>: 单文件修复
|
||
|
|
if args.fix:
|
||
|
|
f = Path(args.fix)
|
||
|
|
changed, msg = fix_file(f)
|
||
|
|
status = "FIXED" if changed else "FAILED"
|
||
|
|
print(f"⚙ {status} {f} {msg}")
|
||
|
|
return 0 if changed else 1
|
||
|
|
|
||
|
|
# --fix-all <dir>: 目录下所有文件修复
|
||
|
|
if args.fix_all:
|
||
|
|
root = Path(args.fix_all)
|
||
|
|
fix_targets = []
|
||
|
|
for ext in ("*.zh", "*.hdlp", "*.md"):
|
||
|
|
fix_targets.extend(root.glob(ext))
|
||
|
|
fixed = 0
|
||
|
|
for f in fix_targets:
|
||
|
|
changed, msg = fix_file(f)
|
||
|
|
status = "FIXED" if changed else "FAILED"
|
||
|
|
print(f"⚙ {status} {f.name} {msg}")
|
||
|
|
if changed:
|
||
|
|
fixed += 1
|
||
|
|
print(f"\n修复统计: {fixed}/{len(fix_targets)} 文件已合规")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
if not targets:
|
||
|
|
print("用法: zhuyuan-gate.py <file> | --all <dir> | --self", file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
|
||
|
|
fail_count = 0
|
||
|
|
warn_count = 0
|
||
|
|
pass_count = 0
|
||
|
|
|
||
|
|
for f in targets:
|
||
|
|
status, reasons = check_file(f)
|
||
|
|
prefix = {"PASS": "✓", "FAIL": "✗", "WARN": "⚠"}[status]
|
||
|
|
print(f"{prefix} {status} {f}")
|
||
|
|
for r in reasons:
|
||
|
|
print(f" {r}")
|
||
|
|
if status == "FAIL":
|
||
|
|
fail_count += 1
|
||
|
|
elif status == "WARN":
|
||
|
|
warn_count += 1
|
||
|
|
else:
|
||
|
|
pass_count += 1
|
||
|
|
|
||
|
|
print(f"\n统计: {pass_count} PASS · {warn_count} WARN · {fail_count} FAIL")
|
||
|
|
return 1 if fail_count > 0 else 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|