105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
# analyze_opt_trace.py - 还原客户端每帧的 opt 位序列，按「模型片段」汇总
|
||
#
|
||
# 数据来源：镜像站 module_24635.js 的 Ht.t7h 打点，每条记录：
|
||
# rid = 位读取器 id（同一帧共享）、pos = 读位前绝对位索引、bit = 读到的位、
|
||
# slots/elemKeys = 内层读器（或列表元素）的槽位名（全局唯一，可反查模型）。
|
||
#
|
||
# 输出：把同一帧里「连续属于同一个模型」的读位合并成片段：
|
||
# 模型@起始位 × 位数 = 位串
|
||
# 这样可以直观看出某个模型到底消费了几位（例如 Item3DCC 是 4 位还是 5 位）。
|
||
#
|
||
# 用法: python tools/analyze_opt_trace.py [--only bs,Tu,tu] [--frames 5]
|
||
import json
|
||
import os
|
||
import sys
|
||
from collections import defaultdict
|
||
|
||
Root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
def Arg(Name, Def):
|
||
if '--' + Name in sys.argv:
|
||
I = sys.argv.index('--' + Name)
|
||
if I + 1 < len(sys.argv):
|
||
return sys.argv[I + 1]
|
||
return Def
|
||
|
||
|
||
NamesDir = os.path.abspath(Arg('names', os.path.join(Root, 'names')))
|
||
TracePath = os.path.join(NamesDir, 'opt_trace.json')
|
||
FieldsPath = os.path.join(NamesDir, 'codec_fields.json')
|
||
|
||
|
||
def LoadIndex():
|
||
"""槽位集合 → codec 键。用 Fields[].Slot（比 Slots 更全），再补 Slots。"""
|
||
with open(FieldsPath, 'r', encoding='utf-8') as F:
|
||
Codecs = json.load(F).get('Codecs') or {}
|
||
Index = {}
|
||
for Key, C in Codecs.items():
|
||
Slots = tuple(sorted({str(F2.get('Slot')) for F2 in (C.get('Fields') or []) if F2.get('Slot')}))
|
||
if Slots and Slots not in Index:
|
||
Index[Slots] = Key
|
||
for Key, C in Codecs.items():
|
||
Slots = tuple(sorted(S.get('Slot') for S in (C.get('Slots') or [])))
|
||
if Slots and Slots not in Index:
|
||
Index[Slots] = Key
|
||
return Index
|
||
|
||
|
||
def Main():
|
||
if not os.path.exists(TracePath):
|
||
print('缺少 ' + TracePath + '（先让镜像站进一局）')
|
||
return 1
|
||
with open(TracePath, 'r', encoding='utf-8') as F:
|
||
Data = json.load(F)
|
||
Rows = Data.get('trace') or []
|
||
print('opt 打点：%d 条（采集于 %s）' % (len(Rows), Data.get('at', '?')))
|
||
|
||
Index = LoadIndex()
|
||
Only = [S for S in (Arg('only', 'Tu,tu,bs') or '').split(',') if S]
|
||
MaxFrames = int(Arg('frames', '5') or 5)
|
||
|
||
Frames = defaultdict(list)
|
||
for R in Rows:
|
||
Frames[R.get('rid')].append(R)
|
||
|
||
def NameOf(R):
|
||
Slots = tuple(sorted(R.get('slots') or []))
|
||
Key = Index.get(Slots)
|
||
if Key:
|
||
return Key.split('::')[-1]
|
||
Elem = tuple(sorted(R.get('elemKeys') or []))
|
||
if Elem:
|
||
K2 = Index.get(Elem)
|
||
return 'list<' + (K2.split('::')[-1] if K2 else '?') + '>'
|
||
return '?' + ''.join(Slots)
|
||
|
||
def Segments(Seq):
|
||
Out = []
|
||
for R in Seq:
|
||
N = NameOf(R)
|
||
if Out and Out[-1][0] == N and Out[-1][2] == R.get('pos'):
|
||
Out[-1][1] += str(R.get('bit') and 1 or 0)
|
||
Out[-1][2] = R.get('pos', 0) + 1
|
||
else:
|
||
Out.append([N, str(R.get('bit') and 1 or 0), R.get('pos', 0) + 1, R.get('pos', 0)])
|
||
return Out
|
||
|
||
Shown = 0
|
||
for Rid in sorted(Frames, key=lambda X: (X is None, X)):
|
||
Seq = Frames[Rid]
|
||
Segs = Segments(Seq)
|
||
if Only and not any(S[0] in Only for S in Segs):
|
||
continue
|
||
Shown += 1
|
||
if Shown > MaxFrames:
|
||
continue
|
||
Line = ' '.join('%s@%d×%d=%s' % (S[0], S[3], len(S[1]), S[1]) for S in Segs[:40])
|
||
print(' rid=%s（%d 位）：%s' % (Rid, len(Seq), Line))
|
||
print(' 匹配帧数：%d' % Shown)
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(Main())
|