chronaxis.results.export_files
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用有限缓冲写新导出目录，全部文件真正关闭后才独占发布。23供信号CSV及新报文导出共用：输入是已编码的有限bytes，输出为按模式返回的目录或实际路径清单。4临时文件保留供排查；失败不覆盖用户目录、不删除已有数据。累计spool先预约，5RAM按自有槽释放；关闭失败保真实句柄供重试。文件和目录均使用平台不覆盖移动；不支持的环境明确拒绝。此层不读取原始日志、不解码，也不计算插值。6"""78from hashlib import sha2569import json10from pathlib import Path11import platform12from uuid import uuid41314from .budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner15from .errors import ResultClosedError, ResultFormatError161718class ExportFiles:19 """拥有当前/挂起的有限输出文件、消费游标及可选处理owner，最多16文件。"""2021 def __init__(self, destination, budget, *, integrity=True, timestamped=False):22 """预检输出并创建独占临时目录。参数: destination输出根、budget共享账本，integrity控制摘要，timestamped为内部选择的公开时间命名模式。2324 返回: owner。异常: 路径/平台/预算/I/O错；清理失败附export_cleanup_owner。25 """26 if not isinstance(destination,(str,Path)) or not isinstance(budget,ResourceBudget):27 raise TypeError('导出需要目录路径和ResourceBudget')28 if type(integrity) is not bool:29 raise TypeError('integrity必须为bool')30 self._integrity=integrity31 self._timestamped = timestamped32 self._failed = False33 self._temporary_identity = None34 self.destination = Path(destination).absolute()35 # 只复用符合规则的结果根；已有文件仍通过独占命名保护。36 from .platform_files import require_publication_platform37 require_publication_platform()38 try:39 self.destination.lstat()40 except FileNotFoundError:41 pass42 else:43 if not timestamped or not self.destination.is_dir() or self.destination.is_symlink():44 raise FileExistsError(self.destination)45 self._budget = budget46 self._held_cost = self._fixed_cost = self._row_cost = 047 self._closed = False48 self.reader = self.worker = self.stream = self.verifier = self.temporary = None49 self._parked = {}50 self._name = self._hash = None51 self._bytes = 052 self.files = []53 try:54 _reserve_owned_ram(self,131072+len(str(self.destination).encode('utf-8'))*32,'_fixed_cost')55 self.destination.parent.mkdir(parents=True,exist_ok=True)56 self.temporary = self.destination.parent/('.'+self.destination.name+'.'+uuid4().hex+'.incomplete')57 self.temporary.mkdir(exist_ok=False)58 from .owned_directory import directory_identity59 self._temporary_identity = directory_identity(self.temporary)60 except BaseException as error:61 self._failed = True62 _cleanup_owner(self,error,'export_cleanup_owner','导出目录构造')63 raise6465 @property66 def closed(self):67 """返回真实关闭状态。参数: 无。返回: bool。异常: 无。68 """69 return self._closed7071 def open_file(self, name):72 """打开一个固定平面文件。参数: name安全文件名。返回: None。异常: 未关闭/重复/非法名/I/O错。73 """74 if self._closed or self.stream is not None:75 raise ResultClosedError('导出已关闭或前一文件未关闭')76 if (type(name) is not str or not name or len(name)>128 or Path(name).name!=name77 or any(c in name for c in '/\\:') or name in ('.','..') or len(self.files)+len(self._parked)>=16):78 raise ValueError('导出文件名或数量非法')79 if name in self._parked or any(item['name']==name for item in self.files):80 raise ValueError('导出文件重复')81 # 多张表共用同次输入时会同时存在缓冲；每个新文件单独预约，不各自借满总预算。82 _reserve_owned_ram(self,131072,'_fixed_cost')83 self.stream = (self.temporary/name).open('xb',buffering=65536)84 self._name,self._hash,self._bytes = name,sha256() if self._integrity else None,08586 def select_file(self, name):87 """切换有限输出表，不重新打开或覆盖。参数: name已开或新固定文件名。返回: None。异常: 打开/状态错。88 """89 if self._closed:90 raise ResultClosedError('导出已关闭')91 if self._name == name:92 return93 if self.stream is not None:94 self._parked[self._name] = (self.stream,self._hash,self._bytes)95 self.stream = self._name = self._hash = None96 if name in self._parked:97 self.stream,self._hash,self._bytes = self._parked.pop(name)98 self._name = name99 else:100 self.open_file(name)101102 def append(self, payload):103 """有限块实际写入并更新摘要。参数: payload最多64KiB bytes。返回: None。异常: 配额/写入/结构错。104 """105 if self._closed or self.stream is None:106 raise ResultClosedError('导出文件未打开')107 if type(payload) is not bytes or len(payload)>65536:108 raise ResultFormatError('导出块需要至多64KiB bytes')109 # 配额在任何写动作之前预约，短写继续；零进展不能伪装成功。110 self._budget.reserve_spool(len(payload))111 position = 0112 while position < len(payload):113 written = self.stream.write(payload[position:])114 if type(written) is not int or not 0 < written <= len(payload)-position:115 raise OSError('导出写入没有合法进展')116 position += written117 if self._hash is not None: self._hash.update(payload)118 self._bytes += len(payload)119120 def close_file(self):121 """真实关闭后登记摘要。参数: 无。返回: None。异常: 关闭失败保留句柄供重试。122 """123 if self.stream is None:124 return125 self.stream.close()126 self.stream = None127 self.files.append(dict(name=self._name,bytes=self._bytes,128 **({} if self._hash is None else {'sha256':self._hash.hexdigest()})))129 self._name = self._hash = None130131 def close_reader(self):132 """关闭已接管的输入游标。参数: 无。返回: None。异常: 原关闭错误，保留游标供重试。133 """134 if self.reader is not None:135 close = getattr(self.reader,'close',None)136 if close is not None:137 close()138 self.reader = None139140 def close_files(self):141 """关闭当前及挂起文件。参数: 无。返回: None。异常: 原关闭错误，未关的仍在owner中。142 """143 self.close_file()144 while self._parked:145 name = next(iter(self._parked))146 self.select_file(name)147 self.close_file()148149 def register_external(self,name):150 """登记同目录内已关闭的外部writer文件。参数: name固定文件名。返回: None。异常: 名称/读取/关闭错。151152 外部writer已付spool，本层仅有限回读计算摘要，不再次计算磁盘费用。153 """154 if (self._closed or self.verifier is not None or type(name) is not str or not name or len(name)>128 or Path(name).name!=name155 or any(c in name for c in '/\\:') or name in ('.','..')156 or len(self.files)+len(self._parked)+(self.stream is not None)>=16157 or name==self._name or name in self._parked or any(f['name']==name for f in self.files)):158 raise ValueError('外部导出文件名称或状态非法')159 # 未请求完整性诊断时，只登记外部writer已关闭文件的实际大小，不为摘要回读。160 if not self._integrity:161 self.files.append({'name':name,'bytes':(self.temporary/name).stat().st_size})162 return163 self.verifier=(self.temporary/name).open('rb')164 digest=sha256(); size=0165 while True:166 chunk=self.verifier.read(65536)167 if not chunk:168 break169 digest.update(chunk); size+=len(chunk)170 self.verifier.close()171 self.verifier=None172 self.files.append({'name':name,'sha256':digest.hexdigest(),'bytes':size})173174 def finish(self, metadata):175 """写完成清单并原子安装目录。参数: metadata有界JSON事实。返回: 内部目录Path或公开ExportManifest。176177 异常: 序列化/配额/关闭/竞态目标错误；失败保.incomplete目录，不冒报发布成功。178 """179 from ..execution.history import _metadata_bytes180 # 输入完整消费并真实关闭、数据文件关闭后才写完成清单。181 self.close_reader()182 if self.worker is not None:183 self.worker.close()184 self.worker = None185 if self.verifier is not None:186 self.verifier.close()187 self.verifier=None188 self.close_files()189 publication = None190 result = None191 try:192 # 内部导出仍使用新目录；公开调用在已有根下使用同次时间文件名。193 if self._timestamped:194 from .publication import PublicationNames195 publication = PublicationNames(self.destination, (*[f['name'] for f in self.files], 'manifest.json'))196 metadata = dict(metadata)197 if 'exact_source_index' in metadata:198 metadata['exact_source_index'] = publication.names[metadata['exact_source_index']]199 _reserve_owned_ram(self,65536+_metadata_bytes(metadata)*16,'_row_cost')200 try:201 files = tuple(dict(f, name=publication.names[f['name']]) if publication else dict(f) for f in self.files)202 document = dict(metadata,export_version=1,status='completed',files=files)203 payload = (json.dumps(document,ensure_ascii=False,allow_nan=False,sort_keys=True)+'\n').encode('utf-8')204 self.open_file('manifest.json')205 # 清单受声明规模和RAM预算约束，按有限块写出。206 for start in range(0,len(payload),65536):207 self.append(payload[start:start+65536])208 self.close_file()209 finally:210 _release_owned_ram(self,self._row_cost,'_row_cost')211 self.close()212 if publication is None:213 from .platform_files import move_no_replace214 move_no_replace(self.temporary,self.destination)215 return self.destination216 from .manifest import ExportManifest217 result = ExportManifest(self.destination, self.destination/publication.names['manifest.json'],218 tuple(self.destination/f['name'] for f in files))219 publication.publish(self.temporary, 'manifest.json')220 self.temporary.rmdir()221 return result222 except BaseException as error:223 if publication is not None and publication.completed:224 error.completed_manifest = result225 if publication is not None:226 try:227 publication.close()228 except BaseException as cleanup:229 error.add_note('导出命名锁释放失败：' + str(cleanup))230 raise231232 def close(self):233 """只关闭自有资源，保留诊断文件。参数: 无。返回: None。异常: 原关闭/释放错，允许重试。234 """235 if self._closed:236 try:237 self._cleanup_failed_output()238 except BaseException:239 self._closed = False240 raise241 return242 self.close_reader()243 if self.worker is not None:244 self.worker.close()245 self.worker = None246 if self.verifier is not None:247 self.verifier.close()248 self.verifier=None249 self.close_files()250 _release_owned_ram(self,self._row_cost,'_row_cost')251 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')252 self._cleanup_failed_output()253 self._closed = True254255 def _cleanup_failed_output(self):256 """公开导出失败时清理本次临时目录。参数: 无。返回: None。异常: 身份或IO错误，可重试close。"""257 if self._failed and self._timestamped and self.temporary is not None:258 from .owned_directory import remove_owned_directory259 remove_owned_directory(self.temporary, self._temporary_identity)260