1
{1001,"示例角色",7,10225,600000,[8]="1234567890123456789"}

1
原文件字节 + 指定编码
2
  → 按 LF/CRLF 取物理行，记录行号
3
  → 前五列整数 + 第六列完整 Lua 数据文本
4
  → 递归数据解析：显式键/顺序项/嵌套表/转义
5
  → 保留完整 payload；按事件编号投影 fields
6
  → 玩家/NPC/物件分域建立资料索引，保留每条历史记录
7
  → 按分析目标选时间区间和统计口径，再计算指标

1
总伤害（包含反弹）   = Σ事件 Σk=0..5 R.get(k, 0)
2
总伤害（不含反弹）   = Σ事件 Σk=0..4 R.get(k, 0)
3
有效伤害             = Σ事件 R.get(13, 0)
4
总治疗               = Σ事件 R.get(6, 0)
5
有效治疗             = Σ事件 R.get(14, 0)
6
DPS / HPS            = 已明确的伤害量 / 治疗量 ÷ 明确的统计秒数

1
#!/usr/bin/env python3
2
"""JCL 数据解析示例：Python 3.10+，仅标准库，不执行 Lua。"""
3
import argparse
4
import codecs
5
import hashlib
6
import json
7
import math
8
import re
9
import sys
10
from collections import Counter
11
from pathlib import Path
12

13

14
class LuaTable:
15
    def __init__(self, entries):
16
        self.entries = entries
17

18
    def get(self, key, default=None):
19
        # 数字键 1 与字符串键 "1" 不合并；原 entries 也保留重复键。
20
        for k, value in reversed(self.entries):
21
            if type(k) is type(key) and k == key:
22
                return value
23
        return default
24

25

26
class LuaDataParser:
27
    number = re.compile(r"[+-]?(?:0[xX][0-9a-fA-F]+|(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)")
28
    name = re.compile(r"[A-Za-z_][A-Za-z_0-9]*")
29

30
    def __init__(self, text, encoding="utf-8"):
31
        self.text = text
32
        self.pos = 0
33
        self.encoding = codecs.lookup(encoding).name.replace("utf-8-sig", "utf-8")
34

35
    def fail(self, message):
36
        raise ValueError(f"Lua 列 {self.pos + 1}: {message}")
37

38
    def space(self):
39
        while self.pos < len(self.text) and self.text[self.pos].isspace():
40
            self.pos += 1
41

42
    def take(self, token):
43
        self.space()
44
        if self.text.startswith(token, self.pos):
45
            self.pos += len(token)
46
            return True
47
        return False
48

49
    def need(self, token):
50
        if not self.take(token):
51
            self.fail(f"应为 {token!r}")
52

53
    def string(self):
54
        quote = self.text[self.pos]
55
        self.pos += 1
56
        result = bytearray()
57
        escapes = {"a": "\a", "b": "\b", "f": "\f", "n": "\n", "r": "\r",
58
                   "t": "\t", "v": "\v", "\\": "\\", '"': '"', "'": "'"}
59
        while self.pos < len(self.text):
60
            char = self.text[self.pos]
61
            self.pos += 1
62
            if char == quote:
63
                return result.decode(self.encoding)
64
            if char != "\\":
65
                result.extend(char.encode(self.encoding))
66
                continue
67
            if self.pos == len(self.text):
68
                self.fail("字符串转义被截断")
69
            char = self.text[self.pos]
70
            self.pos += 1
71
            if char in escapes:
72
                result.extend(escapes[char].encode(self.encoding))
73
            elif char.isascii() and char.isdigit():
74
                digits = char
75
                while len(digits) < 3 and self.pos < len(self.text) and self.text[self.pos] in "0123456789":
76
                    digits += self.text[self.pos]
77
                    self.pos += 1
78
                if int(digits) > 255:
79
                    self.fail("十进制转义超出一个字节")
80
                result.append(int(digits))
81
            else:
82
                self.fail(f"不支持的字符串转义 \\{char}")
83
        self.fail("字符串未闭合")
84

85
    def value(self):
86
        self.space()
87
        if self.pos == len(self.text):
88
            self.fail("缺少值")
89
        if self.text[self.pos] in "\"'":
90
            return self.string()
91
        if self.take("{"):
92
            entries, implicit = [], 1
93
            while not self.take("}"):
94
                if self.take("["):
95
                    key = self.value()
96
                    if type(key) not in (int, float, str, bool):
97
                        self.fail("表键必须是标量且不能为 nil")
98
                    if type(key) is float and key.is_integer():
99
                        key = int(key)
100
                    self.need("]")
101
                    self.need("=")
102
                else:
103
                    self.space()
104
                    start = self.pos
105
                    match = self.name.match(self.text, self.pos)
106
                    if match:
107
                        self.pos = match.end()
108
                    if match and self.take("="):
109
                        key = match.group()
110
                    else:
111
                        self.pos = start
112
                        key = implicit
113
                        implicit += 1
114
                entries.append((key, self.value()))
115
                if self.take("}"):
116
                    break
117
                if not (self.take(",") or self.take(";")):
118
                    self.fail("缺少表项分隔符")
119
            return LuaTable(entries)
120
        match = self.number.match(self.text, self.pos)
121
        if match:
122
            self.pos = match.end()
123
            token = match.group()
124
            if "x" in token.lower():
125
                return int(token, 16)
126
            if not any(c in token.lower() for c in ".e"):
127
                return int(token)
128
            value = float(token)
129
            if not math.isfinite(value):
130
                self.fail("非有限数字")
131
            return value
132
        match = self.name.match(self.text, self.pos)
133
        if match and match.group() in ("true", "false", "nil"):
134
            self.pos = match.end()
135
            return {"true": True, "false": False, "nil": None}[match.group()]
136
        self.fail("仅支持 Lua 数据字面量")
137

138
    def parse(self):
139
        value = self.value()
140
        self.space()
141
        if self.pos != len(self.text):
142
            self.fail("表后存在额外内容")
143
        if not isinstance(value, LuaTable):
144
            self.fail("JCL 载荷必须为表")
145
        return value
146

147

148
# 位置均为 Lua 一基索引；名字属于本文输出接口，不是文件中的键名。
149
FIELDS = {
150
    1: ("fight_time", "fighting uuid duration_ms map_id"),
151
    2: ("player_enter", "id"), 3: ("player_leave", "id"),
152
    4: ("player_info", "id name force_id kungfu_id equip_score equipment talents guid zhenpai"),
153
    5: ("player_fight", "id fighting life max_life mana max_mana"),
154
    6: ("npc_enter", "id"), 7: ("npc_leave", "id"),
155
    8: ("npc_info", "id name template_id employer_id x y z face"),
156
    9: ("npc_fight", "id fighting life max_life mana max_mana"),
157
    10: ("doodad_enter", "id"), 11: ("doodad_leave", "id"),
158
    12: ("doodad_info", "id template_id x y z face"),
159
    13: ("buff_update", "target_id delete index can_cancel buff_id stack end_frame init_raw level source_id valid_raw left_frame"),
160
    14: ("npc_say", "text speaker_id channel name"),
161
    15: ("warning", "warning_type text"),
162
    16: ("party_add", "team_id member_id group_index"),
163
    17: ("party_online", "team_id member_id online_flag"),
164
    18: ("system_message", "text channel"),
165
    19: ("cast", "caster_id skill_id level"),
166
    20: ("cast_respond", "caster_id skill_id level respond"),
167
    21: ("effect", "caster_id target_id react_raw scheme effect_id level critical result_count results"),
168
    22: ("block", "caster_id target_id scheme effect_id level damage_type"),
169
    23: ("shield", "caster_id target_id scheme effect_id level"),
170
    24: ("miss", "caster_id target_id scheme effect_id level"),
171
    25: ("hit", "caster_id target_id scheme effect_id level"),
172
    26: ("dodge", "caster_id target_id scheme effect_id level"),
173
    27: ("common_health", "target_id delta_life"),
174
    28: ("death", "target_id killer_id"),
175
    29: ("target_information", "target_type id x y z face life max_life mana max_mana absorb_value"),
176
}
177

178

179
def named(table, names):
180
    if not isinstance(table, LuaTable):
181
        return {}
182
    names = names.split()
183
    return {names[key - 1]: value for key, value in table.entries
184
            if type(key) is int and 1 <= key <= len(names)}
185

186

187
def json_default(value):
188
    if isinstance(value, LuaTable):
189
        return {"$table": value.entries}
190
    raise TypeError(type(value).__name__)
191

192

193
def parse_file(path, encoding):
194
    if codecs.lookup(encoding).name not in ("utf-8", "utf-8-sig", "gbk", "gb18030"):
195
        raise ValueError("示例只支持 utf-8、utf-8-sig、gbk、gb18030")
196
    raw = path.read_bytes()
197
    events, counts, entities = [], Counter(), {}
198
    # 逐物理行严格解码，错误定位到文件行；不猜编码、不忽略坏字节。
199
    lines = raw.split(b"\n")
200
    if lines and lines[-1] == b"":
201
        lines.pop()
202
    for lineno, raw_line in enumerate(lines, 1):
203
        try:
204
            line = raw_line.removesuffix(b"\r").decode(encoding)
205
            if lineno == 1:
206
                line = line.removeprefix("\ufeff")
207
            if not line:
208
                raise ValueError("空记录")
209
            columns = line.split("\t", 5)
210
            if len(columns) != 6:
211
                raise ValueError(f"需要六列，实际 {len(columns)} 列")
212
            if any(re.fullmatch(r"-?\d+", col) is None for col in columns[:5]):
213
                raise ValueError("前五列必须为整数")
214
            crc, frame, unix_s, client_ms, event_type = map(int, columns[:5])
215
            payload = LuaDataParser(columns[5], encoding).parse()
216
        except (ValueError, RecursionError) as exc:
217
            raise ValueError(f"{path.name}: 第 {lineno} 行: {exc}") from exc
218
        kind, names = FIELDS.get(event_type, ("unknown", ""))
219
        fields = named(payload, names)
220
        event = {"line": lineno, "crc": crc, "frame": frame, "unix_s": unix_s,
221
                 "client_ms": client_ms, "event_type": event_type, "kind": kind,
222
                 "fields": fields, "payload": payload, "raw_line": line}
223
        if event_type == 4:
224
            equipment = fields.get("equipment")
225
            if isinstance(equipment, LuaTable):
226
                # 只解释跨来源一致的前五项；附魔尾部存在版本/文档冲突，留在原表。
227
                event["equipment_common"] = [
228
                    {"entry_key": k, "fields": named(v, "slot table_type item_id strength sockets")}
229
                    for k, v in equipment.entries]
230
            talents = fields.get("talents")
231
            if isinstance(talents, LuaTable):
232
                event["talents_named"] = [
233
                    {"entry_key": k, "fields": named(v, "index skill_id level")}
234
                    for k, v in talents.entries]
235
        if event_type == 21 and isinstance(fields.get("results"), LuaTable):
236
            results = fields["results"]
237
            event["result_fields"] = {}
238
            result_names = {0: "physical", 1: "solar", 2: "neutral", 3: "lunar", 4: "poison",
239
                            5: "reflected", 6: "heal", 7: "steal_life", 8: "absorb_heal",
240
                            9: "absorb_damage", 10: "shield_damage", 11: "parry", 12: "insight",
241
                            13: "effective_damage", 14: "effective_heal",
242
                            15: "transfer_life", 16: "transfer_mana"}
243
            for key, value in results.entries:
244
                if type(key) is int and key in result_names:
245
                    event["result_fields"][result_names[key]] = value
246
        if event_type in (4, 8, 12) and type(fields.get("id")) is int:
247
            entity_kind = {4: "player", 8: "npc", 12: "doodad"}[event_type]
248
            entity_key = f"{entity_kind}:{fields['id']}"
249
            entity = entities.setdefault(entity_key, {"record_lines": [], "last_present_fields": {}})
250
            entity["record_lines"].append(lineno)
251
            # 缺项不覆盖旧值；显式 nil 仍覆盖。历史事件是时间点真值，末次索引仅便于查找。
252
            entity["last_present_fields"].update(fields)
253
        counts[event_type] += 1
254
        events.append(event)
255
    return {"source": {"file": path.name, "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest(),
256
                       "encoding": encoding, "crc_verified": False},
257
            "event_counts": dict(sorted(counts.items())), "entities": entities, "events": events}
258

259

260
def main():
261
    parser = argparse.ArgumentParser(description=__doc__)
262
    parser.add_argument("input", type=Path)
263
    parser.add_argument("--encoding", default="utf-8-sig", help="默认 utf-8-sig；简体客户端样本可指定 gbk")
264
    args = parser.parse_args()
265
    try:
266
        output = parse_file(args.input, args.encoding)
267
    except (OSError, ValueError, LookupError) as exc:
268
        parser.exit(2, f"解析失败: {exc}\n")
269
    # 输出固定 UTF-8 JSON；先完整解析再写出，错误时不会产生半份 JSON。
270
    encoded = json.dumps(output, ensure_ascii=False, allow_nan=False, default=json_default)
271
    sys.stdout.buffer.write(encoded.encode("utf-8") + b"\n")
272

273

274
if __name__ == "__main__":
275
    main()

1
python3 '/实际路径/jcl_parse.py' '/实际路径/战斗.jcl' --encoding gbk > '/实际路径/战斗.json'

1
python3 - <<'PY'
2
from pathlib import Path
3
rows = [
4
    '0\t100\t1700000000\t1000\t21\t{1001,2001,0,1,123,1,false,3,{[0]=100,[13]=80,[14]=0}}',
5
    '0\t101\t1700000000\t1060\t28\t{2001}',
6
]
7
Path('example.jcl').write_text('\n'.join(rows) + '\n', encoding='utf-8')
8
PY
9
python3 jcl_parse.py example.jcl > example.json

1
{
2
  "line": 1,
3
  "client_ms": 1000,
4
  "event_type": 21,
5
  "kind": "effect",
6
  "fields": {
7
    "caster_id": 1001,
8
    "target_id": 2001,
9
    "react_raw": 0,
10
    "scheme": 1,
11
    "effect_id": 123,
12
    "level": 1,
13
    "critical": false,
14
    "result_count": 3,
15
    "results": {"$table": [[0, 100], [13, 80], [14, 0]]}
16
  },
17
  "result_fields": {"physical": 100, "effective_damage": 80, "effective_heal": 0}
18
}

1
P1 16f0637b7005f96d03bf01d68a76ce2bdec62158c352de14ea13cd1c55efa727
2
P2 99634624801396d319981684e6b7a6f513c6fc6caaa0f0bf39638a282f7e81cf
3
P3 787e52b92ba87682e1cb21251f6deb3df105646c253e3f9dc8a7ed69f7ad5d15
4
P4 6ec97a9d3a384c6ac23d7bf28ef650c7e321bff3f9ae694bc3e5d42221f5ace5

1
python3 - <<'PY'
2
import importlib.util
3
from pathlib import Path
4
script = Path('/实际路径/jcl_parse.py')
5
sample_dir = Path('/实际路径/samples')
6
spec = importlib.util.spec_from_file_location('jcl_example', script)
7
module = importlib.util.module_from_spec(spec)
8
spec.loader.exec_module(module)
9
for path in sorted(sample_dir.glob('*.jcl')):
10
    data = module.parse_file(path, 'gbk')
11
    print(path.name, data['source']['sha256'], len(data['events']), data['event_counts'])
12
PY