TermiNexus
b083d27165
- codec_fields：module_87072::T5 记为单字段 bytes23；pq.tankState 的 TypeCodec 由 85183::jq 改指 87072::T5 - 生成器支持 bytes<N>（ByteArray / R.ReadRaw(N) / W.WriteRaw），重生成 Module87072、Module85183 - 审计链：lib_space_codec 新增 bytes<N>；audit_pt_boundaries / audit_bit_budget 支持 --body/--plain；semantic 暴露 Out.Bits - 新增工具：audit_bit_budget（nullmap 位预算）、probe_pt_record（pt 项级）、analyze_item_trace（客户端项级打点对照）、dump_codec_src - wss_server 新增 --dump-plain：把解密+解压后的裸帧体落到 plain/，供离线审计 - 镜像打点：opt 全量位位置、pt 项级字节/位区间、TankCC 逐字段区间；重新打包 unpacked.pack.json
541 lines
23 KiB
Python
541 lines
23 KiB
Python
#!/usr/bin/env python3
|
||
# GenerateProtocol.py —— 从 3DTank/names 的协议表生成 Kotlin 数据类与 Codec
|
||
#
|
||
# 输入（全部由 3DTank/tools 的提取器产出，不要手改）：
|
||
# names/codec_fields.json 655 个 codec 的字段（类型已解析成 bool/i32/opt<...>/list<...>/模型名）
|
||
# names/downlink_table.json 681 条下行方法（methodId + 有序字段）
|
||
# names/uplink_table.json 361 条上行方法（methodId + 负载字段）
|
||
# names/method_table.json methodId → modelId（用来给下行命令取语义类名）
|
||
#
|
||
# 输出（入库、可手改；重跑会覆盖）：
|
||
# net/src/jsMain/kotlin/Net/Generated/Models/Module<id>.kt
|
||
# net/src/jsMain/kotlin/Net/Generated/Downlink/Module<id>.kt
|
||
# net/src/jsMain/kotlin/Net/Generated/Uplink/Module<id>.kt
|
||
# net/src/jsMain/kotlin/Net/Generated/Registry.kt
|
||
# net/src/jsMain/kotlin/Net/Generated/Metadata.kt
|
||
#
|
||
# 命名规则：全部 PascalCase（首字母大写，常见缩写大写），属性名只影响可读性，不影响线格式。
|
||
import json
|
||
import re
|
||
import shutil
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
NAMES = ROOT.parent / "3DTank" / "names"
|
||
OUT = ROOT / "net" / "src" / "jsMain" / "kotlin" / "Net" / "Generated"
|
||
|
||
# 这些 codec 的线形不能由字段表直译，改由 Net/Protocol 下手写（数据类仍由生成器产出）。
|
||
# module_85183::jq = TankState 的「字段表版」；真正上线的是 module_87072 注册的 T5（23 字节位打包）
|
||
OVERLAY_CODECS = {"module_85183::jq"}
|
||
OVERLAY_NAMES = {"module_85183::jq": "TankStateCodec"}
|
||
|
||
# ModelService 的 4 条容器命令：布局由 Net/Protocol/Containers.kt 手写，直接覆盖注册表
|
||
REGISTRY_OVERRIDES = {
|
||
"9038756763225879447": "LtCodec", # _unloadObjects
|
||
"7640916300855664666": "PtCodec", # _loadObjectsData
|
||
"3216143066888387731": "ZtCodec", # _loadDependencies
|
||
"8179162314198150518": "AliasCodec", # _setModelCommandAlias
|
||
}
|
||
|
||
PRIMS = {
|
||
"bool": ("Boolean", "R.ReadBool()", "W.WriteBool({})"),
|
||
"i8": ("Byte", "R.ReadByte()", "W.WriteByte({}.toInt())"),
|
||
"i16": ("Short", "R.ReadShort()", "W.WriteShort({}.toInt())"),
|
||
"i32": ("Int", "R.ReadInt()", "W.WriteInt({})"),
|
||
"i64": ("Long", "R.ReadLong()", "W.WriteLong({})"),
|
||
"f32": ("Float", "R.ReadFloat()", "W.WriteFloat({})"),
|
||
"f64": ("Double", "R.ReadDouble()", "W.WriteDouble({})"),
|
||
"string": ("String", "R.ReadString()", "W.WriteString({})"),
|
||
"bytes": ("ByteArray", "R.ReadBytes()", "W.WriteBytes({})"),
|
||
"array": ("ByteArray", "R.ReadBytes()", "W.WriteBytes({})"),
|
||
"vector3": ("Vector3", "R.ReadVector3()", "W.WriteVector3({})"),
|
||
}
|
||
|
||
ACRONYMS = {"id", "uid", "sid", "cid", "cc", "ccu", "url", "uri", "ip", "api", "gpu", "fps", "hud", "db", "hp", "xp"}
|
||
KOTLIN_KEYWORDS = {
|
||
"as", "break", "class", "continue", "do", "else", "false", "for", "fun", "if", "in",
|
||
"interface", "is", "null", "object", "package", "return", "super", "this", "throw",
|
||
"true", "try", "typealias", "typeof", "val", "var", "when", "while", "by", "catch",
|
||
"constructor", "delegate", "dynamic", "field", "file", "finally", "get", "import",
|
||
"init", "param", "property", "receiver", "set", "setparam", "where", "it",
|
||
}
|
||
|
||
|
||
def ToPascal(Name: str) -> str:
|
||
"""camelCase / snake_case / 缩写 → PascalCase（缩写大写）。"""
|
||
if not Name:
|
||
return "Field"
|
||
Parts = [P for P in re.split(r"[^0-9A-Za-z]+", Name) if P]
|
||
if not Parts:
|
||
return "Field"
|
||
Out = []
|
||
for Part in Parts:
|
||
Low = Part.lower()
|
||
if Low in ACRONYMS:
|
||
Out.append(Low.upper())
|
||
elif Part.isupper() and len(Part) > 1:
|
||
Out.append(Part)
|
||
else:
|
||
Out.append(Part[0].upper() + Part[1:])
|
||
Result = "".join(Out)
|
||
if Result and Result[0].isdigit():
|
||
Result = "F" + Result
|
||
return Result
|
||
|
||
|
||
def KotlinName(Name: str) -> str:
|
||
return f"`{Name}`" if Name in KOTLIN_KEYWORDS else Name
|
||
|
||
|
||
def Txt(Node) -> str:
|
||
Kind = Node[0]
|
||
if Kind == "model":
|
||
return "#" + Node[1]
|
||
if Kind in ("opt", "list", "enum"):
|
||
return f"{Kind}<{Txt(Node[1])}>"
|
||
if Kind == "map":
|
||
return f"map<{Txt(Node[1])},{Txt(Node[2])}>"
|
||
return Kind
|
||
|
||
|
||
def SplitTopLevel(Text: str):
|
||
Depth = 0
|
||
for I, C in enumerate(Text):
|
||
if C == "<":
|
||
Depth += 1
|
||
elif C == ">":
|
||
Depth -= 1
|
||
elif C == "," and Depth == 0:
|
||
return Text[:I].strip(), Text[I + 1:].strip()
|
||
raise ValueError(f"map 参数不合法：{Text!r}")
|
||
|
||
|
||
class Resolver:
|
||
"""把协议表里的类型串解析成类型节点；TypeCodec 是裸类名跨模块时的唯一可靠指针。"""
|
||
|
||
def __init__(self, CodecKeyToName):
|
||
self.CodecKeyToName = CodecKeyToName
|
||
self.NameSet = set(CodecKeyToName.values())
|
||
|
||
def Parse(self, Text: str, TypeCodec: str = ""):
|
||
S = (Text or "").strip()
|
||
if S.startswith("opt<") and S.endswith(">"):
|
||
return ("opt", self.Parse(S[4:-1], TypeCodec))
|
||
if S.startswith("list<") and S.endswith(">"):
|
||
return ("list", self.Parse(S[5:-1], TypeCodec))
|
||
if S.startswith("enum<") and S.endswith(">"):
|
||
return ("enum", self.Parse(S[5:-1], TypeCodec))
|
||
if S.startswith("map<") and S.endswith(">"):
|
||
K, V = SplitTopLevel(S[4:-1])
|
||
return ("map", self.Parse(K, TypeCodec), self.Parse(V, TypeCodec))
|
||
# 定长二进制块 bytes<N>：出处 module_87072.js:53876 的 T5.t7h ——
|
||
# TankState 在线上是 23 字节打包姿态（内部按 19/13/13/13 位量化 4 个向量），
|
||
# 我们只按定长吃掉、原样写回，不解释内容。
|
||
Fixed = re.fullmatch(r"bytes(\d+)", S)
|
||
if Fixed:
|
||
return ("bytes", int(Fixed.group(1)))
|
||
if S in PRIMS:
|
||
return (S,)
|
||
if TypeCodec and TypeCodec in self.CodecKeyToName:
|
||
return ("model", self.CodecKeyToName[TypeCodec])
|
||
if S in self.NameSet:
|
||
return ("model", S)
|
||
raise ValueError(f"未知类型 {S!r}（TypeCodec={TypeCodec!r}）")
|
||
|
||
|
||
def KotlinType(Node) -> str:
|
||
Kind = Node[0]
|
||
if Kind == "opt":
|
||
return KotlinType(Node[1]) + "?"
|
||
if Kind == "list":
|
||
return "List<" + KotlinType(Node[1]) + ">"
|
||
if Kind == "map":
|
||
return f"Map<{KotlinType(Node[1])}, {KotlinType(Node[2])}>"
|
||
if Kind == "enum":
|
||
return "Int"
|
||
if Kind == "bytes":
|
||
return "ByteArray"
|
||
if Kind == "model":
|
||
return Node[1]
|
||
return PRIMS[Kind][0]
|
||
|
||
|
||
def DecodeExpr(Node) -> str:
|
||
Kind = Node[0]
|
||
if Kind == "opt":
|
||
return "Wire.ReadOpt(R, N) { " + DecodeExpr(Node[1]) + " }"
|
||
if Kind == "list":
|
||
return "Wire.ReadList(R) { " + DecodeExpr(Node[1]) + " }"
|
||
if Kind == "map":
|
||
return "Wire.ReadMap(R, { " + DecodeExpr(Node[1]) + " }, { " + DecodeExpr(Node[2]) + " })"
|
||
if Kind == "enum":
|
||
# 枚举宽度在数据上自相矛盾，暂按 i32（最优口径）：
|
||
# · 源码：EnumCodecInfo（vn → Kt）读写都是空实现 → 0 字节；
|
||
# · 实测 A：EmissionColorCC 的 4×i32 正好到帧尾，threshold 这个 enum 必须 0 字节；
|
||
# · 实测 B：Item3DCC 之外有 21 个模型在 enum=0 时整体错位（P_ 的 category 必须 4 字节）。
|
||
# 全量审计：enum=4 → 6 处未命中 / 127 项；enum=0 → 21 处。先保留 4，等客户端位痕迹定案。
|
||
return "R.ReadInt()"
|
||
if Kind == "bytes":
|
||
return f"R.ReadRaw({Node[1]})"
|
||
if Kind == "model":
|
||
return Node[1] + "Codec.Decode(R, N)"
|
||
return PRIMS[Kind][1]
|
||
|
||
|
||
def EncodeExpr(Node, Value: str) -> str:
|
||
Kind = Node[0]
|
||
if Kind == "opt":
|
||
return "Wire.WriteOpt(N, " + Value + ") { " + EncodeExpr(Node[1], "it") + " }"
|
||
if Kind == "list":
|
||
return "Wire.WriteList(W, " + Value + ") { " + EncodeExpr(Node[1], "it") + " }"
|
||
if Kind == "map":
|
||
return ("Wire.WriteMap(W, " + Value + ", { " + EncodeExpr(Node[1], "it") + " }, { " +
|
||
EncodeExpr(Node[2], "it") + " })")
|
||
if Kind == "enum":
|
||
return "W.WriteInt(" + Value + ")"
|
||
if Kind == "bytes":
|
||
return "W.WriteRaw(" + Value + ")"
|
||
if Kind == "model":
|
||
return Node[1] + "Codec.Encode(W, N, " + Value + ")"
|
||
return PRIMS[Kind][2].replace("{}", Value)
|
||
|
||
|
||
def FieldList(Items):
|
||
Out = []
|
||
for I, F in enumerate(Items, 1):
|
||
Out.append({
|
||
"Order": F.get("Order") or I,
|
||
"Name": F.get("Name") or "",
|
||
"Type": F.get("Type") or "?",
|
||
"TypeCodec": F.get("TypeCodec") or "",
|
||
})
|
||
return Out
|
||
|
||
|
||
def UniqueFieldNames(Fields):
|
||
Used = {}
|
||
Out = []
|
||
for F in Fields:
|
||
Base = ToPascal(F["Name"]) if F["Name"] and F["Name"] not in ("?", "(匿名)") else f"Field{F['Order']}"
|
||
Base = KotlinName(Base)
|
||
if Base in Used:
|
||
Used[Base] += 1
|
||
Base = f"{Base}{Used[Base]}"
|
||
else:
|
||
Used[Base] = 1
|
||
Out.append(Base)
|
||
return Out
|
||
|
||
|
||
class ClassDef:
|
||
def __init__(self, Name, Fields, Module="", Comment="", CodecKey=""):
|
||
self.Name = Name
|
||
self.Fields = Fields
|
||
self.Module = Module
|
||
self.Comment = Comment
|
||
self.CodecKey = CodecKey
|
||
|
||
|
||
def Main():
|
||
CodecFields = json.loads((NAMES / "codec_fields.json").read_text("utf-8"))
|
||
Downlink = json.loads((NAMES / "downlink_table.json").read_text("utf-8"))
|
||
Uplink = json.loads((NAMES / "uplink_table.json").read_text("utf-8"))
|
||
MethodTable = json.loads((NAMES / "method_table.json").read_text("utf-8"))
|
||
|
||
Codecs = CodecFields.get("Codecs", {})
|
||
|
||
# ---------- 0. 运行时权威绑定（若已采集）----------
|
||
# codec_fields 里的 ModelId/ModelIds 是「短类名 + codec 槽位名」指纹 + 源码登记点推出来的；
|
||
# 短类名在 126 个模块里大量复用，误绑会让 pt 项静默串位（最后读出 modelId=8 这种不存在值）。
|
||
# names/model_binding.json 来自客户端运行时 ModelReg.n7o 注册采样 + u7h 源码反查模块，
|
||
# 一旦存在就以它为准：只保留它给出的 modelId（其余 codec 在运行时没注册，ByModelId 里也不该有）。
|
||
BindingPath = NAMES / "model_binding.json"
|
||
if BindingPath.exists():
|
||
Binding = json.loads(BindingPath.read_text("utf-8")).get("Entries") or []
|
||
BindingOverride = {}
|
||
for Row in Binding:
|
||
BindingOverride.setdefault(Row["Key"], []).append(str(Row["ModelId"]))
|
||
Dropped = 0
|
||
for Key, C in Codecs.items():
|
||
Ids = BindingOverride.get(Key)
|
||
if Ids:
|
||
C["ModelId"] = Ids[0]
|
||
C["ModelIds"] = Ids
|
||
else:
|
||
if C.get("ModelId") or C.get("ModelIds"):
|
||
Dropped += 1
|
||
C["ModelId"] = None
|
||
C["ModelIds"] = []
|
||
print(f"[绑定] 采用运行时权威绑定：{len(Binding)} 条 modelId / {len(BindingOverride)} 个 codec，"
|
||
f"丢弃 {Dropped} 个无运行时注册的旧绑定", file=sys.stderr)
|
||
|
||
# ---------- 1. 给每个 codec 分配唯一类名 ----------
|
||
CodecKeyToName = {}
|
||
NameOwner = {}
|
||
ModelDefs = {}
|
||
for Key, C in Codecs.items():
|
||
Base = ToPascal(C.get("ModelName") or f"{C.get('ModuleId','0')}_{C.get('Codec','X')}")
|
||
Name = Base
|
||
if Name in NameOwner and NameOwner[Name] != Key:
|
||
Name = f"{Base}M{C.get('ModuleId','0')}{ToPascal(C.get('Codec','X'))}"
|
||
NameOwner[Name] = Key
|
||
CodecKeyToName[Key] = Name
|
||
ModelDefs[Key] = ClassDef(Name, FieldList(C.get("Fields") or []), C.get("Module", ""),
|
||
C.get("ModelName") or "codec " + Key, Key)
|
||
|
||
# ---------- 2. 解析模型字段（TypeCodec 已覆盖跨模块裸类名；解析不了的如实列出） ----------
|
||
Resolved = {} # codecKey → (ClassDef, Nodes)
|
||
Failed = {} # codecKey → 原因
|
||
R = Resolver(CodecKeyToName)
|
||
for Key, Def in ModelDefs.items():
|
||
try:
|
||
Nodes = [R.Parse(F["Type"], F["TypeCodec"]) for F in Def.Fields]
|
||
except ValueError as E:
|
||
Failed[Key] = str(E)
|
||
continue
|
||
Resolved[Key] = (Def, Nodes)
|
||
# 依赖级联：引用了「解析失败的模型」的模型也一并跳过，保证生成物一定能编译
|
||
def ModelRefs(Node):
|
||
Kind = Node[0]
|
||
if Kind == "model":
|
||
return {Node[1]}
|
||
if Kind in ("opt", "list", "enum"):
|
||
return ModelRefs(Node[1])
|
||
if Kind == "map":
|
||
return ModelRefs(Node[1]) | ModelRefs(Node[2])
|
||
return set()
|
||
|
||
while True:
|
||
FailedNames = {CodecKeyToName[K] for K in Failed}
|
||
Cascaded = []
|
||
for Key, (Def, Nodes) in Resolved.items():
|
||
Refs = set()
|
||
for N in Nodes:
|
||
Refs |= ModelRefs(N)
|
||
if Refs & FailedNames:
|
||
Cascaded.append((Key, "引用了解析失败的模型 " + ", ".join(sorted(Refs & FailedNames))))
|
||
if not Cascaded:
|
||
break
|
||
for Key, Why in Cascaded:
|
||
Failed[Key] = Why
|
||
Resolved.pop(Key, None)
|
||
|
||
Active = {K: V for K, V in CodecKeyToName.items() if K in Resolved}
|
||
|
||
# ---------- 3. 下行/上行命令 ----------
|
||
ModelIdToName = {}
|
||
for Key, C in Codecs.items():
|
||
if Key not in Resolved:
|
||
continue
|
||
for Mid in C.get("ModelIds") or []:
|
||
if C.get("ModelName"):
|
||
ModelIdToName[str(Mid)] = CodecKeyToName[Key]
|
||
MethodToModelId = {str(Row["MethodId"]): str(Row["ModelId"]) for Row in MethodTable.get("Rows", [])}
|
||
|
||
R = Resolver(Active)
|
||
SigToName = {}
|
||
for _, (Def, Nodes) in Resolved.items():
|
||
Sig = "|".join(Txt(N) for N in Nodes)
|
||
SigToName.setdefault(Sig, Def.Name)
|
||
NameUsed = set(CodecKeyToName.values())
|
||
|
||
def AllocName(Base, Sig, Suffix):
|
||
if Sig in SigToName:
|
||
return SigToName[Sig], False
|
||
Name = Base
|
||
if Name in NameUsed:
|
||
Name = f"{Base}{Suffix}"
|
||
NameUsed.add(Name)
|
||
SigToName[Sig] = Name
|
||
return Name, True
|
||
|
||
DownDefs = []
|
||
DownRegistry = [] # (MethodId, className)
|
||
for Cmd in Downlink.get("Commands", []):
|
||
Fields = FieldList(Cmd.get("Fields") or [])
|
||
Mid = str(Cmd["MethodId"])
|
||
try:
|
||
Nodes = [R.Parse(F["Type"], F["TypeCodec"]) for F in Fields]
|
||
except ValueError as E:
|
||
print(f"[跳过] 下行 {Mid} {Cmd.get('OwnerClass')}：{E}", file=sys.stderr)
|
||
continue
|
||
Sig = "|".join(Txt(N) for N in Nodes)
|
||
Base = ModelIdToName.get(MethodToModelId.get(Mid, ""), "")
|
||
if not Base:
|
||
Base = f"Down{Cmd.get('Module','').replace('module_','')}{ToPascal(Cmd.get('OwnerClass') or 'Cmd')}{Mid[-4:]}"
|
||
Name, IsNew = AllocName(Base, Sig, Mid[-4:])
|
||
if IsNew:
|
||
DownDefs.append(ClassDef(Name, Fields, Cmd.get("Module", ""), "下行 " + Mid))
|
||
DownRegistry.append((Mid, Name))
|
||
|
||
UpDefs = []
|
||
UpRegistry = []
|
||
for Cmd in Uplink.get("Commands", []):
|
||
Fields = FieldList(Cmd.get("Fields") or [])
|
||
Mid = str(Cmd["MethodId"])
|
||
try:
|
||
Nodes = [R.Parse(F["Type"], F["TypeCodec"]) for F in Fields]
|
||
except ValueError as E:
|
||
print(f"[跳过] 上行 {Mid} {Cmd.get('OwnerClass')}：{E}", file=sys.stderr)
|
||
continue
|
||
Sig = "|".join(Txt(N) for N in Nodes)
|
||
Base = f"Up{Cmd.get('Module','').replace('module_','')}{ToPascal(Cmd.get('OwnerClass') or 'Cmd')}"
|
||
Name, IsNew = AllocName(Base, Sig, Mid[-4:])
|
||
if IsNew:
|
||
UpDefs.append(ClassDef(Name, Fields, Cmd.get("Module", ""), "上行 " + Mid))
|
||
UpRegistry.append((Mid, Name))
|
||
|
||
# ---------- 4. 写文件 ----------
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
for Sub in ("Models", "Downlink", "Uplink"):
|
||
Target = OUT / Sub
|
||
if Target.exists():
|
||
shutil.rmtree(Target)
|
||
Target.mkdir(parents=True, exist_ok=True)
|
||
|
||
Header = ("// 本文件由 TankiOffline/tools/GenerateProtocol.py 生成。\n"
|
||
"// 允许手改；重跑生成器会覆盖。类型口径见 3DTank/docs/KtRebuild_Plan.md。\n"
|
||
"package Net.Generated\n\n"
|
||
"import Core.ByteReader\nimport Core.ByteWriter\nimport Core.NullmapCursor\n"
|
||
"import Core.NullmapWriter\nimport Core.Vector3\n"
|
||
"import Net.Wire.ModelCodec\nimport Net.Wire.Wire\nimport Net.Wire.ErasedCodec\n"
|
||
"import Net.Protocol.TankStateCodec\nimport Net.Protocol.LtCodec\n"
|
||
"import Net.Protocol.PtCodec\nimport Net.Protocol.ZtCodec\nimport Net.Protocol.AliasCodec\n\n")
|
||
|
||
def EmitClass(Def: ClassDef, Nodes):
|
||
Names = UniqueFieldNames(Def.Fields)
|
||
Lines = [f"// {Def.Comment}｜{Def.Module}"]
|
||
if Nodes:
|
||
Lines.append(f"data class {Def.Name}(")
|
||
for N, P in zip(Nodes, Names):
|
||
Lines.append(f" val {P}: {KotlinType(N)},")
|
||
Lines.append(")")
|
||
else:
|
||
Lines.append(f"class {Def.Name}")
|
||
Lines.append("")
|
||
if Def.CodecKey in OVERLAY_CODECS:
|
||
Lines.append("// 该模型的 codec 由 Net/Protocol 下手写（线形不是字段直译），此处不生成。")
|
||
Lines.append("")
|
||
return Lines
|
||
Lines.append(f"object {Def.Name}Codec : ModelCodec<{Def.Name}> {{")
|
||
Lines.append(f" override fun Decode(R: ByteReader, N: NullmapCursor): {Def.Name} {{")
|
||
for N, P in zip(Nodes, Names):
|
||
Lines.append(f" val {P} = {DecodeExpr(N)}")
|
||
if Nodes:
|
||
Lines.append(f" return {Def.Name}({', '.join(Names)})")
|
||
else:
|
||
Lines.append(f" return {Def.Name}()")
|
||
Lines.append(" }")
|
||
Lines.append("")
|
||
Lines.append(f" override fun Encode(W: ByteWriter, N: NullmapWriter, V: {Def.Name}) {{")
|
||
for N, P in zip(Nodes, Names):
|
||
Lines.append(f" {EncodeExpr(N, 'V.' + P)}")
|
||
Lines.append(" }")
|
||
Lines.append("}")
|
||
Lines.append("")
|
||
return Lines
|
||
|
||
def GroupByModule(Items):
|
||
Out = {}
|
||
for Def, Nodes in Items:
|
||
Out.setdefault(Def.Module or "Misc", []).append((Def, Nodes))
|
||
return Out
|
||
|
||
for Group, Items in GroupByModule(Resolved.values()).items():
|
||
Path_ = OUT / "Models" / f"{ToPascal(Group.replace('module_', 'Module'))}.kt"
|
||
Path_.write_text(Header + "\n".join(sum((EmitClass(D, N) for D, N in Items), [])), "utf-8")
|
||
for Group, Items in GroupByModule([(D, None) for D in DownDefs]).items():
|
||
Path_ = OUT / "Downlink" / f"{ToPascal(Group.replace('module_', 'Module'))}.kt"
|
||
Path_.write_text(Header + "\n".join(sum((EmitClass(D, [R.Parse(F['Type'], F['TypeCodec']) for F in D.Fields])
|
||
for D, _ in Items), [])), "utf-8")
|
||
for Group, Items in GroupByModule([(D, None) for D in UpDefs]).items():
|
||
Path_ = OUT / "Uplink" / f"{ToPascal(Group.replace('module_', 'Module'))}.kt"
|
||
Path_.write_text(Header + "\n".join(sum((EmitClass(D, [R.Parse(F['Type'], F['TypeCodec']) for F in D.Fields])
|
||
for D, _ in Items), [])), "utf-8")
|
||
|
||
# ---------- 5. 注册表 ----------
|
||
Emitted = {D.Name for D, _ in Resolved.values()} | {D.Name for D in DownDefs} | {D.Name for D in UpDefs}
|
||
|
||
def Registry(Entries, Title):
|
||
Lines = [f"object {Title} {{", " val ByMethodId: Map<Long, ErasedCodec> = mapOf("]
|
||
Seen = set()
|
||
for Mid, Name in Entries:
|
||
if Name not in Emitted or Mid in Seen or Mid in REGISTRY_OVERRIDES:
|
||
continue
|
||
Seen.add(Mid)
|
||
Lines.append(f" {Mid}L to {Name}Codec,")
|
||
if Title == "DownlinkRegistry":
|
||
for Mid, Expr in REGISTRY_OVERRIDES.items():
|
||
Lines.append(f" {Mid}L to {Expr},")
|
||
Lines.append(" )")
|
||
Lines.append("}")
|
||
return Lines
|
||
|
||
def RegistryWithModelIds(ResolvedMap, NameMap):
|
||
Lines = ["object ModelRegistry {", " val ByModelId: Map<Long, ErasedCodec> = mapOf("]
|
||
Seen = set()
|
||
|
||
def Emit(OnlySource: bool):
|
||
for Key, (Def, Nodes) in ResolvedMap.items():
|
||
C = Codecs[Key]
|
||
HasSource = bool(C.get("ModelHow"))
|
||
if OnlySource != HasSource:
|
||
continue
|
||
Expr = OVERLAY_NAMES.get(Key, NameMap[Key] + "Codec")
|
||
Ids = []
|
||
if C.get("ModelId"):
|
||
Ids.append(str(C["ModelId"]))
|
||
Ids += [str(M) for M in (C.get("ModelIds") or [])]
|
||
for Mid in Ids:
|
||
if Mid in Seen:
|
||
continue
|
||
Seen.add(Mid)
|
||
Lines.append(f" {Mid}L to {Expr},")
|
||
|
||
# 先信「源码登记点」（ModelHow 非空），再用指纹/旧 join 兜底：
|
||
# 同一个 modelId 可能被两个 codec 认领（如 2108103923322474513 = CaptchaCC / DonationData），
|
||
# 源码登记点才是真正上线的那一个。
|
||
Emit(True)
|
||
Emit(False)
|
||
Lines.append(" )")
|
||
Lines.append("}")
|
||
return Lines
|
||
|
||
(OUT / "Registry.kt").write_text(
|
||
Header + "\n".join(Registry(DownRegistry, "DownlinkRegistry")) + "\n\n" +
|
||
"\n".join(Registry(UpRegistry, "UplinkRegistry")) + "\n\n" +
|
||
"\n".join(RegistryWithModelIds(Resolved, CodecKeyToName)) + "\n", "utf-8")
|
||
|
||
# ---------- 6. 元数据 ----------
|
||
def MetaEntry(Def: ClassDef, Nodes):
|
||
Mid = Def.Comment.split()[-1]
|
||
FS = ", ".join(f'ProtocolFieldMeta({F["Order"]}, "{F["Name"]}", "{Txt(N)}")'
|
||
for F, N in zip(Def.Fields, Nodes))
|
||
return f' {Mid}L to ProtocolCommandMeta({Mid}L, "{Def.Name}", "{Def.Module}", listOf({FS})),'
|
||
|
||
def NodesOf(Def):
|
||
return [R.Parse(F["Type"], F["TypeCodec"]) for F in Def.Fields]
|
||
|
||
Meta = ["package Net.Generated", "",
|
||
"data class ProtocolFieldMeta(val Order: Int, val Name: String, val WireType: String)", "",
|
||
"data class ProtocolCommandMeta(val MethodId: Long, val Name: String, val Module: String, val Fields: List<ProtocolFieldMeta>)", "",
|
||
"object ProtocolMetadata {", " val Downlink: Map<Long, ProtocolCommandMeta> = mapOf("]
|
||
Meta += [MetaEntry(D, NodesOf(D)) for D in DownDefs]
|
||
Meta += [" )", " val Uplink: Map<Long, ProtocolCommandMeta> = mapOf("]
|
||
Meta += [MetaEntry(D, NodesOf(D)) for D in UpDefs]
|
||
Meta += [" )", "}"]
|
||
(OUT / "Metadata.kt").write_text("\n".join(Meta) + "\n", "utf-8")
|
||
|
||
print(f"模型类：{len(Resolved)} / {len(Codecs)}（跳过 {len(Failed)}）")
|
||
print(f"下行命令：{len(DownDefs)} 个新类，注册 {len({M for M, N in DownRegistry if N in Emitted})} 条")
|
||
print(f"上行命令：{len(UpDefs)} 个新类，注册 {len({M for M, N in UpRegistry if N in Emitted})} 条")
|
||
if Failed:
|
||
print("跳过的 codec：")
|
||
for Key, Why in Failed.items():
|
||
print(f" · {Key}: {Why}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
Main()
|