106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b: 把 modules/world_sync/models/ 下的表定义真源同步到应用打包镜像落点。
|
||
|
||
铁律：apps/{app}/pkgs/{module}/models/*.json 是 build 阶段产物，禁止手工双写。
|
||
本脚本是唯一允许的写入通道 —— 真源只有一份（modules/ 侧），镜像由 cp 语义逐字节覆盖生成。
|
||
|
||
用法：
|
||
python3 modules/world_sync/scripts/sync_models_mirror.py
|
||
python3 modules/world_sync/scripts/sync_models_mirror.py --check # 只校验不写，不一致退出码 1
|
||
"""
|
||
|
||
import argparse
|
||
import hashlib
|
||
import os
|
||
import shutil
|
||
import sys
|
||
|
||
# 表定义真源清单（相对机构工作空间根）
|
||
SOURCE_TABLES = [
|
||
"pbl_runtime_event.json",
|
||
"pbl_entity_state.json",
|
||
]
|
||
|
||
MODULE_ROOT = "modules/world_sync"
|
||
SOURCE_DIR = os.path.join(MODULE_ROOT, "models")
|
||
MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")
|
||
|
||
|
||
def _find_workspace_root():
|
||
"""从脚本位置向上回溯，找到同时含 modules/ 与 apps/ 的工作空间根。"""
|
||
here = os.path.dirname(os.path.abspath(__file__))
|
||
cur = here
|
||
for _ in range(10):
|
||
if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")):
|
||
return cur
|
||
parent = os.path.dirname(cur)
|
||
if parent == cur:
|
||
break
|
||
cur = parent
|
||
# 兜底：脚本位于 modules/{m}/scripts/x.py，则根为上三级
|
||
return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir))
|
||
|
||
|
||
def sha256_of(path):
|
||
h = hashlib.sha256()
|
||
with open(path, "rb") as fh:
|
||
for chunk in iter(lambda: fh.read(65536), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
|
||
def sync(root, check_only=False):
|
||
"""同步真源 -> 镜像。返回 (是否全部一致, 明细列表)。"""
|
||
src_dir = os.path.join(root, SOURCE_DIR)
|
||
dst_dir = os.path.join(root, MIRROR_DIR)
|
||
details = []
|
||
all_ok = True
|
||
|
||
if not check_only:
|
||
os.makedirs(dst_dir, exist_ok=True)
|
||
|
||
for name in SOURCE_TABLES:
|
||
src = os.path.join(src_dir, name)
|
||
dst = os.path.join(dst_dir, name)
|
||
if not os.path.isfile(src):
|
||
details.append("MISSING-SOURCE: %s" % src)
|
||
all_ok = False
|
||
continue
|
||
|
||
if not check_only:
|
||
shutil.copyfile(src, dst)
|
||
|
||
src_hash = sha256_of(src)
|
||
if not os.path.isfile(dst):
|
||
details.append("MISSING-MIRROR: %s" % dst)
|
||
all_ok = False
|
||
continue
|
||
dst_hash = sha256_of(dst)
|
||
same = (src_hash == dst_hash)
|
||
all_ok = all_ok and same
|
||
details.append("%s %s" % ("SAME" if same else "DIFF", name))
|
||
details.append(" sha256 src=%s" % src_hash)
|
||
details.append(" sha256 dst=%s" % dst_hash)
|
||
|
||
return all_ok, details
|
||
|
||
|
||
def main(argv=None):
|
||
parser = argparse.ArgumentParser(description="sync world_sync table-definition sources to app mirror")
|
||
parser.add_argument("--check", action="store_true", help="only verify, do not write")
|
||
args = parser.parse_args(argv)
|
||
|
||
root = _find_workspace_root()
|
||
ok, details = sync(root, check_only=args.check)
|
||
print("workspace root: %s" % root)
|
||
print("mode: %s" % ("check" if args.check else "sync"))
|
||
for line in details:
|
||
print(line)
|
||
print("RESULT: %s" % ("PASS" if ok else "FAIL"))
|
||
return 0 if ok else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|