vent_agent_2.0/scripts/verify_main_fan_batch.py

#!/usr/bin/env python3
"""只读取已归档数据，输出开发核验；不调用模型、网关或推送，不代替业务判识。"""
import argparse
from collections import Counter, defaultdict
from datetime import datetime, timedelta
import hashlib
from html import escape
import json
import math
from pathlib import Path
import sqlite3
from statistics import median
from zipfile import ZipFile
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parents[1]
ZONE = ZoneInfo('Asia/Shanghai')


def read(path):
    return json.loads(path.read_text(encoding='utf-8'))


def sha(path):
    with path.open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def time(value):
    result = datetime.fromisoformat(str(value))
    return result.replace(tzinfo=ZONE) if result.tzinfo is None else result.astimezone(ZONE)


def local_path(base, virtual):
    result = (base / virtual.lstrip('/')).resolve()
    if not result.is_relative_to(base.resolve()):
        raise ValueError('归档路径超出会话目录')
    return result


def read_segment(base, segment):
    if segment.get('archive'):
        archive = segment['archive']
        item = archive['members']['data']
        with ZipFile(local_path(base, archive['path'])) as source:
            content = source.read(item['name'])
        if hashlib.sha256(content).hexdigest() != item['sha256']:
            raise ValueError('压缩数据成员指纹不一致')
        return json.loads(content)
    return read(local_path(base, segment['data_path']))


def point_statistics(segments, base, system, *, include_series=False):
    """保留原始计数；按正常标签或有限数值、窗口过滤及时间去重后计算变化。"""
    counts, values, examples = Counter(), defaultdict(set), []
    windows = []
    for segment in segments:
        if segment['status'] != 'completed':
            counts['failed_windows'] += 1
            continue
        document = read_segment(base, segment)
        if system == 'safety':
            samples = document['samples']
        else:
            rows = document.get('payload', {}).get('data')
            if not isinstance(rows, list) or any(not isinstance(row, dict)
                    or str(row.get('nodeId')) != segment['node_id']
                    or not isinstance(row.get('sbNodeCurve'), list) for row in rows):
                counts['unrecognized_windows'] += 1
                continue
            samples = [item for row in rows for item in row['sbNodeCurve']]
        counts['nonempty_windows' if samples else 'empty_windows'] += 1
        windows.append({'start_time': segment['start_time'], 'end_time': segment['end_time'],
                        'sample_count': len(samples), 'data_path': segment.get('data_path'),
                        'archive': segment.get('archive')})
        for sample in samples:
            counts['raw_sample_count'] += 1
            try:
                at = time(sample['time' if system == 'safety' else 'dataTime'])
            except (KeyError, ValueError, TypeError):
                counts['invalid_time_count'] += 1
                continue
            if not time(segment['start_time']) <= at <= time(segment['end_time']):
                counts['out_of_window_count'] += 1
                continue
            if system == 'safety':
                if sample.get('status_label') not in (None, '正常'):
                    counts['invalid_status_count'] += 1
                    continue
                value = sample.get('value_label')
                if value not in ('开', '停'):
                    counts['unrecognized_value_count'] += 1
                    continue
            else:
                try:
                    value = float(sample['realData'])
                    if not math.isfinite(value):
                        raise ValueError('非有限数值')
                except (KeyError, ValueError, TypeError, OverflowError):
                    counts['unrecognized_value_count'] += 1
                    continue
            if values[at]:
                counts['repeated_timestamp_count'] += 1
            values[at].add(value)
            if len(examples) < 3:
                examples.append(sample)
    conflicts = [at.isoformat() for at, found in values.items() if len(found) > 1]
    ordered = sorted((at, next(iter(found))) for at, found in values.items() if len(found) == 1)
    gaps = [(after[0] - before[0]).total_seconds() for before, after in zip(ordered, ordered[1:])]
    result = {key: counts[key] for key in ('raw_sample_count', 'nonempty_windows', 'empty_windows',
        'unrecognized_windows', 'failed_windows', 'invalid_time_count', 'out_of_window_count',
        'invalid_status_count', 'unrecognized_value_count', 'repeated_timestamp_count')}
    result.update(unique_usable_sample_count=len(ordered), conflicting_timestamp_count=len(conflicts),
        conflicting_timestamps=conflicts, first_time=ordered[0][0].isoformat() if ordered else None,
        last_time=ordered[-1][0].isoformat() if ordered else None,
        median_gap_seconds=median(gaps) if gaps else None, largest_gap_seconds=max(gaps) if gaps else None,
        examples=examples, windows=windows)
    if system == 'safety':
        result['state_counts'] = dict(Counter(value for _, value in ordered))
        result['transitions'] = [{'previous_sample_at': before[0].isoformat(), 'observed_at': after[0].isoformat(),
            'from_state': before[1], 'to_state': after[1], 'gap_seconds': (after[0] - before[0]).total_seconds()}
            for before, after in zip(ordered, ordered[1:]) if before[1] != after[1]]
    else:
        numeric = [value for _, value in ordered]
        result.update(zero_count=sum(value == 0 for value in numeric),
            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)
        if include_series:
            result['series'] = [{'time': at.isoformat(), 'value': value} for at, value in ordered]
    return result


def verify_mine(connection, state, mine):
    base = ROOT / 'data/workspaces' / hashlib.sha256(state['user_name'].encode()).hexdigest() / hashlib.sha256(mine['session_id'].encode()).hexdigest()
    contexts = connection.execute('SELECT id,context_json FROM main_fan_workflows WHERE user_name=? AND session_id=? ORDER BY updated_at DESC',
        (state['user_name'], mine['session_id'])).fetchall()
    manifests = [(identifier, json.loads(context)) for identifier, context in contexts
                 if json.loads(context).get('history_manifest')]
    manifests = [(identifier, context) for identifier, context in manifests
        if context['history_manifest']['cutoff'] == state['cutoff']
        and context['task_input'].get('measurements', {}).get('batch', {}).get('batch_id') == state['batch_id']]
    if not manifests:
        raise ValueError('该矿尚无归档历史数据：' + mine['dept_id'])
    workflow_id, context = manifests[0]
    manifest = context['history_manifest']
    checked, errors = [], []
    fingerprints = {virtual: expected for segment in manifest['segments'] for virtual, expected in segment['files'].items()}
    for virtual, expected in fingerprints.items():
        path = local_path(base, virtual)
        actual = sha(path) if path.is_file() else None
        entry = {'path': virtual, 'expected_sha256': expected, 'actual_sha256': actual, 'ok': actual == expected}
        checked.append(entry)
        if not entry['ok']:
            errors.append(entry)
    if errors:
        raise ValueError('归档原件指纹不一致，停止统计：' + str(errors[:3]))
    points = []
    for point in manifest['points']:
        segments = [s for s in manifest['segments'] if (s['system'], s['node_id']) == (point['system'], point['node_id'])]
        points.append({**point, **point_statistics(segments, base, point['system'], include_series=True)})
    groups = defaultdict(list)
    for point in points:
        if point['system'] == 'safety':
            for event in point['transitions']:
                groups[event['observed_at']].append({'node_id': point['node_id'], 'fan_id': point['fan_id'],
                    'label': point['label'], **event})
    candidates = []
    for observed, evidence in sorted(groups.items()):
        stopped = {event['fan_id'] for event in evidence if event['to_state'] == '停'}
        started = {event['fan_id'] for event in evidence if event['to_state'] == '开'}
        if not (stopped and started and stopped.isdisjoint(started)):
            continue
        candidates.append({'status': 'candidate', 'observed_at': observed, 'from_fans': sorted(stopped),
            'to_fans': sorted(started), 'from_stopped_at': None, 'to_started_at': None, 'interval_minutes': None,
            'state_changes': evidence, 'limitations': ['只按相同观测时刻分组；各级测点与整机及通风井对应尚未确认。',
            '实际变化落在前一采样与本次观测之间，不能把相同观测时间换算成0分钟切换。']})
    for point in points:
        series = point.pop('series', [])
        if point['role'] != 'flow':
            continue
        for candidate in candidates:
            observed = time(candidate['observed_at'])
            before = [sample for sample in series if time(sample['time']) <= observed]
            after = [sample for sample in series if time(sample['time']) > observed]
            candidate.setdefault('nearest_flow_samples', []).append({'node_id': point['node_id'],
                'unit': point['unit'], 'before': before[-1] if before else None,
                'after': after[0] if after else None,
                'note': '相邻风量采样仅作证据展示，不作为稳定窗口均值；测点归属仍待确认。'})
    archives = []
    for receipt in context.get('report_archives', []):
        archive = read(local_path(base, receipt['archive_path']))
        page_checks = []
        for page, source in archive['facts']['page_sources'].items():
            proof = archive.get('page_read_evidence', {}).get(str(page), {})
            page_checks.append(proof.get('complete') is True and sha(local_path(base, source)) == proof.get('sha256'))
        pdf = local_path(base, receipt['path'])
        archives.append({'source_sha256': receipt['sha256'], 'source_hash_matches': sha(pdf) == receipt['sha256'],
            'is_mock': receipt['is_mock'], 'binding': receipt['binding'], 'version': archive['version'],
            'page_count': archive['facts']['page_count'], 'verified_page_reads': sum(page_checks),
            'report_count': len(archive['facts']['reports']), 'reports': archive['facts']['reports'],
            'archive_path': receipt['archive_path'], 'not_target_mine_report': True})
    return {'dept_id': mine['dept_id'], 'mine_name': mine['mine_name'], 'mine_id': mine['mine_id'],
        'session_id': mine['session_id'], 'workflow_id': workflow_id, 'workspace_root': str(base),
        'transport': {key: manifest[key] for key in ('expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')},
        'original_files_checked': len(checked), 'original_file_checks': checked, 'points': points,
        'switch_candidates': candidates, 'report_archives': archives,
        'limitations': ['请求分段全部成功仅证明传输覆盖，不保证各段都有数据或采样连续。',
            '通风井身份和一级、二级测点与整机的映射待确认，切换仅为候选。',
            '风量测点只有1号风机，不能自动代表2号机或全井风量；0读数不等同于全矿停风。',
            '不同系统的开停字段不能直接替换，需核对采样、含义及设备关联。',
            '本地核验不产生生产判识结论；模型权限恢复后再由Agent复核。']}


def table(headers, rows):
    return '<table><thead><tr>' + ''.join('<th>' + escape(str(h)) + '</th>' for h in headers) + '</tr></thead><tbody>' + ''.join(
        '<tr>' + ''.join('<td>' + escape(str(value)) + '</td>' for value in row) + '</tr>' for row in rows) + '</tbody></table>'


def render(result):
    markdown = ['# 主通风机本地数据核验', '', '本页为已归档真实数据的独立开发核验。Agent补充复核因模型服务403待完成；不作为正式业务结论。旧AI初稿存在“没有风量历史”的错误，已单独标记为未通过复核。', '',
        f"查询范围：{result['start_time']} 至 {result['cutoff']}，北京时间，连续{result['lookback_days']}天，每次至多24小时。", '',
        f"只读取本批次已保存的{len(result['mines'])}个矿，未新增网关请求、未调用模型、未推送。", '']
    parts = ['<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>主通风机本地数据核验</title>',
        '<style>body{max-width:1180px;margin:40px auto;padding:0 24px;font:16px/1.75 system-ui;color:#18323c;background:#f6f8fa}h1,h2,h3{line-height:1.4}table{width:100%;border-collapse:collapse;background:white;margin:18px 0;font-size:14px}th,td{padding:11px;text-align:left;border:1px solid #dce5e9;overflow-wrap:anywhere}.notice{padding:18px;background:#fff0d2;border-left:5px solid #b47a18;border-radius:5px}a{color:#006579}code{overflow-wrap:anywhere}</style>',
        '<h1>主通风机 · 本地数据核验</h1><div class="notice">已完成归档数据核验。模型服务返回403，Agent补充复核待完成。旧AI初稿存在风量遗漏，已标记为未通过复核。本页不作生产判识结论。</div>',
        '<p>时间范围：' + escape(result['start_time'] + ' 至 ' + result['cutoff']) + '（北京时间）。只核验已选矿井，未新增网关请求、未调用模型、未推送。</p>',
        '<p><a href="本地数据核验.json">完整统计与证据JSON</a> · <a href="本地数据核验.md">Markdown报告</a> · <a href="../index.html">批次首页</a></p>']
    for mine in result['mines']:
        coverage = mine['transport']
        parts += ['<h2>' + escape(mine['mine_name']) + ' · ' + mine['dept_id'] + '</h2>',
            f"<p>{len(mine['points'])}个测点，{coverage['completed_segments']}/{coverage['expected_segments']}个查询分段成功；核对原件指纹{mine['original_files_checked']}份。通风井身份待确认。</p>",
            '<h3>安全监控状态变化</h3>']
        markdown += [f"## {mine['mine_name']}（deptId={mine['dept_id']}）", '', f"查询分段：{coverage['completed_segments']}/{coverage['expected_segments']}成功；归档请求、响应和数据共{mine['original_files_checked']}份，SHA256一致。分段成功不代表采样连续。", '', '### 安全监控状态变化', '']
        safety = [p for p in mine['points'] if p['system'] == 'safety']
        parts.append(table(['测点', '原始样本', '去重后正常样本', '变化次数', '最大采样间隔(秒)'],
            [[p['label'], p['raw_sample_count'], p['unique_usable_sample_count'], len(p['transitions']), p['largest_gap_seconds']] for p in safety]))
        for point in safety:
            markdown.append(f"- {point['label']}：原始{point['raw_sample_count']}条，去重正常{point['unique_usable_sample_count']}条，变化{len(point['transitions'])}次；nodeId={point['node_id']}。")
        parts.append('<p>以下为配对变化候选。表中时间是首次观察到新状态的时刻，实际变化发生在前后两个采样之间，整机切换间隔暂不确定。</p>')
        event_rows = []
        markdown += ['', '### 切换候选（实际切换时刻及间隔未确认）', '']
        for candidate in mine['switch_candidates']:
            label = '、'.join(candidate['from_fans']) + ' → ' + '、'.join(candidate['to_fans'])
            markdown += [f"观测时刻 {candidate['observed_at']}：{label}。", '']
            for event in candidate['state_changes']:
                event_rows.append([event['label'], event['from_state'] + '→' + event['to_state'], event['previous_sample_at'], event['observed_at'], event['gap_seconds']])
                markdown.append(f"- {event['label']} {event['from_state']}→{event['to_state']}，采样范围（{event['previous_sample_at']}, {event['observed_at']}]，间隔{event['gap_seconds']}秒。")
            for neighbor in candidate.get('nearest_flow_samples', []):
                description = '；'.join(('前一条' if side == 'before' else '后一条') + '：' + (
                    f"{neighbor[side]['time']}，{neighbor[side]['value']} {neighbor['unit']}" if neighbor[side] else '无可用样本')
                    for side in ('before', 'after'))
                markdown.append('- 相邻风量采样（非稳定窗口均值）：' + description + '。')
            markdown.append('')
        parts.append(table(['测点', '状态变化', '前一采样', '首次观察新状态', '间隔(秒)'], event_rows))
        parts.append(table(['候选观测时刻', '前一条风量采样', '后一条风量采样'], [
            [candidate['observed_at']] + [f"{row[side]['time']} / {row[side]['value']} {row['unit']}" if row[side] else '无可用样本'
                for side in ('before', 'after')]
            for candidate in mine['switch_candidates'] for row in candidate.get('nearest_flow_samples', [])]))
        parts += ['<p>相邻风量采样仅作证据展示，不代表稳定窗口均值。</p>', '<h3>重大设备风量与状态样本</h3>']
        equipment = [p for p in mine['points'] if p['system'] == 'equipment']
        parts.append(table(['测点', '有数据/空日段', '原始/去重样本', '数值范围', '正值数', '最大采样间隔(秒)'],
            [[p['label'], f"{p['nonempty_windows']}/{p['empty_windows']}", f"{p['raw_sample_count']}/{p['unique_usable_sample_count']}",
              f"{p['minimum']}～{p['maximum']}", p['positive_count'], p['largest_gap_seconds']] for p in equipment]))
        markdown += ['### 风量与重大设备状态', '']
        for point in equipment:
            markdown.append(f"- {point['label']}：有数据{point['nonempty_windows']}段、空{point['empty_windows']}段；原始{point['raw_sample_count']}条，去重可用{point['unique_usable_sample_count']}条；数值{point['minimum']}～{point['maximum']}，正值{point['positive_count']}条；nodeId={point['node_id']}。")
            if point['role'] == 'flow':
                flow_note = f"风量历史实际有{point['raw_sample_count']}条记录，去重后{point['zero_count']}条为0，正值{point['positive_count']}条。字段为data[].sbNodeCurve[].dataTime/realData，单位沿用实时列表{point['unit']}。0值属于已有数据，不能描述为没有历史。"
                if not point['positive_count']:
                    flow_note += '当前无正值样本可用于可靠的切换前后稳定风量比较。'
                parts.append('<p class="notice">' + escape(flow_note) + '</p>')
                markdown += ['', flow_note, '']
        parts.append('<h3>模拟PDF与档案</h3>')
        for archive in mine['report_archives']:
            names = '、'.join(sorted({report.get('tested_unit', '原件未提取矿名') for report in archive['reports']}))
            text = f"模拟接口返回原有测试PDF：{archive['page_count']}页、{archive['report_count']}份报告，实际全页读取证据{archive['verified_page_reads']}/{archive['page_count']}页，原件SHA256核对{'一致' if archive['source_hash_matches'] else '不一致'}。原件矿名为{names}，仅作测试夹具，不能作为本矿正式报告。"
            parts.append('<p>' + escape(text) + '</p><p>SHA256：<code>' + archive['source_sha256'] + '</code></p>')
            markdown += ['', '### 模拟PDF', '', text, '', 'SHA256：`' + archive['source_sha256'] + '`。', '']
        parts.append('<h3>核验限制与后续</h3><ul>' + ''.join('<li>' + escape(item) + '</li>' for item in mine['limitations']) + '</ul>')
        markdown += ['### 核验限制', ''] + ['- ' + item for item in mine['limitations']] + ['']
    queue = result['delivery_checks']
    text = f"本批次隔离数据库：business_jobs={queue['business_jobs']}，business_push_targets={queue['business_push_targets']}；批次publish=false。正式PDF接口、通风井和整机映射、Agent补充复核仍待完成。"
    parts += ['<p>' + escape(text) + '</p></html>']
    markdown += [text, '']
    return '\n'.join(markdown), '\n'.join(parts)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--batch-dir', type=Path, required=True)
    args = parser.parse_args()
    output = args.batch_dir.resolve()
    state = read(output / 'batch.json')
    if any(len(mine.get('shafts', [])) != 1 for mine in state['mines']):
        parser.error('本核验脚本适用于每矿一个已保存的候选系统；多井请分别按井核验，避免遗漏。')
    connection = sqlite3.connect((output / 'runtime/business.sqlite').as_uri() + '?mode=ro', uri=True)
    try:
        result = {'mode': 'independent_local_data_verification', 'agent_review': 'pending_model_access_403',
            'production_conclusion': None, 'batch_id': state['batch_id'], 'cutoff': state['cutoff'],
            'start_time': (time(state['cutoff']) - timedelta(days=state['lookback_days'])).isoformat(),
            'lookback_days': state['lookback_days'],
            'mines': [verify_mine(connection, state, mine) for mine in state['mines']],
            'delivery_checks': {table: connection.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0]
                                for table in ('business_jobs', 'business_push_targets')}}
    finally:
        connection.close()
    directory = output / 'verification'
    directory.mkdir(exist_ok=True)
    markdown, html = render(result)
    (directory / '本地数据核验.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    (directory / '本地数据核验.md').write_text(markdown, encoding='utf-8')
    (directory / '本地数据核验.html').write_text(html, encoding='utf-8')
    print(json.dumps({'status': 'verified', 'mine_count': len(result['mines']),
        'report': str(directory / '本地数据核验.html'), 'delivery_checks': result['delivery_checks']}, ensure_ascii=False))


if __name__ == '__main__':
    main()