chronaxis.results.readable
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""将结果流关联成可直接阅读的CSV，不重新读取或计算输入。23事件、窗口及计算的既有排序不同，先用临时SQLite索引关联，再逐行发布。4索引只有本次结果，固定缓存、禁止mmap、写前预付页面；关闭后删除，不发布5原表副本。CSV、JSONL旧路径不调用本模块。所有权保存在AnalysisResult供失败恢复。6"""78# SQLite属于Python标准库；只创建本次发布目录中的私有临时文件。9import csv10import hashlib11import json12import sqlite313from contextlib import contextmanager14from itertools import chain15from math import ceil16from .budget import _reserve_owned_ram, _release_owned_ram17from .errors import ResultFormatError18from .summary import summary_fields19from .readable_format import STATES, REASONS, seconds, explanation, quality_text2021IDENTITY = ('dataset_id', 'entity_id', 'session_id')22LABELS = ('数据集', '对象', '会话')23FILES = {'events': '事件结果.csv', 'segments': '窗口结果.csv', 'calculations': '会话结果.csv',24 'sequences': '配对结果.csv', 'selections': '选择结果.csv', 'aggregations':'聚合结果.csv',25 'contributions': '贡献明细.csv', 'input_trace': '输入追溯.csv'}262728def _json(value):29 """编码临时索引值。参数: 平坦字段。返回: JSON文本。异常: 非有限值拒绝。"""30 return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(',', ':'))313233class ReadableWriter:34 """拥有单次发布的关联索引及有限列定义；close可重试，不拥有输入或正式结果。"""3536 def __init__(self, owner, directory, iterator_factory):37 """仅安装可恢复字段，资源在prepare取得。3839 参数: owner为AnalysisResult，directory为本次临时目录，iterator_factory为结果游标。40 返回: 未准备的writer。异常: 无文件副作用。41 """42 self.owner, self.directory, self.iterator_factory = owner, directory, iterator_factory43 self.path = directory / '.readable-index.sqlite'44 self.connection = None45 self._pending_reader = None46 self._pending_output = None47 self._budget = owner.spool.budget48 self._held_cost = self._ram_cost = 049 self.pages = 050 self.closed = False51 self.metrics = {'session': {}, 'event': {}, 'segment': {}}52 self.counts, self.infos = {}, {}53 self.confirmation = False54 self.quality_handle = None55 self.coverage = None5657 def _execute(self, sql, values=(), size=0):58 """写前预留SQLite页面，上限由数据库本身强制执行。5960 参数: sql/values为受控语句，size为本条JSON字符数。返回: cursor。61 异常: 配额、SQL、IO或取消传播。预留包含B树分裂余量，不按总行数预占RAM。62 """63 # page_count包含事务中已分配页；保守覆盖UTF8值、索引键及B树分裂。64 count = self.connection.execute('PRAGMA page_count').fetchone()[0]65 needed = count + 16 + ceil(size * 12 / 4096)66 if needed > self.pages:67 self._budget.reserve_spool((needed - self.pages) * 4096)68 self.pages = needed69 self.connection.execute(f'PRAGMA max_page_count={needed}')70 return self.connection.execute(sql, values)7172 @contextmanager73 def _reader(self, table, *, indexed=False):74 """拥有一次结果流或索引游标，关闭失败留给writer重试。7576 参数: table为当前表；indexed选择自有SQLite游标，默认结果流。77 返回: 上下文中的可迭代读者。异常: 读错优先，关闭错附记；恢复owner保留。78 """79 if self._pending_reader is not None:80 raise RuntimeError('可读结果仍有未关闭的游标，请先重试关闭')81 reader = failure = None82 try:83 reader = (self.connection.execute('SELECT n,payload FROM objects WHERE kind=? ORDER BY n', (table,))84 if indexed else self.iterator_factory(self.owner, table))85 self._pending_reader = reader86 yield reader87 except GeneratorExit:88 # 主动结束生成器时，关闭失败必须让调用方看到，不能吞掉资源错误。89 raise90 except BaseException as error:91 failure = error92 raise93 finally:94 try:95 if reader is not None:96 reader.close()97 except BaseException as cleanup:98 target = cleanup if failure is None else failure99 target.readable_cleanup_owner = self100 if failure is None:101 raise102 failure.add_note('可读结果游标清理附加失败: ' + type(cleanup).__name__)103 else:104 self._pending_reader = None105 finally:106 # 仍未关闭的资源保存在明确owner；错误帧不再另持一份游标引用。107 reader = failure = None108109 @contextmanager110 def _output(self, path):111 """持有唯一输出句柄，写入与关闭同时失败时保留原写入错误。112113 参数: path为本次新CSV路径。返回: 独占创建的二进制句柄。114 异常: 原写入错误优先；未关闭句柄归writer，关闭可以重试。115 """116 if self._pending_output is not None:117 raise RuntimeError('可读结果仍有未关闭的输出，请先重试关闭')118 handle = failure = None119 try:120 handle = path.open('xb')121 self._pending_output = handle122 yield handle123 except BaseException as error:124 failure = error125 raise126 finally:127 try:128 if handle is not None:129 handle.close()130 except BaseException as cleanup:131 original = cleanup if failure is None else failure132 original.readable_cleanup_owner = self133 if failure is None:134 raise135 failure.add_note('可读CSV输出关闭附加失败: ' + type(cleanup).__name__)136 finally:137 if handle is not None and handle.closed:138 self._pending_output = None139 handle = failure = None140141 def _rows(self, table):142 """按原游标逐条读已计算结果，保持严格解析和真实关闭。143144 参数: table为启用表。返回: 平坦字段生成器。异常: 读取/校验/关闭错误传播。145 """146 row = fields = diagnostic = None147 try:148 with self._reader(table) as iterator:149 count = 0150 for row in iterator:151 count += 1152 fields = summary_fields(row)153 # 显式质量诊断仍随唯一必要读取得到，不另扫一遍结果。154 if self.quality_handle is not None:155 state_field = {'selections': 'selection', 'contributions': 'reason',156 'input_trace': 'record_type'}.get(table, 'status')157 key = table + ':' + fields[state_field]158 self.coverage[key] = self.coverage.get(key, 0) + 1159 diagnostic = {key: fields[key] for key in ('sequence', *IDENTITY, 'quality', state_field)}160 diagnostic['table'] = table161 self.quality_handle.write(_json(diagnostic) + '\n')162 yield fields163 if count != self.owner.spool.counts.get(table, 0):164 raise ResultFormatError('可读导出行数与暂存不一致')165 finally:166 # 生成器结束或出错后，不让本层异常帧保留刚读出的行。167 row = fields = diagnostic = iterator = None168169 def prepare(self):170 """建立必要关联和列定义，不读取原始日志。171172 参数: 无。返回: None。异常: 重复身份、重复指标、配额/IO传播。173 """174 if not any(self.owner.spool.counts.get(table, 0) for table in ('events', 'segments', 'sequences', 'calculations')):175 # 只有逐条明细或全空结果时没有关联消费者，连空索引也不创建。176 self.confirmation = True177 return178 _reserve_owned_ram(self, 262144, '_ram_cost')179 # 禁止系统临时排序文件和mmap；后续查询只走已声明索引。180 self._budget.reserve_spool(65536)181 self.pages = 16182 self.connection = sqlite3.connect(self.path, isolation_level=None)183 self.connection.execute('PRAGMA page_size=4096')184 self.connection.execute('PRAGMA max_page_count=16')185 self.connection.execute('PRAGMA cache_size=-64')186 self.connection.execute('PRAGMA mmap_size=0')187 self.connection.execute('PRAGMA journal_mode=OFF')188 self.connection.execute('PRAGMA synchronous=OFF')189 self.connection.execute('PRAGMA temp_store=MEMORY')190 self._execute('CREATE TABLE objects(kind TEXT,d TEXT,e TEXT,s TEXT,id TEXT,n INTEGER,payload TEXT,'191 'PRIMARY KEY(kind,d,e,s,id),UNIQUE(kind,n)) WITHOUT ROWID')192 self._execute('CREATE TABLE metrics(d TEXT,e TEXT,s TEXT,scope TEXT,parent TEXT,name TEXT,payload TEXT,'193 'PRIMARY KEY(d,e,s,scope,parent,name)) WITHOUT ROWID')194 self.connection.execute('BEGIN')195 # 每表编号来自既有稳定结果排序，不截短哈希，也不跨会话错联。196 for table in ('events', 'segments', 'sequences', 'calculations'):197 if table not in self.owner.spool.tables:198 continue199 for number, row in enumerate(self._rows(table), 1):200 payload = _json(row)201 identity = tuple(row[name] for name in IDENTITY)202 if table == 'calculations':203 name, scope = row['calculation_id'], row['scope']204 definition = (row['unit'], row['operation'])205 if scope != 'session':206 self._number('events' if scope == 'event' else 'segments', row, row['parent_id'])207 if name not in self.metrics[scope]:208 _reserve_owned_ram(self, 2048 + 64 * (len(name) + len(row['unit'])), '_ram_cost')209 self.metrics[scope][name] = definition210 elif self.metrics[scope][name] != definition:211 raise ResultFormatError('同名计算的单位或算子不一致')212 self._execute('INSERT INTO metrics VALUES(?,?,?,?,?,?,?)',213 (*identity, scope, row['parent_id'] or '', name, payload), len(payload))214 # session本身也是关联对象；多项指标只建立一次会话结果行。215 if scope == 'session':216 self._execute('INSERT OR IGNORE INTO objects VALUES(?,?,?,?,?,?,?)',217 ('calculations', *identity, '', number, _json(dict(zip(IDENTITY, identity)))), len(payload))218 else:219 key = {'events': 'event_id', 'segments': 'segment_id', 'sequences': 'result_id'}[table]220 self._execute('INSERT INTO objects VALUES(?,?,?,?,?,?,?)',221 (table, *identity, row[key], number, payload), len(payload))222 # 一次发布共用事务，避免每条结果触发一次磁盘提交；缓存与页面上限仍逐次生效。223 self.connection.execute('COMMIT')224 self.confirmation = bool({'sequences', 'segments'} & set(self.owner.spool.tables)) or not self.metrics['event'] or any(225 operation != 'integral' for _, operation in self.metrics['event'].values())226 if not self.confirmation:227 # 同一计划可能还有独立事件；不能因别的规则做了积分就隐藏其确认时刻。228 self.confirmation = self.connection.execute(229 "SELECT 1 FROM objects o WHERE o.kind='events' AND NOT EXISTS "230 "(SELECT 1 FROM metrics m WHERE m.d=o.d AND m.e=o.e AND m.s=o.s "231 "AND m.scope='event' AND m.parent=o.id) LIMIT 1").fetchone() is not None232233 def _number(self, kind, row, identifier):234 """将完整父ID映射为本次表编号，缺失关系直接报错。235236 参数: kind为events/segments，row含完整身份，identifier可为None。237 返回: 编号或空字符串。异常: ResultFormatError为悬空父关联。238 """239 if identifier is None:240 return ''241 if self.connection is None:242 raise ResultFormatError('可读输出缺少父结果')243 found = self.connection.execute('SELECT n FROM objects WHERE kind=? AND d=? AND e=? AND s=? AND id=?',244 (kind, *(row[k] for k in IDENTITY), identifier)).fetchone()245 if found is None:246 raise ResultFormatError(f'可读输出缺少{kind}父结果，不能静默丢失关联')247 return found[0]248249 def _metric_columns(self, scope):250 """创建已付预算的有限指标列。参数: scope。返回: 列名列表。异常: 无。"""251 columns = []252 for name, (unit, operation) in self.metrics[scope].items():253 # min_time是时间范围内最小值，不是最小时刻；指标单位必须原样保留。254 columns.extend((f'{name}（{unit}）', f'{name}：状态', f'{name}：说明'))255 return columns256257 def _metric_values(self, scope, row, parent):258 """按同一完整身份取指标，未产生的指标不伪装为完整或零。259260 参数: scope/row/parent定位范围。返回: 有限单行列表。异常: 关联/JSON错误传播。261 """262 values = []263 for name, (unit, operation) in self.metrics[scope].items():264 found = self.connection.execute('SELECT payload FROM metrics WHERE d=? AND e=? AND s=? AND scope=? AND parent=? AND name=?',265 (*(row[k] for k in IDENTITY), scope, parent or '', name)).fetchone()266 if found is None:267 values.extend(('', '无此项结果', ''))268 continue269 metric = json.loads(found[0])270 value = metric['value']271 values.extend((value, STATES.get(metric['status'], metric['status']), explanation(metric)))272 return values273274 def _base(self, table, row, number):275 """形成业务列，保留事件/窗口/配对不同含义。276277 参数: table、平坦row、本次number。返回: 单行列表。异常: 悬空关联传播。278 """279 if table == 'calculations':280 return self._metric_values('session', row, '')281 if table == 'events':282 values = [number, row['rule_id'], seconds(row['start_ns'])]283 if self.confirmation:284 values.append(seconds(row['confirmed_ns']))285 values.extend((seconds(row['end_ns']), STATES[row['status']], quality_text(row['quality'])))286 return values + self._metric_values('event', row, row['event_id'])287 if table == 'segments':288 return [number, row['window_id'], seconds(row['start_ns']), seconds(row['end_ns']),289 STATES[row['status']], '、'.join(str(self._number('events', row, p)) for p in row['parent_event_ids']),290 quality_text(row['quality'])] + self._metric_values('segment', row, row['segment_id'])291 return [number, row['rule_id'], self._number('events', row, row['request_id']),292 self._number('events', row, row['response_id']), seconds(row['decided_ns']), seconds(row['delay_ns']),293 STATES[row['status']], quality_text(row['quality'])]294295 def _write(self, table, columns, rows):296 """借用行流写最终CSV，并随写计算显式请求的摘要。297298 参数: table/columns/rows为本次有限列和借用的流；不关闭调用者的rows。299 返回: 路径。异常: 编码、IO或取消原样传播；仅清理本层编码缓冲。300 """301 path = self.directory / FILES[table]302 digest = hashlib.sha256() if self.owner.metadata.integrity_diagnostics else None303 count, size = 0, 0304 buffer = writer = values = payload = failure = None305 try:306 # BOM使Windows表格软件能识别中文；chain不把关闭操作转发给借用流。307 with self._output(path) as handle:308 header = b'\xef\xbb\xbf'309 handle.write(header)310 size += len(header)311 if digest is not None:312 digest.update(header)313 import io314 buffer = io.StringIO(newline='')315 writer = csv.writer(buffer)316 for index, values in enumerate(chain((columns,), rows)):317 buffer.seek(0)318 buffer.truncate(0)319 writer.writerow(values)320 payload = buffer.getvalue().encode('utf-8')321 if handle.write(payload) != len(payload):322 raise OSError('可读CSV写入不完整')323 size += len(payload)324 if digest is not None:325 digest.update(payload)326 if index:327 count += 1328 self.counts[table] = count329 info = dict(name=path.name, bytes=size)330 if digest is not None:331 info['sha256'] = digest.hexdigest()332 self.infos[path.name] = info333 return path334 except BaseException as exc:335 failure = exc336 raise337 finally:338 # traceback可以继续保留本帧，但不继续持有整行和CSV缓冲；rows由调用方拥有。339 rows = columns = values = payload = writer = None340 try:341 if buffer is not None:342 buffer.close()343 except BaseException as cleanup:344 if failure is None:345 raise346 failure.add_note('可读CSV缓冲清理附加失败: ' + type(cleanup).__name__)347 finally:348 buffer = failure = None349350 def _write_owned_rows(self, table, columns, *, detail=False):351 """写入本对象刚创建的行流，成功、失败或取消后结束该流。352353 参数: table/columns定位本次表；detail选择明细流，默认主表。354 返回: 已写路径。异常: 原写入异常优先；关闭失败由原游标/父结果继续承担。355 """356 rows = failure = None357 try:358 rows = self._detail_rows(table) if detail else self._main_rows(table)359 return self._write(table, columns, rows)360 except BaseException as exc:361 failure = exc362 raise363 finally:364 try:365 if rows is not None:366 rows.close()367 except BaseException as cleanup:368 if failure is None:369 raise370 failure.add_note('可读结果行流清理附加失败: ' + type(cleanup).__name__)371 finally:372 rows = failure = columns = None373374 def write(self, quality_handle=None, coverage=None):375 """发布全部关联主表；明细由后续专门分支逐条处理。376377 参数: quality_handle/coverage为显式质量诊断，默认不生成。返回: 路径tuple。378 异常: 关联/预算/IO错误传播。379 """380 self.quality_handle, self.coverage = quality_handle, coverage381 self.prepare()382 identity = self.owner.spool.identity_summary or (None, None, None)383 self.varying = tuple(index for index, value in enumerate(identity) if value is None)384 paths = []385 columns = {386 'calculations': self._metric_columns('session'),387 'events': ['区间', '规则', '开始时间（s）'] + (['确认时间（s）'] if self.confirmation else [])388 + ['结束时间（s）', '区间状态', '说明'] + self._metric_columns('event'),389 'segments': ['窗口编号', '窗口', '开始时间（s）', '结束时间（s）', '窗口状态', '父事件区间', '说明']390 + self._metric_columns('segment'),391 'sequences': ['配对编号', '规则', '请求区间', '响应区间', '判定时间（s）', '延迟（s）', '配对状态', '说明'],392 }393 for table in ('events', 'segments', 'calculations', 'sequences'):394 if table not in self.owner.spool.tables:395 continue396 if table == 'calculations' and not self.metrics['session']:397 continue398 paths.append(self._write_owned_rows(table, [LABELS[i] for i in self.varying] + columns[table]))399 # 明细无需重排或宽表聚合，直接从原暂存逐行写出，不进入SQLite索引。400 detail_columns = {401 'aggregations': ['聚合项','分组','结果','单位','状态','原因','候选数','参与数','定义排除数','质量排除数',402 '参与的部分值','匹配','确认超时','观测不足','未决','已完整发现总体','质量说明'],403 'selections': ['计算项', '统计范围', '所属区间', '时间（s）', '值', '单位', '数值状态', '选择状态', '说明'],404 'contributions': ['计算项', '来源', '原记录序号', '求值序号', '贡献值', '说明'],405 'input_trace': ['来源', '原记录序号', '时间（s）', '记录类型', '输入位置', '字段值', '说明'],406 }407 for table, names in detail_columns.items():408 if table in self.owner.spool.tables:409 paths.append(self._write_owned_rows(table, [LABELS[i] for i in self.varying] + names, detail=True))410 return tuple(paths)411412 def _detail_rows(self, table):413 """直接输出请求的明细，保留未知、值类型及必要定位。414415 参数: table为明细表。返回: 有界CSV行。异常: 缺失父对象或原读取错误。416 """417 rows = row = values = failure = None418 try:419 rows = self._rows(table)420 for row in rows:421 if table == 'aggregations':422 values=[row['aggregation_id'],_json(dict(zip(row['group_keys'],row['group_values']))),423 row['value'],row['unit'],{'complete':'完整','partial':'部分','indeterminate':'不能确定','empty':'确定为空'}[row['status']],424 {'ok':'正常','partial_population':'总体不完整','no_observation':'无观测','insufficient_data':'观测不足',425 'empty_population':'无参与值','insufficient_count':'样本数不足','numeric_error':'数值不能确定'}[row['reason']],426 row['candidate_count'],row['included_count'],row['definition_excluded_count'],row['quality_excluded_count'],427 row['included_partial_count'],row['matched_count'],row['timeout_count'],row['insufficient_observation_count'],428 row['pending_count'],'是' if row['discovery_complete'] else '否',quality_text(row['quality'])]429 elif table == 'selections':430 scope = row['scope']431 parent = '' if scope == 'session' else self._number(432 'events' if scope == 'event' else 'segments', row, row['parent_id'])433 values = [row['calculation_id'], {'session': '会话', 'event': '事件', 'segment': '窗口'}[scope],434 parent, seconds(row['time_ns']), row['value'], row['unit'],435 '有效' if row['valid'] else '无效', STATES[row['selection']], explanation(row)]436 elif table == 'contributions':437 values = [row['calculation_id'], row['source'], row['original_sequence'], row['evaluation_index'],438 row['value'], explanation(row)]439 else:440 kinds = {'decoded_message': '解码报文', 'signal_sample': '信号样本', 'signal_clock': '信号时刻'}441 values = [row['source'], row['original_sequence'], seconds(row['time_ns']), kinds[row['record_type']],442 row['locator'], row['values_json'], quality_text(row['quality'])]443 yield [row[IDENTITY[i]] for i in self.varying] + values444 except GeneratorExit:445 # 外层主动关闭时，内部关闭错误仍须传回实际所有者。446 raise447 except BaseException as exc:448 failure = exc449 raise450 finally:451 try:452 if rows is not None:453 rows.close()454 except BaseException as cleanup:455 if failure is None:456 raise457 failure.add_note('可读明细输入清理附加失败: ' + type(cleanup).__name__)458 finally:459 rows = row = values = failure = None460461 def _main_rows(self, table):462 """按索引顺序产生一行，查询不在RAM排序全表。463464 参数: table为主表。返回: CSV行生成器。异常: JSON/关联错误。465 """466 payload = row = None467 try:468 if self.connection is None:469 return470 with self._reader(table, indexed=True) as cursor:471 for number, payload in cursor:472 row = json.loads(payload)473 yield [row[IDENTITY[i]] for i in self.varying] + self._base(table, row, number)474 finally:475 # 生成器结束或出错后，不让本层异常帧保留刚读出的行。476 payload = row = cursor = None477478 def presentation(self):479 """仅返回共同身份和必要使用说明，不增加ID映射副本。480481 参数: 无。返回: 清单字典。异常: 无。482 """483 return dict(version=1, time_unit='s', time_coordinates='保留原分析时间坐标，不从零重排，不推断日期或时区',484 empty_cell='无数值；请结合本指标状态和说明，不等于0',485 common_identity={IDENTITY[i]: value for i, value in enumerate(self.owner.spool.identity_summary or ())486 if value is not None})487488 def close(self):489 """关闭并删除本次索引，失败保留真实owner供重试。490491 参数: 无。返回: None。异常: IO/释放错误传播，不删除输入或其他文件。492 """493 if self.closed:494 return495 if self._pending_output is not None:496 handle = self._pending_output497 try:498 handle.close()499 finally:500 if handle.closed:501 self._pending_output = None502 if self._pending_reader is not None:503 self._pending_reader.close()504 self._pending_reader = None505 if self.connection is not None:506 self.connection.close()507 self.connection = None508 self.path.unlink(missing_ok=True)509 self.metrics.clear()510 _release_owned_ram(self, self._ram_cost, '_ram_cost')511 self.closed = True512