vent_agent_2.0/extensions/main_fan/history_compaction.py

"""历史数据的无损证据压缩与观测状态区间；不判定整机切换或合规性。"""
from collections import Counter, defaultdict
from contextlib import ExitStack
from copy import deepcopy
import hashlib
import json
import math
from statistics import median
from zipfile import ZIP_DEFLATED, ZipFile

from extensions.main_fan import equipment, gateway_queries

VERSION = 'main-fan-observed-intervals-v2-open-close-labels'


class RecordReader:
    """兼容原逐段JSON和单个ZIP中的原件，不解压出数百个文件。"""
    def __init__(self, resolve):
        self.resolve = resolve
        self.stack = ExitStack()
        self.archives = {}

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return self.stack.__exit__(*args)

    def bytes(self, record, kind='data'):
        if record.get('archive'):
            archive = record['archive']
            if archive['path'] not in self.archives:
                self.archives[archive['path']] = self.stack.enter_context(ZipFile(self.resolve(archive['path'])))
            item = archive['members'][kind]
            content = self.archives[archive['path']].read(item['name'])
            expected = item['sha256']
        else:
            virtual = next(path for path in record['files'] if path.endswith('/' + kind + '.json'))
            content = self.resolve(virtual).read_bytes()
            expected = record['files'][virtual]
        if hashlib.sha256(content).hexdigest() != expected:
            raise ValueError('历史证据指纹不一致：' + record['key'] + '/' + kind)
        return content

    def data(self, record):
        return json.loads(self.bytes(record))


def compact_evidence(records, archive_path, resolve):
    """原件逐项校验后归入一个ZIP；调用者先持久化新索引，再清理本轮暂存文件。"""
    completed = [row for row in records if row['status'] == 'completed']
    if not completed:
        return records, {}, [], {}
    if all(row.get('archive') for row in completed) and len({row['archive']['path'] for row in completed}) == 1:
        files = {path: sha for row in completed for path, sha in row['files'].items()}
        with ZipFile(resolve(completed[0]['archive']['path'])) as archive:
            original_bytes = sum(item.file_size for item in archive.infolist())
        # 同一批次重试不改写已经完成的压缩证据。
        return records, files, [], {'archive_path': completed[0]['archive']['path'],
            'compressed_bytes': resolve(completed[0]['archive']['path']).stat().st_size,
            'original_file_count': len(completed) * 3, 'original_bytes': original_bytes}
    destination = resolve(archive_path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_name(destination.name + '.writing')
    result, cleanup, size = deepcopy(records), [], 0
    try:
        with RecordReader(resolve) as reader, ZipFile(temporary, 'w', ZIP_DEFLATED, compresslevel=6) as archive:
            for record in result:
                if record['status'] != 'completed':
                    continue
                members = {}
                for kind in ('request', 'response', 'data'):
                    content = reader.bytes(record, kind)
                    name = record['key'] + '/' + kind + '.json'
                    archive.writestr(name, content)
                    size += len(content)
                    members[kind] = {'name': name, 'sha256': hashlib.sha256(content).hexdigest()}
                cleanup.extend(record['files'] if not record.get('archive') else [])
                record['archive'] = {'path': archive_path, 'members': members}
                record.pop('data_path', None)
                record.pop('python_path', None)
        with ZipFile(temporary) as archive:
            if archive.testzip() is not None:
                raise ValueError('历史压缩证据校验失败')
        temporary.replace(destination)
    finally:
        temporary.unlink(missing_ok=True)
    sha = hashlib.sha256(destination.read_bytes()).hexdigest()
    for record in result:
        if record['status'] == 'completed':
            record['files'] = {archive_path: sha}
    return result, {archive_path: sha}, cleanup, {'archive_path': archive_path,
        'original_file_count': len(completed) * 3, 'original_bytes': size,
        'compressed_bytes': destination.stat().st_size}


def _state(sample, system):
    if system == 'safety':
        if sample.get('status_label') not in (None, '正常') or str(sample.get('status_code')) not in ('None', '', '0'):
            return None, 'abnormal_status'
        label = sample.get('value_label')
        if label in ('停', '关'):
            return '停', ''
        return ('开', '') if label == '开' else (None, 'unrecognized_state')
    try:
        value = float(sample['value'])
        if value in (0, 1):
            return '原始值' + str(int(value)), ''
    except (KeyError, TypeError, ValueError):
        pass
    return None, 'unrecognized_state'


def summarize_point(point, records, reader):
    counts, timestamps = Counter(), defaultdict(list)
    for record in records:
        if record['status'] != 'completed':
            counts['failed_windows'] += 1
            continue
        document = reader.data(record)
        if point['system'] == 'safety':
            samples = document['samples']
        else:
            document, shape = equipment._curve_data(document['payload'], point['node_id'])
            if shape['sample_count'] is None:
                counts['unrecognized_windows'] += 1
                continue
            samples = [sample for series in document['series'] for sample in series['samples']]
        counts['nonempty_windows' if samples else 'empty_windows'] += 1
        for index, sample in enumerate(samples):
            counts['raw_sample_count'] += 1
            try:
                at = gateway_queries.parse_time(sample['time'])
            except (KeyError, TypeError, ValueError):
                counts['invalid_time_count'] += 1
                continue
            if not gateway_queries.parse_time(record['start_time']) <= at <= gateway_queries.parse_time(record['end_time']):
                counts['out_of_window_count'] += 1
                continue
            if point['role'] == 'fan_state':
                value, reason = _state(sample, point['system'])
            else:
                try:
                    value = float(sample['value'])
                    if not math.isfinite(value):
                        raise ValueError('非有限数值')
                    reason = ''
                except (KeyError, ValueError, TypeError, OverflowError):
                    value, reason = None, 'invalid_numeric_value'
            if reason:
                counts['invalid_value_count'] += 1
            timestamps[at].append((value, reason, {'segment': record['key'], 'sample_index': index}))
    timeline = []
    for at, items in sorted(timestamps.items()):
        counts['duplicate_time_count'] += len(items) - 1
        variants = {(value, reason) for value, reason, _ in items}
        if len(variants) > 1:
            counts['conflicting_timestamp_count'] += 1
            value, reason = None, 'conflicting_samples'
        else:
            value, reason = next(iter(variants))
        timeline.append((at, value, reason, items[0][2]))
    deltas = [(after[0] - before[0]).total_seconds() for before, after in zip(timeline, timeline[1:])]
    cadence = median(deltas) if deltas else None
    gap_limit = 5 * (cadence if len(deltas) >= 3 else (60 if point['system'] == 'safety' else 3000))
    gaps = [{'after': before[0].isoformat(), 'before': after[0].isoformat(), 'seconds': delta}
        for before, after, delta in zip(timeline, timeline[1:], deltas) if gap_limit and delta > gap_limit]
    quality = {key: counts[key] for key in ('raw_sample_count', 'nonempty_windows', 'empty_windows',
        'failed_windows', 'unrecognized_windows', 'invalid_time_count', 'out_of_window_count',
        'invalid_value_count', 'duplicate_time_count', 'conflicting_timestamp_count')}
    quality.update(unique_sample_count=len(timeline), median_sample_interval_seconds=cadence,
        gap_threshold_seconds=gap_limit, largest_sample_interval_seconds=max(deltas) if deltas else None,
        gap_count=len(gaps))
    result = {**point, 'quality': quality, 'first_observed_at': timeline[0][0].isoformat() if timeline else None,
        'last_observed_at': timeline[-1][0].isoformat() if timeline else None, 'sampling_gaps': gaps}
    if point['role'] == 'flow':
        numeric = [value for _, value, reason, _ in timeline if not reason]
        result['values'] = {'numeric_sample_count': len(numeric), 'zero_count': numeric.count(0),
            'positive_count': sum(value > 0 for value in numeric), 'negative_count': sum(value < 0 for value in numeric),
            'minimum': min(numeric) if numeric else None, 'maximum': max(numeric) if numeric else None,
            'stable_window_comparison': 'not_computed'}
        return result
    intervals, changes = [], []
    previous, last_valid, interrupted = None, None, False
    for at, value, reason, source in timeline:
        gap = bool(previous and gap_limit and (at - previous[0]).total_seconds() > gap_limit)
        if (intervals and previous and (value, reason) == (previous[1], previous[2]) and not gap):
            intervals[-1].update(observed_until=at.isoformat(), last_source=source)
            intervals[-1]['sample_count'] += 1
        else:
            intervals.append({'state': value or '未知', 'observed_from': at.isoformat(), 'observed_until': at.isoformat(),
                'sample_count': 1, 'reason': reason, 'first_source': source, 'last_source': source})
        if not reason:
            if last_valid and value != last_valid[1]:
                changes.append({'from_state': last_valid[1], 'to_state': value,
                    'after': last_valid[0].isoformat(), 'by': at.isoformat(),
                    'uncertainty_seconds': (at - last_valid[0]).total_seconds(),
                    'intervening_unknown_or_gap': interrupted or gap, 'machine_switch_confirmed': False,
                    'evidence': [last_valid[3], source]})
            last_valid, interrupted = (at, value, reason, source), False
        else:
            interrupted = True
        previous = (at, value, reason, source)
    result.update(state_intervals=intervals, state_changes=changes,
        state_value_semantics='开停标签来自接口；“关”归一为“停”，原标签保留在采样证据中'
            if point['system'] == 'safety' else '0/1保留原值，开停含义待核对')
    return result


def summarize(manifest, reader):
    points = [summarize_point(point, [row for row in manifest['segments']
        if (row['system'], row['node_id']) == (point['system'], point['node_id'])], reader)
        for point in manifest['points']]
    return {'schema_version': VERSION, 'cutoff': manifest['cutoff'], 'lookback_days': manifest['lookback_days'],
        'transport': {key: manifest[key] for key in ('expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')},
        'interval_semantics': 'observed_from/until是首末采样时刻；不外推到查询边界或长缺测段。after/by表示变化发生在(after,by]内，不是精确切换时刻。',
        'gap_policy': '大于各测点中位采样间隔5倍时分段；此阈值仅用于缺测提示，不是切换合规标准。',
        'machine_mapping': '按测点和系统保留，不把一级/二级自动合成为整机，不合并跨系统状态。',
        'points': points}


def agent_view(details, details_path, *, full=False):
    """整理状态观测区间；full=True 为单次工作流展开全部已有区间和变化。"""
    result = {key: value for key, value in details.items() if key != 'points'}
    result['details_path'] = details_path
    result['state_period_semantics'] = ('state_periods仅归并同状态的有效观测。observed_parts>1表示中间含未知状态或长缺测，'
        '不是整段持续开/停；具体分段和未知区间见details_path。state_changes只给变化的采样边界。')
    result['points'] = []
    for index, point in enumerate(details['points']):
        view = {key: point[key] for key in ('system', 'node_id', 'label', 'fan_id', 'role', 'unit', 'mapping_confirmed',
            'quality', 'first_observed_at', 'last_observed_at') if key in point}
        view['details_pointer'] = '/points/' + str(index)
        if point['role'] == 'flow':
            view['values'] = point['values']
        else:
            periods = []
            for interval in point['state_intervals']:
                if interval['state'] == '未知':
                    continue
                if periods and periods[-1]['state'] == interval['state']:
                    periods[-1]['last_observed_at'] = interval['observed_until']
                    periods[-1]['sample_count'] += interval['sample_count']
                    periods[-1]['observed_parts'] += 1
                else:
                    periods.append({'state': interval['state'], 'first_observed_at': interval['observed_from'],
                        'last_observed_at': interval['observed_until'], 'sample_count': interval['sample_count'], 'observed_parts': 1})
            view['state_period_count'] = len(periods)
            view['state_periods'] = periods if full or len(periods) <= 20 else periods[:10] + periods[-10:]
            view['state_periods_truncated'] = not full and len(periods) > 20
            view['state_interval_count'] = len(point['state_intervals'])
            view['unknown_interval_count'] = sum(row['state'] == '未知' for row in point['state_intervals'])
            view['state_change_count'] = len(point['state_changes'])
            changes = point['state_changes']
            view['state_changes'] = [{key: value for key, value in change.items() if key != 'evidence'}
                for change in (changes if full or len(changes) <= 20 else changes[:10] + changes[-10:])]
            view['state_changes_truncated'] = not full and len(changes) > 20
            view['state_value_semantics'] = point['state_value_semantics']
        result['points'].append(view)
    return result