122 lines
6.3 KiB
Python
122 lines
6.3 KiB
Python
"""Inventory CW's local protobuf graph and dispatcher coverage (no Client calls).
|
|
|
|
This is a discovery artifact, not a gameplay or wire-compatibility test.
|
|
Command identity comes from the revision-tagged upstream table. The message
|
|
graph retains obfuscated names rather than assigning guessed meanings.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import pathlib
|
|
import re
|
|
|
|
|
|
def declarations(source):
|
|
result = {}
|
|
for match in re.finditer(r"^\s*(message|enum)\s+(\w+)\s*\{", source, re.M):
|
|
depth, end = 1, match.end()
|
|
while end < len(source) and depth:
|
|
depth += (source[end] == "{") - (source[end] == "}")
|
|
end += 1
|
|
if depth:
|
|
raise ValueError("Unterminated declaration: " + match[2])
|
|
result[match[2]] = (match[1], source[match.end():end - 1])
|
|
return result
|
|
|
|
|
|
def inventory(root):
|
|
schema = root / "protocol/proto/StarRail.proto"
|
|
upstream = (root / "protocol/proto/upstream_cmdids.go").read_text(encoding="utf-8")
|
|
revision = re.search(r'CommandSourceRevision = "([^"]+)"', upstream)[1]
|
|
commands = {name: int(number) for name, number in
|
|
re.findall(r'^\s*"(\w+)":\s*(\d+),', upstream, re.M)}
|
|
types = declarations(schema.read_text(encoding="utf-8"))
|
|
family = {name: number for name, number in commands.items()
|
|
if name.startswith("GridFight")}
|
|
# BuyGoodRsp is the explicitly named response of GridFightBuyGoodsCsReq
|
|
# in the current schema. Include the source declaration, without inventing
|
|
# a GridFightBuyGoodsScRsp alias or choosing a numeric response here.
|
|
seeds = set(family) | {"GridFightGameInfo", "BuyGoodRsp"}
|
|
reachable, queue = {}, sorted(seeds)
|
|
while queue:
|
|
name = queue.pop()
|
|
if name in reachable or name not in types:
|
|
continue
|
|
kind, body = types[name]
|
|
deps = sorted(set(re.findall(r"\b\w+\b", body)) & types.keys() - {name})
|
|
reachable[name] = {"kind": kind, "references": deps, "body": body.strip()}
|
|
queue.extend(deps)
|
|
handlers = {}
|
|
for path in sorted((root / "internal/gameserver").glob("*.go")):
|
|
text = path.read_text(encoding="utf-8")
|
|
for match in re.finditer(r'\b(cwHandleNamed|cwHandle|handleNamed|handle|rawNamed|dummy)\("([^"]+)"', text):
|
|
if match[2].startswith("GridFight"):
|
|
handlers.setdefault(match[2], []).append({
|
|
"file": path.relative_to(root).as_posix(), "kind": match[1],
|
|
"line": text.count("\n", 0, match.start()) + 1,
|
|
})
|
|
return {
|
|
"command_source_revision": revision,
|
|
"schema_sha256": hashlib.sha256(schema.read_bytes()).hexdigest(),
|
|
"limitations": [
|
|
"Local schema graph only; field semantics and current Client tags require separate evidence.",
|
|
"Handler registration does not establish correct behavior or completed gameplay.",
|
|
"Non-GridFight command names used by pending selections require separate mapping.",
|
|
],
|
|
"commands": dict(sorted(family.items())),
|
|
"registrations": handlers,
|
|
"types": dict(sorted(reachable.items())),
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=pathlib.Path, default=pathlib.Path(__file__).resolve().parents[1])
|
|
parser.add_argument("--output", type=pathlib.Path, required=True)
|
|
parser.add_argument("--reference-root", type=pathlib.Path,
|
|
help="Optional source checkout to compare; never executes its server")
|
|
parser.add_argument("--markdown", type=pathlib.Path,
|
|
help="Write request/pending-action coverage for implementation planning")
|
|
args = parser.parse_args()
|
|
result = inventory(args.root.resolve())
|
|
reference = {}
|
|
if args.reference_root:
|
|
for path in sorted(args.reference_root.rglob("HandlersGridFight*.cs")):
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
for name in re.findall(r"\[Opcode\(CmdIds\.(\w+CsReq)\)\]", text):
|
|
reference.setdefault(name, []).append(path.relative_to(args.reference_root).as_posix())
|
|
result["reference_source_handlers"] = reference
|
|
if args.markdown:
|
|
requests = sorted(name for name in result['commands'] if name.endswith('CsReq'))
|
|
lines = ['# CW dispatcher and source comparison', '',
|
|
'Generated by `tools/inventory_cw.py` from the current revision table and source registrations.',
|
|
'**Registration is not gameplay verification.** A registered pending dispatcher still needs each action implemented.',
|
|
'Reference source names are evidence for comparison only; they are not protocol aliases.', '',
|
|
'| Current request | Go registration | Reference same name |',
|
|
'| --- | --- | --- |']
|
|
for name in requests:
|
|
handlers = result['registrations'].get(name.removesuffix('CsReq'), [])
|
|
go = ', '.join(f"`{h['file']}:{h['line']}` ({h['kind']})" for h in handlers) or 'Missing'
|
|
lines.append(f"| `{name}` | {go} | {'Yes' if name in reference else 'Not matched'} |")
|
|
root = args.root.resolve()
|
|
dispatcher = (root / 'internal/gameserver/currency_wars_entry_choice.go').read_text(encoding='utf-8')
|
|
body = result['types']['GridFightHandleBonfireHealHpCsReq']['body']
|
|
arms = re.findall(r'^\s*\w+\s+(\w+)\s*=\s*\d+;', body, re.M)
|
|
lines += ['', '## Pending action arms', '',
|
|
'“Referenced” means the dispatcher inspects the field; it does not prove the entire action/effect works.', '',
|
|
'| Schema arm | Dispatcher |', '| --- | --- |']
|
|
for arm in arms:
|
|
if arm in ('queue_position',):
|
|
continue
|
|
lines.append(f"| `{arm}` | {'Referenced' if 'req.Get' + arm + '()' in dispatcher else 'Missing'} |")
|
|
args.markdown.parent.mkdir(parents=True, exist_ok=True)
|
|
args.markdown.write_text('\n'.join(lines) + '\n', encoding='utf-8')
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(f"CW: {len(result['commands'])} commands, {len(result['types'])} reachable declarations, "
|
|
f"{len(result['registrations'])} handler registration names")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|