99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b 表定义镜像同步器：modules/world_sync/models/ 是唯一真源，
|
||
apps/scense/pkgs/world_sync/models/ 是打包镜像（禁止手工双写）。
|
||
|
||
用法:
|
||
python modules/world_sync/scripts/sync_models_to_app.py # 同步镜像
|
||
python modules/world_sync/scripts/sync_models_to_app.py --check # 只校验，不写
|
||
|
||
行为:
|
||
1. 逐个把真源表定义 JSON 以「规范化的同一份字节」写入镜像落点（覆盖式）；
|
||
2. 打印两两 sha256 比对结果；--check 模式下若不一致退出码 1。
|
||
|
||
这样 QC 抽检任一落点都拿到同一权威文本，消除「多份互相矛盾定义」。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
# 本文件位于 modules/world_sync/scripts/，向上三级 = 机构工作空间根
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
WS_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
||
|
||
SRC_DIR = os.path.join("modules", "world_sync", "models")
|
||
DST_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")
|
||
|
||
# 只同步 M11b-2 交付的两张表定义（其余 world_sync*.json 属历史既有文件，不在本任务范围）
|
||
TABLES = ("pbl_runtime_event.json", "pbl_entity_state.json")
|
||
|
||
|
||
def sha256_of(path: str) -> str:
|
||
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 canonical_bytes(path: str) -> bytes:
|
||
"""读入 JSON 后按统一方式序列化，保证真源与镜像字节级一致（消除缩进/空格差异）。"""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
return (json.dumps(data, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--check", action="store_true", help="只比对不写入")
|
||
args = ap.parse_args()
|
||
|
||
src_dir = os.path.join(WS_ROOT, SRC_DIR)
|
||
dst_dir = os.path.join(WS_ROOT, DST_DIR)
|
||
if not os.path.isdir(src_dir):
|
||
print(f"FAIL: 真源目录不存在 {src_dir}", file=sys.stderr)
|
||
return 1
|
||
os.makedirs(dst_dir, exist_ok=True)
|
||
|
||
bad = 0
|
||
for name in TABLES:
|
||
src = os.path.join(src_dir, name)
|
||
dst = os.path.join(dst_dir, name)
|
||
blob = canonical_bytes(src)
|
||
if args.check:
|
||
same = os.path.isfile(dst) and open(dst, "rb").read() == blob
|
||
print(f"[check] {name}: mirror={'OK' if same else 'DRIFT'}")
|
||
if not same:
|
||
bad += 1
|
||
continue
|
||
with open(dst, "wb") as fh:
|
||
fh.write(blob)
|
||
# 写完立刻自证：真源规范化后与镜像 sha256 必须相同
|
||
s_src, s_dst = sha256_of(src), sha256_of(dst)
|
||
if s_src != s_dst:
|
||
# 真源本身不是规范化字节时，把规范化结果同时回写两处，保证两两相同
|
||
with open(src, "wb") as fh:
|
||
fh.write(blob)
|
||
s_src, s_dst = sha256_of(src), sha256_of(dst)
|
||
print(f"[sync ] {name}")
|
||
print(f" src sha256 = {s_src} ({os.path.join(SRC_DIR, name)})")
|
||
print(f" dst sha256 = {s_dst} ({os.path.join(DST_DIR, name)})")
|
||
print(f" MATCH = {s_src == s_dst}")
|
||
if s_src != s_dst:
|
||
bad += 1
|
||
|
||
if bad:
|
||
print(f"RESULT: FAIL ({bad} 个镜像不一致)")
|
||
return 1
|
||
print("RESULT: OK (真源 modules/world_sync/models -> 镜像 apps/scense/pkgs/world_sync/models 已对齐)")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|