chronaxis.results.frame_sort
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""为离线报文导出进行有界磁盘排序，不把整段报文留在内存。23输入为有限标量排序键和新导出自己的JSON记录，输出按键稳定排列的记录流。4每批最多64行，归并只开两路输入；旧run留作诊断且累计计入spool。旧调用默认用摘要绑定每行与结束标记；新版仅显式完整性诊断时启用。5不启用摘要时仍校验行结构、行长和结束计数，拒绝截断，不提供防篡改证明。6所有句柄在owner上登记，持续关闭失败仍可重试；本模块不读原日志或加载DBC。7"""89from hashlib import blake2b10from hmac import compare_digest11import json12from pathlib import Path13from secrets import token_bytes1415from .budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner16from .errors import ResultClosedError, ResultFormatError171819class FrameSort:20 """拥有一个新目录、最多64行缓冲和三个实际文件，封口后只允许顺序回读。"""2122 def __init__(self,path,budget,*,check_cancel=None,buffer_rows=64,integrity=True):23 """预约并创建暂存。参数: path新目录、budget、取消回调及批行数。返回: owner。异常: 配置/预算/I/O错。24 """25 if type(buffer_rows) is not int or not 1<=buffer_rows<=64:26 raise ValueError('报文排序缓冲需要1至64行')27 if type(integrity) is not bool: raise TypeError("integrity需要bool")28 self.integrity=integrity29 self.path=Path(path); self._budget=budget; self.check_cancel=check_cancel30 self._held_cost=self._fixed_cost=self._buffer_cost=031 self.buffer_rows=buffer_rows; self.buffer=[]; self.handles=[]32 self._closed=False; self.state='building'; self.count=self.runs=033 self.level=0; self.secret=None; self.cursor=None; self.pending_file=None34 try:35 # 两路解码树/编码/缓冲及句柄均有固定工作空间，实际行缓冲另计。36 _reserve_owned_ram(self,6*1048576,'_fixed_cost')37 self.path.mkdir(parents=True,exist_ok=False); self.secret=token_bytes(32) if integrity else None38 except BaseException as error:39 _cleanup_owner(self,error,'frame_sort_cleanup_owner','报文排序构造')40 raise4142 def _check(self):43 """在有界I/O间检查取消。参数: 无。返回: None。异常: 关闭或取消。44 """45 if self._closed:46 raise ResultClosedError('报文排序已关闭')47 if self.check_cancel is not None:48 self.check_cancel()4950 def _open(self,path,mode):51 """登记真实句柄后交付。参数: path、模式。返回: 文件。异常: 打开或登记失败。52 """53 self.pending_file=path.open(mode,buffering=65536)54 stream=self.pending_file55 self.handles.append(stream); self.pending_file=None56 return stream5758 def _close_file(self,stream):59 """关闭成功才撤销登记。参数: stream实际流。返回: None。异常: 关闭错保责任。60 """61 stream.close(); self.handles.remove(stream)6263 def _mac(self,path,index,body):64 """把完整行绑定当前run及顺序。参数: path/index/body。返回: 64字节摘要。异常: 原类型错。65 """66 digest=blake2b(key=self.secret,digest_size=32)67 digest.update(path.name.encode('ascii')); digest.update(str(index).encode('ascii'))68 digest.update(b'\0'); digest.update(body)69 return digest.hexdigest().encode('ascii')7071 def _write(self,stream,index,packet):72 """有界编码并实写一行。参数: stream/index/packet。返回: None。异常: 配额/超行/写入错。73 """74 self._check()75 body=json.dumps(packet,ensure_ascii=False,allow_nan=False,separators=(',',':')).encode('utf-8')76 if len(body)>65536:77 raise ResultFormatError('报文排序单行超过64KiB')78 payload=(self._mac(Path(stream.name),index,body)+b' ' if self.integrity else b'')+body+b'\n'79 self._budget.reserve_spool(len(payload))80 position=081 while position<len(payload):82 written=stream.write(payload[position:])83 if type(written) is not int or not 0<written<=len(payload)-position:84 raise OSError('报文排序写入无进展')85 position+=written8687 def _read(self,path):88 """逐行核完整性，必须读到合法结束记录。参数: path本次run。返回: 记录生成器。异常: 改写/截断/关闭错。89 """90 stream=self._open(path,'rb'); index=0; primary=None91 try:92 while True:93 self._check(); raw=stream.readline(65604)94 if not raw.endswith(b'\n') or len(raw)>(65602 if self.integrity else 65537) or self.integrity and raw[64:65]!=b' ':95 raise ResultFormatError('报文排序行截断或超限')96 body=raw[65:-1] if self.integrity else raw[:-1]97 if self.integrity and not compare_digest(raw[:64],self._mac(path,index,body)):98 raise ResultFormatError('报文排序行内容或顺序被修改')99 packet=json.loads(body)100 if type(packet) is dict:101 if packet!={'end':index} or stream.read(1):102 raise ResultFormatError('报文排序结束记录不一致')103 return104 if type(packet) is not list or len(packet)!=3 or type(packet[0]) is not list:105 raise ResultFormatError('报文排序行结构非法')106 index+=1107 yield (tuple(packet[0]),packet[1],packet[2])108 except BaseException as error:109 primary=error110 raise111 finally:112 # 校验/取消是主错误；关闭错误只能附注，真实句柄保owner供再次close。113 try:114 self._close_file(stream)115 except BaseException as cleanup:116 if primary is None:117 raise118 primary.add_note('报文排序回读关闭待恢复: '+repr(cleanup))119120 def append(self,key,packet):121 """预约后缓冲一条。参数: key有限同类型标量tuple、packet有限JSON对象。返回: None。异常: 结构/预算/I/O错。122 """123 self._check()124 if self.state!='building' or type(key) is not tuple or len(key)>32 or any(type(v) not in (int,str) for v in key):125 raise ValueError('报文排序需要building及有限标量键')126 encoded=json.dumps(packet,ensure_ascii=False,allow_nan=False,separators=(',',':')).encode('utf-8')127 if len(encoded)>65000:128 raise ResultFormatError('报文记录超过排序行能力')129 # JSON往返冻结调用者数据，避免追加后修改改变尚未写盘的结果。130 _reserve_owned_ram(self,4096+len(encoded)*32,'_buffer_cost')131 self.buffer.append((key,self.count,json.loads(encoded))); self.count+=1132 if len(self.buffer)>=self.buffer_rows:133 self._flush()134135 def _flush(self):136 """稳定排序有限缓冲并保存一个run。参数: 无。返回: None。异常: 排序/I/O/配额错。137 """138 if not self.buffer:139 return140 self.buffer.sort(key=lambda item:(item[0],item[1]))141 stream=self._open(self.path/f'0-{self.runs}.jsonl','xb')142 for index,packet in enumerate(self.buffer):143 self._write(stream,index,packet)144 self._write(stream,len(self.buffer),{'end':len(self.buffer)})145 self._close_file(stream); self.runs+=1; self.buffer.clear()146 _release_owned_ram(self,self._buffer_cost,'_buffer_cost')147148 def _merge(self,left,right,target):149 """只保两路当前行进行稳定归并。参数: 两个run及新目标。返回: None。异常: 输入校验/写入/关闭错。150 """151 first=self._read(left); second=self._read(right) if right is not None else iter(())152 stream=self._open(target,'xb'); count=0; primary=None153 try:154 a=next(first,None); b=next(second,None)155 while a is not None or b is not None:156 choose_first=b is None or a is not None and (a[0],a[1])<=(b[0],b[1])157 row=a if choose_first else b158 self._write(stream,count,row); count+=1159 if choose_first:160 a=next(first,None)161 else:162 b=next(second,None)163 self._write(stream,count,{'end':count}); self._close_file(stream)164 except BaseException as error:165 primary=error166 raise167 finally:168 cleanup_error=None169 for reader in (first,second):170 close=getattr(reader,'close',None)171 if close is not None:172 try:173 close()174 except BaseException as cleanup:175 cleanup_error=cleanup_error or cleanup176 if cleanup_error is not None:177 if primary is None:178 raise cleanup_error179 primary.add_note('报文归并游标关闭待恢复: '+repr(cleanup_error))180181 def seal(self):182 """完成所有有界归并，空输入也有完整结束记录。参数: 无。返回: None。异常: 输入/预算/取消/I/O错。183 """184 self._check()185 if self.state!='building':186 raise ResultClosedError('报文排序不能重复封口')187 self._flush()188 if not self.runs:189 stream=self._open(self.path/'0-0.jsonl','xb')190 self._write(stream,0,{'end':0}); self._close_file(stream); self.runs=1191 while self.runs>1:192 for index in range(0,self.runs,2):193 self._merge(self.path/f'{self.level}-{index}.jsonl',194 self.path/f'{self.level}-{index+1}.jsonl' if index+1<self.runs else None,195 self.path/f'{self.level+1}-{index//2}.jsonl')196 self.level+=1; self.runs=(self.runs+1)//2197 self.state='sealed'198199 def rows(self):200 """交付单一可关闭回读游标。参数: 无。返回: (key,序号,记录)生成器。异常: 非sealed或已有游标。201 """202 self._check()203 if self.state!='sealed' or self.cursor is not None:204 raise ResultClosedError('报文排序未封口或已交付游标')205 self.cursor=self._read(self.path/f'{self.level}-0.jsonl')206 return self.cursor207208 def close(self):209 """关闭全部实际文件后释放自有预算；保暂存供诊断。参数: 无。返回: None。异常: 原关闭/释放错可重试。210 """211 if self._closed:212 return213 if self.cursor is not None:214 self.cursor.close(); self.cursor=None215 if self.pending_file is not None:216 self.pending_file.close(); self.pending_file=None217 while self.handles:218 self._close_file(self.handles[-1])219 self.buffer.clear(); self.secret=None220 _release_owned_ram(self,self._buffer_cost,'_buffer_cost')221 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self._closed=True222