chronaxis.results.artifact_exports
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""拥有本次新增导出目录及有界索引，发布时只复制并核对实际文件。23每个导出至多16个普通文件；片段数量增长只增加磁盘索引，不积累路径列表。4仅请求完整性诊断时保存摘要；复制沿必要读取计算，普通运行核文件身份与大小。5"""67from pathlib import Path8from ..core.validation import require_text9import os10import stat11from hashlib import sha25612import json13from .frame_sort import FrameSort14from .budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner15from .errors import ResultClosedError,ResultFormatError16from .raw_exports import _write_all171819class ArtifactExportStore:20 """保存一次会话的信号/报文导出及独立清单，不借用原数据reader。"""2122 def __init__(self,budget,path,session_id,device_label,*,integrity=True):23 """创建独占目录及索引。参数: 预算/path/会话/设备。返回: owner。异常: 配置/IO/预算错。24 """25 if type(integrity) is not bool: raise TypeError('integrity必须为bool')26 self._integrity=integrity; self.index_info=None27 self._budget=budget; self.path=Path(path); self.session_id=session_id; self.device_label=device_label28 self._held_cost=self._fixed_cost=0; self._sort=None; self._input=self._output=self._index=self._cursor=None29 self.count=0; self.state='building'; self._closed=False30 try:31 require_text(device_label, 'device_label')32 _reserve_owned_ram(self,524288,'_fixed_cost')33 self.path.mkdir(parents=True,exist_ok=False)34 try:35 self._sort=FrameSort(self.path/'inventory',budget,integrity=integrity)36 except BaseException as error:37 self._sort=getattr(error,'frame_sort_cleanup_owner',None); raise38 except BaseException as error:39 _cleanup_owner(self,error,'artifact_export_store_cleanup_owner','导出索引构造'); raise4041 @property42 def closed(self):43 """读取真实关闭状态。参数: 无。返回: bool。异常: 无。44 """45 return self._closed4647 @property48 def owned_ram(self):49 """汇总本仓及索引实际费用供父层核账。参数: 无。返回: bytes。异常: 无。50 """51 return self._held_cost+(0 if self._sort is None else self._sort._held_cost)5253 def next_path(self):54 """提供当前导出独占子目录。参数: 无。返回: Path。异常: 非building。55 """56 if self._closed or self.state!='building':57 raise ResultClosedError('导出索引不能新增')58 return self.path/f'export-{self.count:08d}'5960 def _close_slot(self,slot):61 """真实close成功后撤槽。参数: slot责任名。返回: None。异常: 原close错，保槽重试。62 """63 owner=getattr(self,slot)64 if owner is not None:65 owner.close(); setattr(self,slot,None)6667 @staticmethod68 def _identity(path):69 """取得普通文件的替换/修改标识。参数: Path。返回: 四整数tuple。异常: 链接/非普通文件/IO。"""70 info=path.lstat()71 if path.is_symlink() or not stat.S_ISREG(info.st_mode):72 raise ResultFormatError('导出内容必须为普通文件')73 return (info.st_dev,info.st_ino,info.st_size,info.st_mtime_ns)7475 def _file(self,path,destination=None,expected=None,identity=None):76 """必要复制时才读内容；请求摘要则在同次读取计算。参数: 路径/目标/大小摘要/身份。返回: 大小和可选摘要。"""77 block = None78 try:79 if self._input is not None or self._output is not None:80 raise ResultClosedError('导出文件还有未清理句柄')81 before=self._identity(path)82 # 摘要模式会核实际内容，允许修复为原字节后重试；普通模式没有内容依据，仍拒身份变化。83 if identity is not None and tuple(identity)!=before and not self._integrity:84 raise ResultFormatError('导出文件已被替换或修改')85 if expected is not None and before[2]!=expected[0]:86 raise ResultFormatError('导出文件大小与登记不同')87 if destination is None and not self._integrity:88 return before[2],None89 self._input=path.open('rb',buffering=0)90 opened=os.fstat(self._input.fileno())91 if (opened.st_dev,opened.st_ino,opened.st_size,opened.st_mtime_ns)!=before:92 raise ResultFormatError('导出文件在打开时变化')93 if destination is not None:94 self._output=destination.open('xb',buffering=0)95 digest=sha256() if self._integrity else None; count=096 while True:97 block=self._input.read(65536)98 if not block: break99 count+=len(block)100 if count>before[2]: raise ResultFormatError('导出文件超过登记大小')101 if digest is not None: digest.update(block)102 if self._output is not None:103 self._budget.reserve_spool(len(block)); _write_all(self._output,block)104 value=None if digest is None else digest.hexdigest()105 if count!=before[2] or self._identity(path)!=before:106 raise ResultFormatError('导出文件在读取时变化')107 if expected is not None and expected[1] is not None and value!=expected[1]:108 raise ResultFormatError('导出文件内容与登记摘要不同')109 self._close_slot('_output'); self._close_slot('_input')110 # 写成功和真实close已确认；不再把新目标整份回读一次。111 return count,value112 finally:113 # 文件槽仍由本对象的可重试关闭负责；这里只清异常帧内的当前复制块。114 block = None115116 def append(self,export_type,scope,interval,*,segment_id=None,completed_files=None):117 """登记刚生成目录中有限文件的实际身份。参数: 类型/范围/区间/片段ID。返回: None。异常: 文件/索引错。118 """119 path=self.next_path(); files=[]120 # 内部writer交出的有限事实避免登记时再次扫描；旧调用未交事实则按原开关读取。121 supplied=None if completed_files is None else {item['name']:item for item in completed_files}122 if supplied is not None and len(supplied)>16: raise ValueError('导出文件事实超过上限')123 try:124 if export_type not in ('signal_csv','can_frames') or scope not in ('session','segments'):125 raise ValueError('导出种类或范围错误')126 for entry in path.iterdir():127 if len(files)>=16 or not entry.is_file() or entry.is_symlink() or not entry.name.isascii():128 raise ValueError('导出目录只允许至多16个普通文件')129 identity=self._identity(entry)130 if supplied is None:131 size,digest=self._file(entry,identity=identity)132 else:133 fact=supplied.pop(entry.name,None)134 if fact is None or fact['bytes']!=identity[2]:135 raise ResultFormatError('导出文件与writer事实不符')136 size,digest=fact['bytes'],fact.get('sha256') if self._integrity else None137 if self._integrity and (type(digest) is not str or len(digest)!=64):138 raise ResultFormatError('请求摘要但writer没有提供')139 files.append(dict(name=entry.name,bytes=size,_identity=identity,140 **({} if digest is None else {'sha256':digest})))141 if supplied: raise ResultFormatError('writer登记的文件缺失')142 if not any(f['name']=='manifest.json' for f in files):143 raise ValueError('导出缺完成清单')144 self._sort.append((self.count,),dict(index=self.count,export_type=export_type,scope=scope,segment_id=segment_id,145 interval=None if interval is None else [interval.start_ns,interval.end_ns],146 directory=path.name,files=sorted(files,key=lambda f:f['name']),device_label=self.device_label))147 self.count+=1148 except BaseException as error:149 self.state='failed'; error.artifact_export_store_cleanup_owner=self; raise150151 def seal(self):152 """封口目录索引后才允许移交。参数: 无。返回: None。异常: 状态/排序IO错。153 """154 self.next_path(); self._sort.seal(); self.state='sealed'155156 def copy_to(self,destination):157 """逐目录复制到新发布位置，按请求在同次读取核摘要。参数: 新结果根。返回: exports.jsonl路径。异常: IO/完整性错。158 """159 if self._closed or self.state!='sealed' or self._cursor is not None or self._index is not None:160 raise ResultClosedError('导出索引未封口或有发布资源待恢复')161 root=Path(destination)/'exports'; root.mkdir(exist_ok=False)162 index=Path(destination)/'exports.jsonl'163 index_digest=sha256() if self._integrity else None; index_bytes=0164 self.index_info=None165 try:166 self._index=index.open('xb',buffering=0)167 self._cursor=self._sort._read(self._sort.path/f'{self._sort.level}-0.jsonl')168 count=0169 for key,sequence,record in self._cursor:170 if key!=[count] and key!=(count,):171 raise ResultFormatError('导出索引序号不连续')172 if record['directory']!=f'export-{count:08d}':173 raise ResultFormatError('导出索引目录错误')174 target=root/record['directory']; target.mkdir(exist_ok=False)175 for entry in record['files']:176 name=entry['name']177 if Path(name).name!=name or '/' in name or '\\' in name or ':' in name:178 raise ResultFormatError('导出索引包含不安全文件名')179 self._file(self.path/record['directory']/name,target/name,180 (entry['bytes'],entry.get('sha256')),entry['_identity'])181 # 修改标识仅用于当前文件生命周期，不逐文件发布另一套证明。182 del entry['_identity']183 payload=(json.dumps(record,ensure_ascii=False,allow_nan=False,sort_keys=True)+'\n').encode('utf-8')184 self._budget.reserve_spool(len(payload)); _write_all(self._index,payload); count+=1185 if index_digest is not None: index_digest.update(payload)186 index_bytes+=len(payload)187 if count!=self.count:188 raise ResultFormatError('导出索引数量不符')189 self._close_slot('_cursor'); self._close_slot('_index')190 self.index_info=dict(name=index.name,bytes=index_bytes,191 **({} if index_digest is None else {'sha256':index_digest.hexdigest()}))192 return index193 except BaseException as error:194 error.artifact_export_store_cleanup_owner=self195 try:196 self.recover_copy()197 except BaseException as cleanup:198 error.add_note('导出复制清理失败: '+type(cleanup).__name__)199 raise200201 def recover_copy(self):202 """关闭失败发布句柄，可重新发布到另一新目录。参数: 无。返回: None。异常: 实际close错可重试。203 """204 for slot in ('_output','_input','_cursor','_index'):205 self._close_slot(slot)206 # 生成器已退出而真实close持续失败时，句柄仍登记在索引仓，重试必须处理它们。207 if self._sort is not None:208 while self._sort.handles:209 self._sort._close_file(self._sort.handles[-1])210211 def close(self):212 """真实关闭索引及IO后释放自己的费用，保留磁盘。参数: 无。返回: None。异常: 原close错。213 """214 if self._closed:215 return216 self.recover_copy(); self._close_slot('_sort')217 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self._closed=True218