chronaxis.results.result
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""从封口暂存提供流式结果，关闭文件后按时间命名发布，完成清单最后出现。23输入是已排序ResultSpool与运行元数据，输出是行迭代器和OutputManifest。4CSV与JSONL使用同一套平坦摘要字段，不保留旧row_json或结果包装副本。5成功生成表、质量记录及完成清单后才安装目录；已有目标拒绝，失败保留诊断临时目录。6close关闭所有活跃迭代器且幂等；默认清理自有暂存，保留原输入和已发布文件。无旧reader/decoder/writer调用。7单个文件或目录按平台不覆盖移动；Windows保留有限共享冲突重试。8元数据、读槽和发布workspace分别记录实际RAM费用，预算更新后取消仍可恢复清账。9读取器真实handle.close失败保留句柄并拒绝继续读；结果close失败可重试，不提前标closed。10上述重试保证只覆盖托管的结果资源，不覆盖任意局部文件句柄的持续关闭失败。11"""1213# 发布只新增本次文件；标准库逐行读写，不覆盖已有结果。14import csv15from contextlib import nullcontext16from collections.abc import Iterator17from datetime import datetime, timezone18import hashlib19import json20import os21from pathlib import Path22import platform23import tempfile24import time25from .codec import decode_row26from .errors import (AnalysisError, BudgetExceededError, ResultClosedError, ResultFormatError,27 ContributionsUnavailableError)28from .manifest import RunMetadata, OutputManifest29from .artifact_exports import ArtifactExportStore30from .rows import ResultRow, row_table31from .summary import (SummaryCalculationRow, SummaryEventRow, SummarySequenceRow, SummarySegmentRow,32 SummaryContributionRow, SummarySelectionRow, SummaryInputRow, SummaryAggregationRow,33 summary_fields, SUMMARY_MODELS, SUMMARY_COLUMNS)34from .owned_directory import directory_identity, remove_owned_directory35from .spool import ResultSpool, _reserve_owned_ram, _release_owned_ram, _cleanup_owner363738def _json(value) -> str:39 """生成标准紧凑JSON文本，拒绝非有限普通数字。4041 参数: value为受控JSON结构。返回: str。异常: ValueError为非法数值。42 """43 try:44 return json.dumps(value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(',', ':'))45 finally:46 # 原异常/TB和返回值保持；只断本层临时引用，不清调用方帧。47 value = None484950def _row_fields(row) -> dict:51 """提取当前平坦结果的CSV字段；数组单元格使用有限JSON文本。5253 参数: row为当前登记的结果行。返回: 独立字段字典。异常: 未支持模型或非有限值错误。54 """55 try:56 result = summary_fields(row)57 for name in ('parent_event_ids','group_keys','group_values'):58 if name in result:59 result[name] = _json(result[name])60 return result61 finally:62 row = result = None636465class _DigestOutput:66 """借用本次已打开的输出流，同一次写入计算请求的摘要，不拥有关闭责任。"""67 def __init__(self,stream):68 """建立固定状态。参数: 文本或二进制流。返回: owner。异常: 哈希构造错。"""69 self.stream=stream;self.digest=hashlib.sha256();self.size=07071 def write(self,payload):72 """完整写出有限payload后更新摘要，不关闭借用流。7374 参数: str/bytes。返回: 原长度。异常: 短写/IO；回溯不保留本层载荷引用。75 """76 encoded=None77 try:78 count=self.stream.write(payload)79 if count!=len(payload): raise OSError('结果写入不完整')80 encoded=payload.encode('utf-8') if isinstance(payload,str) else payload81 self.digest.update(encoded);self.size+=len(encoded)82 return count83 finally:84 # 只解除本函数的两份引用，不清调用者或写入器的异常帧。85 payload=encoded=None8687 def info(self,name):88 """返回已关闭文件的有限事实。参数: 文件名。返回: dict。异常: 无。"""89 return dict(name=name,bytes=self.size,sha256=self.digest.hexdigest())909192class _RowIterator:93 """独立文件位置的有限行迭代器，实际读槽与文件关闭完成后才closed。9495 _reader_cost为实际已付槽位，_held_cost仅含本迭代器费用。关闭中的迭代器仍在96 owner登记中，成功关handle后再抵销费用、解除登记；失败可由本对象或父close重试。97 handle及内部发布模式的_index_handle保留真实文件，所有实际FD关闭后才抵销读槽。98 """99100 def __init__(self, owner, table: str):101 """先预留解码槽位，再打开只读暂存并登记活跃资源。102103 参数: owner为AnalysisResult，table为固定表名；publication只供内部legacy JSONL发布。104 返回: 公共typed行迭代器，内部发布模式返回已核完整原行对应的有限质量字段。105 异常: 预算/IO/取消原样传播；持续清理失败附row_iterator_cleanup_owner，父登记仍可达。106 """107 owner._require_open()108 self.owner, self.table, self.closed = owner, table, False109 # 在首次预约前准备可清理字段；父元数据覆盖这个尚无读槽的小登记引用。110 self._budget, self._held_cost, self._reader_cost = owner.spool.budget, 0, 0111 self.handle, self._handle_closed, self._closing = None, True, False112 self._payload = None113 if table not in owner.spool.tables:114 if table not in SUMMARY_COLUMNS:115 raise ValueError('未注册结果表')116 # 未配置legacy是实际无文件的空游标；不登记FD、不申请读槽，但保close接口。117 self.closed = True118 return119 try:120 owner._iterators.add(self)121 # 所有表都按成功追加实长付款；原编码硬限和读前长度检查继续保留。122 self.max_row_bytes = owner.spool.row_limit(table)123 _reserve_owned_ram(self, owner.spool.reader_cost_for(table), '_reader_cost')124 self.handle = owner.spool.files[table].open('rb', buffering=8192)125 self._handle_closed = False126 except BaseException as exc:127 _cleanup_owner(self, exc, 'row_iterator_cleanup_owner', '结果读取器构造')128 raise129130 def __iter__(self):131 """保持当前读取位置，不重读输入。132133 参数: 无。返回: self。异常: 无。134 """135 return self136137 def __next__(self) -> ResultRow | dict:138 """读取一条有限行，EOF或异常关闭资源。139140 参数: 无。返回: 公共typed行或内部认证发布的质量字段。异常: StopIteration/关闭/格式/IO错误。141 """142 self.owner._require_open()143 # 只借出当前严格解码过的一行原字节；下一次推进前释放，不建立结果缓存。144 self._payload = None145 if self.closed:146 raise StopIteration147 # 真实关闭失败或读槽未抵销时禁止继续借出半关闭输入。148 if self._closing:149 raise ResultClosedError('结果读取器正在清理')150 try:151 payload = self.handle.readline(self.max_row_bytes + 1)152 if not payload:153 self.close()154 raise StopIteration155 if (len(payload) > self.max_row_bytes or len(payload) > self.owner.spool.max_row_bytes156 or not payload.endswith(b'\n')):157 raise ResultFormatError('结果行超限或未完整结束')158 row = decode_row(payload, self.max_row_bytes)159 if row_table(row) != self.table:160 raise ResultFormatError('结果暂存表与行类型不一致')161 self._payload = payload162 return row163 except BaseException as exc:164 # 解码/读取原异常优先；清理仍失败时父登记和恢复属性保存实际读取器。165 payload = row = None166 try:167 self.close()168 except BaseException as cleanup:169 exc.add_note(f'结果读取器异常清理附加失败: {type(cleanup).__name__}')170 if not self.closed:171 exc.row_iterator_cleanup_owner = self172 raise173 finally:174 # return已先求值；原TB不额外持有本层行和读取bytes。175 payload = row = None176177 def close(self) -> None:178 """抵销槽位、取消登记并关闭句柄；重复调用安全。179180 参数: 无。返回: None。异常: 原文件关闭/预算取消传播；失败句柄与未付清费用留待重试。181 """182 if self.closed:183 return184 self._closing = True185 self._payload = None186 try:187 # FD尚未关完时保留handle及读槽；关闭后抛取消也根据真实handle.closed记录状态。188 if self.handle is not None and not self._handle_closed:189 try:190 self.handle.close()191 finally:192 if self.handle.closed:193 self._handle_closed = True194 _release_owned_ram(self, self._reader_cost, '_reader_cost')195 # 未关闭资源的迭代器不能先从父登记消失；解除登记失败可由父再次close。196 self.owner._iterators.discard(self)197 finally:198 if self._handle_closed and self._held_cost == 0 and self not in self.owner._iterators:199 self.closed = True200201202class AnalysisResult:203 """保存一次分析的结果，提供可关闭的逐行迭代器和结果发布方法。204205 内部状态初始analysis_complete，发布失败为output_failed/cancelled。206 write可重复导出到同一结果目录；每次只读暂存，不重读旧输入。207 单线程使用；执行输入和旧writer成功关闭的前置条件由AnalysisRunner负责。208 _metadata_cost/_writer_cost为实际已付费用；自有读取器各自持有读槽，spool持有其209 固定/缓冲/归并费用。close进入_closing后拒绝读取或发布，所有层成功关闭才_closed。210 """211212 def __init__(self, spool: ResultSpool, metadata: RunMetadata, table_format: str = 'csv', *, artifact_exports=None,213 owned_work=None):214 """接管已封口暂存并写analysis_complete内部清单。215216 参数: spool为已完成暂存，metadata为真实运行事实，table_format为csv/jsonl/readable_csv；217 低层构造器保留csv默认以兼容现有调用，普通配置与AnalysisRunner默认readable_csv。218 artifact_exports为独立信号/报文导出仓；同样接管、按索引复制并校验、关闭。219 owned_work仅为新会话移交的私有根及创建身份，须为spool直接父目录；默认只接管spool目录。220 返回: None；初始化实例状态。异常: 配置拒绝前不接管spool；资源失败保原取消或原221 AnalysisError包装，持续清理失败附analysis_result_cleanup_owner供实际恢复。222 """223 if (not isinstance(spool, ResultSpool) or spool.state != 'analysis_complete'224 or spool._closed or spool._closing):225 raise AnalysisError('AnalysisResult需要已封口且未关闭的暂存')226 if not isinstance(metadata, RunMetadata) or table_format not in ('csv', 'jsonl', 'readable_csv'):227 raise ValueError('结果需RunMetadata和csv/jsonl/readable_csv格式')228 if owned_work is not None:229 if ((type(owned_work) is not tuple) or (len(owned_work) != 2) or (Path(owned_work[0]).resolve() != spool.path.resolve().parent) or (directory_identity(owned_work[0]) != owned_work[1])):230 raise ValueError('结果只能接管同次新会话登记的spool直接父目录')231 if (set(spool.tables) - set(SUMMARY_COLUMNS) or spool._summary_row_count != sum(spool.counts.values())232 or spool._directory_identity is None):233 raise ValueError('当前结果只接受已登记的精简表及对应行，不能混入旧完整结果')234 if artifact_exports is not None and (type(artifact_exports) is not ArtifactExportStore235 or artifact_exports._budget is not spool.budget or artifact_exports.closed or artifact_exports.state!='sealed'236 or artifact_exports.session_id not in (set(metadata.session_ids))):237 raise ValueError('新导出仓需同预算、同会话且封口')238 # 校验通过后才接管spool；首次预约前安装全部清理字段，已付费用从0开始。239 self.spool, self.metadata, self.table_format = spool, metadata, table_format240 self._work_removed = (metadata.retain_work)241 self._owned_work = owned_work or (spool.path, spool._directory_identity)242 self._pending_output_cleanup = None243 self._readable_writer = None # 发布失败时保留真实索引所有者，供close重试。244 self._published_counts = None245 self._presentation = None246 self._contributions_omitted = 'contributions' not in spool.tables247 self._output_tables = spool.tables248 self._artifact_exports=artifact_exports249 self._budget, self._held_cost, self._metadata_cost, self._writer_cost = spool.budget, 0, 0, 0250 self._iterators = set()251 self._closed = self._closing = False252 self._failure_count = 0253 self.status = 'analysis_complete'254 self.manifest_path = spool.path / 'analysis_manifest.json'255 if (not (metadata.detailed_diagnostics or metadata.retain_work)):256 self.manifest_path = None257 try:258 expected = metadata.estimated_bytes()259 if owned_work is not None:260 expected += 4096 + 64 * len(str(owned_work[0]))261 if artifact_exports is not None:262 expected+=65536+64*len(str(artifact_exports.path))263 _reserve_owned_ram(self, expected, '_metadata_cost')264 self._write_state(self.manifest_path, self.status)265 except BaseException as exc:266 spool._fail(exc)267 _cleanup_owner(self, exc, 'analysis_result_cleanup_owner', '结果对象构造')268 if isinstance(exc, (KeyboardInterrupt, SystemExit)):269 raise270 # 保持原非取消异常包装；未返回owner同时挂在外层，调用方无需猜测异常链。271 error = AnalysisError('结果内部清单写入失败')272 recovery = getattr(exc, 'analysis_result_cleanup_owner', None)273 if recovery is not None:274 error.analysis_result_cleanup_owner = recovery275 error.add_note('释放恢复后调用analysis_result_cleanup_owner.close()')276 raise error from exc277278 def _require_open(self) -> None:279 """拒绝关闭后迭代/发布，不自动重新打开已关闭对象。280281 参数: 无。返回: None。异常: ResultClosedError。282 """283 if self._closed or self._closing or self.spool._closed or self.spool._closing:284 raise ResultClosedError('AnalysisResult已关闭或正在清理')285286 def _manifest(self, status: str, outputs=(), coverage=None) -> dict:287 """构造有界清单副本，分开分析状态与发布状态。288289 参数: status为实际状态，outputs为固定文件记录，coverage为固定状态计数。290 返回: dict；元数据转换成本已申请。异常: 无预期异常。291 """292 manifest = {'status': status, 'run': self.metadata.to_dict(), 'python_version': platform.python_version(),293 'ended_at': datetime.now(timezone.utc).isoformat(), 'error_counts': {'output': self._failure_count},294 'row_counts': {table: self.spool.counts[table] for table in self._output_tables},295 'resources': self.spool.budget.snapshot(),296 'result_buffer_rows': self.spool.result_buffer_rows, 'peak_buffer_rows': self.spool.peak_buffer_rows,297 'table_format': self.table_format, 'outputs': list(outputs)}298 if (self.metadata.retain_work):299 manifest['spool_path'] = str(self.spool.path)300 manifest['spool_files'] = {table: str(self.spool.files[table]) for table in self._output_tables}301 if (self.metadata.detailed_diagnostics):302 manifest['coverage'] = coverage or {}303 defaults={table:version for table,_,version in reversed(tuple(SUMMARY_MODELS.values()))}304 versions={table:sorted(self.spool._summary_versions.get(table,{defaults[table]})) for table in self.spool.tables}305 versions={table:values[0] if len(values)==1 else values for table,values in versions.items()}306 if all(type(v) is int for v in versions.values()) and len(set(versions.values())) == 1:307 manifest['row_schema_version'] = next(iter(versions.values()))308 else:309 manifest['row_schema_versions'] = versions310 if self._artifact_exports is not None:311 manifest['exports']={'index_file':'exports.jsonl','directory':'exports',312 'record_count':self._artifact_exports.count,'device_label':self._artifact_exports.device_label}313 if self.table_format == 'readable_csv':314 from .readable_format import seconds, outcome_messages315 # 可读表不是旧行模型的逐行副本；以实际发布行数和专用表示版本为准。316 manifest.pop('row_schema_version', None)317 manifest.pop('row_schema_versions', None)318 if self._published_counts is not None:319 manifest['row_counts'] = dict(self._published_counts)320 if self._presentation is not None:321 manifest['presentation'] = self._presentation322 manifest['result_messages'] = outcome_messages(self.metadata.analysis_outcome)323 run = manifest['run']324 if 'session_boundaries' in run:325 run['session_boundaries_seconds'] = [(name, seconds(start), seconds(end))326 for name, start, end in run.pop('session_boundaries')]327 # 调试资源仍在显式诊断中提供，普通结果先呈现实际文件和业务概况。328 if not self.metadata.detailed_diagnostics:329 for key in ('python_version', 'resources', 'result_buffer_rows', 'peak_buffer_rows'):330 manifest.pop(key, None)331 return manifest332333 @property334 def run_manifest(self) -> dict:335 """返回当前状态快照，发布失败诊断路径另见manifest_path。336337 参数: 无。返回: 独立小型dict。异常: 关闭后仍可查看诊断，不读文件。338 """339 return self._manifest(self.status)340341 def _write_state(self, path: Path, status: str) -> None:342 """追加状态清单到暂存，每份保留且计入累计配额。343344 参数: path为本任务全新清单路径，status为实际状态。返回: None。345 异常: 配额/IO失败；不能生成虚假completed。346 """347 try:348 if (not (self.metadata.detailed_diagnostics or self.metadata.retain_work)):349 return350 payload = (_json(self._manifest(status)) + '\n').encode('utf-8')351 if len(payload) > self.spool.max_row_bytes:352 raise BudgetExceededError('运行清单超过有限元数据大小')353 with path.open('xb', buffering=0) as handle:354 self.spool._write_bytes(handle, payload)355 finally:356 # 失败状态诊断也只断本层文本/输入，保原IO错误与实际写次数。357 payload = self = path = status = None358359 def iter_events(self) -> Iterator[SummaryEventRow]:360 """创建独立事件流，不装入全量列表。361362 参数: 无。返回: SummaryEventRow迭代器。异常: ResultClosedError表示结果已关闭或正在关闭；BudgetExceededError表示无法为游标分配预算。363 创建或迭代期间OSError表示暂存文件读取失败，ResultFormatError表示暂存行格式损坏。364 """365 return _RowIterator(self, 'events')366367 def iter_sequences(self) -> Iterator[SummarySequenceRow]:368 """逐行读取请求配对结果；未配置配对时返回可关闭空游标。369370 参数: 无。返回: SummarySequenceRow迭代器。异常: ResultClosedError表示结果已关闭或正在关闭；BudgetExceededError表示无法为游标分配预算。371 创建或迭代期间OSError表示暂存文件读取失败，ResultFormatError表示暂存行格式损坏。372 """373 return _RowIterator(self, 'sequences')374375 def iter_segments(self) -> Iterator[SummarySegmentRow]:376 """创建独立片段流，保留全部父事件关联。377378 参数: 无。返回: SummarySegmentRow迭代器。异常: ResultClosedError表示结果已关闭或正在关闭；BudgetExceededError表示无法为游标分配预算。379 创建或迭代期间OSError表示暂存文件读取失败，ResultFormatError表示暂存行格式损坏。380 """381 return _RowIterator(self, 'segments')382383 def iter_calculations(self) -> Iterator[SummaryCalculationRow]:384 """创建计算摘要行流；同时检查value、unit、status、reason和quality。385386 参数: 无。返回: SummaryCalculationRow迭代器。异常: ResultClosedError表示结果已关闭或正在关闭；BudgetExceededError表示无法为游标分配预算。387 创建或迭代期间OSError表示暂存文件读取失败，ResultFormatError表示暂存行格式损坏。388 """389 return _RowIterator(self, 'calculations')390391392 def iter_contributions(self) -> Iterator[SummaryContributionRow]:393 """逐行回查真实贡献与skip，原序号和新求值序号分别保留。394395 参数: 无。返回: 独立有预算游标；逐行返回SummaryContributionRow。396 异常: ContributionsUnavailableError表示本次明确不生成明细；关闭、预算、397 IO或格式错误仍保留。原输入/解码器/求值器均不再调用。398 """399 self._require_open()400 if self._contributions_omitted:401 raise ContributionsUnavailableError('本次未生成贡献明细；请将outputs.contributions设为full后重新运行')402 return _RowIterator(self, 'contributions')403404 def iter_aggregations(self) -> Iterator['SummaryAggregationRow']:405 """按组读取聚合行及总体计数。参数: 无。返回: 受结果生命周期约束的迭代器。异常: 关闭或解码错误。"""406 return _RowIterator(self,'aggregations')407408 def iter_selections(self) -> Iterator[SummarySelectionRow]:409 """读取明确请求的真实样本选择行。参数: 无。返回: 有预算游标，未请求为空。异常: ResultClosedError表示结果关闭；BudgetExceededError表示游标预算不足；410 迭代时ResultFormatError表示暂存格式损坏，OSError表示文件读取失败。"""411 return _RowIterator(self,'selections')412413 def iter_input_trace(self) -> Iterator[SummaryInputRow]:414 """读取显式请求的所需输入追溯。参数: 无。返回: 按行产生SummaryInputRow的可关闭游标。415 异常: ResultClosedError表示结果已关闭；AnalysisError表示未请求diagnostics.trace。416 迭代期间可能因暂存损坏报告ResultFormatError，或传播OSError及预算错误。"""417 self._require_open()418 if 'input_trace' not in self.spool.tables:419 raise AnalysisError('本次未生成输入追溯；请设置diagnostics.trace后重新运行')420 return _RowIterator(self,'input_trace')421422 @property423 def contributions_available(self) -> bool:424 """说明本结果是否允许查询逐消息贡献，不将未生成和零行混为一谈。425426 参数: 无。返回: bool；默认未配置逐消息指标仍允许原空游标。427 异常: 无；关闭后可查看这项固定运行事实，但不能重新创建游标。428 """429 return not self._contributions_omitted430431 def _write_table(self, table: str, destination: Path, quality_handle, coverage: dict):432 """逐行生成单表和质量记录，异常或EOF关闭读取句柄。433434 参数: table为固定表，destination为新文件，quality_handle为质量输出。435 返回: 请求摘要时为文件事实，否则None。异常: 预算/编码/写入错误传播，输出with保证关闭。436 """437 iterator = _RowIterator(self, table)438 count = 0439 failure = None440 digest_output = None441 try:442 mode = 'x' if self.table_format == 'csv' else 'xb'443 options = {'encoding': 'utf-8', 'newline': ''} if mode == 'x' else {}444 with destination.open(mode, **options) as output:445 # 新版只在明确请求时随写摘要；CSV使用newline空串，无平台换行二次转换。446 if (self.metadata.integrity_diagnostics):447 digest_output=_DigestOutput(output);output=digest_output448 columns = SUMMARY_COLUMNS[table]449 versions=self.spool._summary_versions.get(table,())450 if 16 in versions:451 columns=columns+('extension_version',)452 if 10 in versions or 16 in versions:453 columns=columns+('coverage_status',)454 writer = csv.DictWriter(output, fieldnames=columns) if mode == 'x' else None455 if writer is not None:456 writer.writeheader()457 for row in iterator:458 # 普通/CSV路径已完整解码；内部legacy发布已核本次完整行证据。459 # 直接复用原字节，公共iterator仍交付严格模型，全部业务字段原样保留。460 payload = iterator._payload461 fields = ((_row_fields(row) if writer is not None else summary_fields(row)) if (writer is not None or quality_handle is not None)462 else None)463 if writer is not None:464 writer.writerow(fields)465 else:466 output.write(payload)467 # 状态/覆盖分布仅使用固定有限键，不按任意signal/rule建立无限dict。468 if quality_handle is not None:469 state_field = {'selections': 'selection', 'contributions': 'reason',470 'input_trace': 'record_type'}.get(table, 'status')471 state_key = table + ':' + fields[state_field]472 coverage[state_key] = coverage.get(state_key, 0) + 1473 quality_record = {key: fields[key] for key in ('sequence', 'dataset_id', 'entity_id', 'session_id', 'quality', state_field)}474 quality_record['table'] = table475 for key in ('selection', 'value_coverage', 'condition_coverage', 'requested_coverage', 'coverage_status'):476 if key in fields:477 quality_record[key] = fields[key]478 quality_handle.write(_json(quality_record) + '\n')479 count += 1480 if count != self.spool.counts[table]:481 raise ResultFormatError('输出表行数与分析暂存不一致')482 return digest_output.info(destination.name) if digest_output is not None else None483 except BaseException as exc:484 failure = exc485 raise486 finally:487 # 输出/编码原错误不能被读槽收尾取消覆盖，失败读取器仍留父登记供close。488 row = fields = payload = quality_record = None489 try:490 iterator.close()491 except BaseException as cleanup:492 if failure is None:493 raise494 failure.add_note(f'发布表读取器清理附加失败: {type(cleanup).__name__}')495496 def _file_info(self, path: Path) -> dict:497 """以固定读取块计算新输出哈希，不保留全文件内容。498499 参数: path为已关闭输出文件。返回: 名称/bytes/SHA256。500 异常: IO错误；调用侧已预留发布临时槽位。501 """502 if not self.metadata.detailed_diagnostics:503 return {'name': path.name, 'bytes': path.stat().st_size}504 digest = hashlib.sha256()505 chunk = None506 try:507 with path.open('rb') as handle:508 while chunk := handle.read(65536):509 digest.update(chunk)510 return {'name': path.name, 'bytes': path.stat().st_size, 'sha256': digest.hexdigest()}511 finally:512 # 保留原异常栈和摘要策略，只释放本层当前块。513 chunk = None514515 def _install_directory(self, temporary: Path, destination: Path) -> None:516 """同文件系统不覆盖安装；Windows短暂共享冲突保留原有限等待。517518 参数: temporary为目标父目录临时目录，destination为全新目标。519 返回: None。异常: NotImplementedError为平台缺少不覆盖移动能力；FileExistsError/IO错误为安装失败。520 边界: 仅原rename的Win5/32最多尝试六次，五次等待合计0.77秒；521 系统调用本身不受该等待时长约束。移动未完成时保留源目录；移动后的检查或关闭也可能报错，不能仅凭异常断定源目录仍在。522 副作用: 只重试当前路径安装，不重新读取、分析或生成任何输出文件。523 """524 # Windows rename对已存在目录拒绝；POSIX普通rename可覆盖空目录，不能静默复用。525 from .platform_files import move_no_replace526 if os.name != 'nt':527 move_no_replace(temporary,destination)528 return529 delays = (.02, .05, .1, .2, .4)530 for attempt in range(len(delays) + 1):531 # 每次等待后重新保护新出现的目标；检查之后的竞争仍由Windows rename拒绝。532 if destination.exists():533 raise FileExistsError(destination)534 try:535 move_no_replace(temporary, destination)536 return537 except PermissionError as exc:538 # 仅恢复短暂访问/共享冲突；权限长期不允许时仍保最后真实OS错误及现场。539 if getattr(exc, 'winerror', None) not in (5, 32):540 raise541 if attempt == len(delays):542 exc.add_note('Windows输出安装六次均被拒绝；已等待0.77秒，保留临时目录供诊断')543 raise544 time.sleep(delays[attempt])545546 def write(self, output_dir: str | Path) -> OutputManifest:547 """从暂存发布带时间文件名的结果，完成清单最后安装。548549 参数: output_dir为可复用输出目录；格式取本结果table_format。550 返回: 安装成功后的OutputManifest。异常: AnalysisError包含原错误链；关闭异常独立。551 副作用: 成功时创建正式输出目录；失败时按retain_work决定是否保留本次临时目录。552 不覆盖已有目标，不重新读取输入日志；关闭后的结果不能发布。553 """554 self._require_open()555 temporary = None556 temporary_identity = None557 publication = None558 manifest = None559 failure = None560 try:561 destination = Path(output_dir).absolute()562 if (not self.metadata.retain_work) and (destination.resolve().is_relative_to(Path(self._owned_work[0]).resolve())):563 raise ValueError('发布目录不能放在关闭后删除的本次私有工作目录内')564 from .publication import PublicationNames, output_root565 destination = output_root(destination)566 # 转换/CSV/质量/hash临时对象也计入账本，不只计算读取数组。567 _reserve_owned_ram(self, self.spool.reader_cost, '_writer_cost')568 destination.parent.mkdir(parents=True, exist_ok=True)569 temporary = Path(tempfile.mkdtemp(prefix=f'.{destination.name}.incomplete-', dir=destination.parent))570 temporary_identity = directory_identity(temporary)571 coverage = {}572 suffix = '.' + self.table_format573 # 仅发布实际请求的表；none不写空贡献文件，也不会打开其内部空暂存。574 paths = tuple(temporary / (table + suffix) for table in self._output_tables)575 quality_path = temporary / 'quality.jsonl'576 detailed = (self.metadata.detailed_diagnostics)577 quality_paths = (quality_path,) if detailed else ()578 written_infos={}579 with (quality_path.open('x', encoding='utf-8', newline='') if detailed else nullcontext(None)) as quality:580 if self.table_format == 'readable_csv':581 from .readable import ReadableWriter582 self._readable_writer = ReadableWriter(self, temporary, _RowIterator)583 paths = self._readable_writer.write(quality, coverage)584 written_infos.update(self._readable_writer.infos)585 self._published_counts = dict(self._readable_writer.counts)586 self._presentation = self._readable_writer.presentation()587 self._readable_writer.close()588 self._readable_writer = None589 else:590 for table, path in zip(self._output_tables, paths):591 info=self._write_table(table, path, quality, coverage)592 if info is not None: written_infos[path.name]=info593 # 原writer在run内已关闭；发布只复制并核验现成BLF/索引，无旧I/O调用。594 extra_paths = ()595 if self._artifact_exports is not None:596 extra_paths+=(self._artifact_exports.copy_to(temporary),)597 written_infos['exports.jsonl']=self._artifact_exports.index_info598 # 所有表和质量句柄已成功关闭；清单作为最后写入内容。599 infos = tuple(written_infos[path.name] if path.name in written_infos else self._file_info(path)600 for path in (*paths, *quality_paths, *extra_paths))601 # 文件全部关闭后分配本次名称；只改清单中的路径，不回读或改写数据行。602 publication = PublicationNames(destination, (*[p.name for p in temporary.iterdir()], 'run_manifest.json'))603 infos = tuple(dict(info, name=publication.names[info['name']]) for info in infos)604 completed = self._manifest('completed', infos, coverage)605 if 'exports' in completed:606 completed['exports'].update(index_file=publication.names['exports.jsonl'], directory=publication.names['exports'])607 completed['manifest_file'] = publication.names['run_manifest.json']608 with (temporary / 'run_manifest.json').open('x', encoding='utf-8', newline='') as handle:609 handle.write((json.dumps(completed, ensure_ascii=False, allow_nan=False, indent=2)610 if self.table_format == 'readable_csv' else _json(completed)) + '\n')611 # 返回模型也在安装前完成校验；安装成功后仅返回，避免校验错误遗留已发布completed。612 manifest = OutputManifest(self.metadata.run_id, destination, destination / publication.names['run_manifest.json'],613 tuple(destination / publication.names[path.name] for path in (*paths, *quality_paths, *extra_paths)),614 tuple(self._published_counts.items()) if self.table_format == 'readable_csv'615 else tuple((table, self.spool.counts[table]) for table in self._output_tables))616 publication.publish(temporary, 'run_manifest.json', self._install_directory)617 try:618 temporary.rmdir()619 except OSError:620 # 正式清单已发布；空暂存交给已有close责任重试，不将结果谎报为未发布。621 self._pending_output_cleanup = (temporary, temporary_identity)622 self._closing = True623 self.status = 'analysis_complete'624 return manifest625 except BaseException as exc:626 failure = exc627 if publication is not None:628 try:629 publication.close()630 except BaseException as cleanup:631 exc.add_note('输出命名锁释放失败：' + str(cleanup))632 published_manifest = manifest if publication is not None and publication.completed else None633 if published_manifest is not None:634 exc.completed_manifest = published_manifest635 if self._readable_writer is not None:636 try:637 self._readable_writer.close()638 self._readable_writer = None639 except BaseException as cleanup:640 self._closing = True641 exc.analysis_result_cleanup_owner = self642 exc.add_note('可读关联索引关闭失败：' + type(cleanup).__name__)643 raw_failure = False644 raw_failure=raw_failure or (self._artifact_exports is not None and645 getattr(exc,'artifact_export_store_cleanup_owner',None) is self._artifact_exports)646 if raw_failure:647 # 取消也必须保最高结果owner；不能到普通异常包装后才补这条真实责任。648 exc.analysis_result_cleanup_owner = self649 self.status = ('analysis_complete' if published_manifest is not None else650 'cancelled' if isinstance(exc, (KeyboardInterrupt, SystemExit)) else 'output_failed')651 self._failure_count += 1652 # 安装失败时移除临时completed名称，把现存清单改标incomplete；只操作本次新文件。653 if (temporary is not None) and (not self.metadata.retain_work):654 self._pending_output_cleanup = (temporary, temporary_identity)655 try:656 remove_owned_directory(*self._pending_output_cleanup)657 self._pending_output_cleanup = None658 except BaseException as cleanup:659 self._closing = True660 exc.analysis_result_cleanup_owner = self661 exc.add_note('输出临时目录清理失败，可重试close：'+type(cleanup).__name__)662 elif temporary is not None:663 marker = temporary / 'run_manifest.json'664 try:665 if marker.exists():666 with marker.open('w', encoding='utf-8') as handle:667 handle.write(_json({'status': self.status, 'run_id': self.metadata.run_id}) + '\n')668 with (temporary / 'failure.json').open('x', encoding='utf-8') as handle:669 handle.write(_json({'status': self.status, 'error_type': type(exc).__name__}) + '\n')670 except BaseException as diagnostic:671 # 诊断写入失败不替换发布原错；原磁盘失败限制及incomplete目录保持。672 if not isinstance(diagnostic, OSError):673 exc.add_note(f'发布临时诊断附加失败: {type(diagnostic).__name__}')674 try:675 path = self.spool.path / f'output_failure-{self._failure_count:06d}.json'676 self._write_state(path, self.status)677 self.manifest_path = (None if (not (self.metadata.detailed_diagnostics or self.metadata.retain_work)) else path)678 except BaseException as diagnostic:679 if not isinstance(diagnostic, (OSError, BudgetExceededError)):680 exc.add_note(f'发布状态诊断附加失败: {type(diagnostic).__name__}')681 if isinstance(exc, (KeyboardInterrupt, SystemExit)):682 raise683 error = AnalysisError(f'结果发布失败，状态{self.status}')684 if published_manifest is not None:685 error.completed_manifest = published_manifest686 if raw_failure or self._pending_output_cleanup is not None or self._readable_writer is not None:687 # 原始复制的实际FD仍由本结果的仓持有；包装不得丢最高恢复入口。688 error.analysis_result_cleanup_owner = self689 failure = error690 raise error from exc691 finally:692 # 局部with已关闭真实输出；发布槽的实际余额由同一owner保存供失败后close。693 try:694 _release_owned_ram(self, self._writer_cost, '_writer_cost')695 except BaseException as cleanup:696 if publication is not None and publication.completed:697 cleanup.completed_manifest = manifest698 if self._writer_cost:699 self._closing = True700 target = cleanup if failure is None else failure701 target.analysis_result_cleanup_owner = self702 if failure is None:703 raise704 failure.add_note(f'发布workspace清理附加失败: {type(cleanup).__name__}')705706 def close(self) -> None:707 """关闭活跃游标并释放资源；默认清理本次自有暂存，保留原始输入与已发布结果。708709 参数: 无。返回: None；重复调用安全。diagnostics.retain_work为True时保留工作现场。710 异常: 关闭失败仍尝试其余句柄，最后抛首个原错误；未完成清理时可重试close。711 """712 if self._closed:713 return714 # 清理中拒绝新读取/发布；尝试全部读取器，但不提前丢失失败句柄或自有费用。715 self._closing = True716 error = None717 for iterator in tuple(self._iterators):718 try:719 iterator.close()720 except BaseException as exc:721 if error is None:722 error = exc723 else:724 error.add_note(f'其他结果读取器关闭失败: {type(exc).__name__}')725 if error is not None:726 # 原首错和关闭顺序不变，最高结果仍保留失败的真实游标供恢复。727 error.analysis_result_cleanup_owner = self728 raise error729 try:730 # 各层只释放自己的实际申请，子spool从不抵销本层或调用方的借用余额。731 if self._readable_writer is not None:732 self._readable_writer.close()733 self._readable_writer = None734 if self._artifact_exports is not None:735 self._artifact_exports.close()736 if not self._artifact_exports.closed:737 raise AnalysisError('新导出仓没有真实关闭')738 _release_owned_ram(self, self._writer_cost, '_writer_cost')739 _release_owned_ram(self, self._metadata_cost, '_metadata_cost')740 self.spool.close()741 if self._pending_output_cleanup is not None:742 remove_owned_directory(*self._pending_output_cleanup)743 self._pending_output_cleanup = None744 if not self._work_removed:745 remove_owned_directory(*self._owned_work)746 self._work_removed = True747 except BaseException as exc:748 # 新原始仓持续关闭失败仍从本结果恢复，不要求调用方绕过其真实父。749 exc.analysis_result_cleanup_owner = self750 raise751 finally:752 if (not self._iterators and self._held_cost == 0 and self.spool._closed753 and self._work_removed and self._pending_output_cleanup is None754 and (self._artifact_exports is None or self._artifact_exports.closed)):755 self._closed = True756757 def __enter__(self) -> 'AnalysisResult':758 """进入结果上下文，不再次执行输入读取。759760 参数: 无。返回: self。异常: ResultClosedError为已关闭。761 """762 self._require_open()763 return self764765 def __exit__(self, exc_type, exc_value, traceback):766 """退出上下文时释放资源，保留业务原异常。767768 参数: exc_type/exc_value/traceback为上下文异常。返回: False，不吞异常。769 异常: 无原业务异常时关闭错误传播，否则原业务异常优先。770 """771 try:772 self.close()773 except BaseException as cleanup:774 if exc_value is None:775 raise776 # 原上下文异常优先，但持续关闭失败仍说明实际恢复入口。777 exc_value.add_note(f'结果上下文清理附加失败: {type(cleanup).__name__}')778 if not self._closed:779 exc_value.analysis_result_cleanup_owner = self780 return False781