vent_agent_2.0/scripts/compact_main_fan_history.py

#!/usr/bin/env python3
"""用已有单矿核验数据生成与生产工具同格式的Agent区间输入示例，不调用网关或模型。"""
import argparse
import hashlib
import json
from pathlib import Path
import shutil
import sqlite3
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from extensions.main_fan import exploration, history_compaction


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 = json.loads((output / 'batch.json').read_text())
    verification = json.loads((output / 'verification/本地数据核验.json').read_text())
    if len(verification['mines']) != 1:
        parser.error('此示例命令仅回放已有单矿核验批次')
    mine = verification['mines'][0]
    chosen = next(item for item in state['mines'] if item['dept_id'] == mine['dept_id'])
    if len(chosen['shafts']) != 1:
        parser.error('此示例命令只回放当前已核验的单候选系统')
    shaft = chosen['shafts'][0]
    base = ROOT / 'data/workspaces' / hashlib.sha256(state['user_name'].encode()).hexdigest() / hashlib.sha256(mine['session_id'].encode()).hexdigest()
    connection = sqlite3.connect((output / 'runtime/business.sqlite').as_uri() + '?mode=ro', uri=True)
    try:
        context = json.loads(connection.execute('SELECT context_json FROM main_fan_workflows WHERE id=?',
            (mine['workflow_id'],)).fetchone()[0])
    finally:
        connection.close()
    original = context['history_manifest']
    prefix = '/workspace/exploration/' + mine['workflow_id'] + '/compacted-history-preview'
    resolve = lambda virtual: base / virtual.lstrip('/')
    records, _, _, storage = history_compaction.compact_evidence(original['segments'], prefix + '/history-evidence.zip', resolve)
    manifest = {**original, 'segments': records, 'storage': storage}
    with history_compaction.RecordReader(resolve) as reader:
        details = history_compaction.summarize(manifest, reader)
    details.update(storage=storage, mine={key: mine[key] for key in ('dept_id', 'mine_id', 'mine_name')},
        shaft={key: shaft[key] for key in ('shaft_id', 'shaft_name', 'identity_status')})
    view = history_compaction.agent_view(details, prefix + '/history-state-details.json')
    for name, value in (('history-agent.json', view), ('history-state-details.json', details), ('history-manifest.json', manifest)):
        exploration._write(resolve(prefix + '/' + name), value)
    destination = output / 'verification/agent-history'
    destination.mkdir(exist_ok=True)
    for name in ('history-agent.json', 'history-state-details.json', 'history-manifest.json', 'history-evidence.zip'):
        shutil.copyfile(resolve(prefix + '/' + name), destination / name)
    lines = ['# 给Agent的整理后数据示例', '', f"来源：{mine['mine_name']}（deptId={mine['dept_id']}），复用已归档40天数据；本次未调用模型或网关。", '',
        'Agent先读history-agent.json；精细连续区间、异常和缺口见history-state-details.json；逐日完整证据在一个ZIP中。', '',
        '下表为同状态有效观测的大段摘要。有效观测段数大于1时，中间含未知或长缺测，不能把起止之间全部认定为持续开/停。实际变化只定位到前后两次采样之间。', '']
    for point in view['points']:
        lines += ['## ' + point['label'], '', 'nodeId：`' + point['node_id'] + '`。', '']
        if point['role'] == 'fan_state':
            lines += ['| 状态 | 首次有效观测 | 最后有效观测 | 有效观测分段数 |', '|---|---|---|---|']
            lines += [f"| {row['state']} | {row['first_observed_at']} | {row['last_observed_at']} | {row['observed_parts']} |" for row in point['state_periods']]
            lines += ['', f"精细分段{point['state_interval_count']}段，其中未知{point['unknown_interval_count']}段；状态变化{point['state_change_count']}次。", '']
        else:
            values = point['values']
            lines += [f"原始{point['quality']['raw_sample_count']}条，数值有效{values['numeric_sample_count']}条，0值{values['zero_count']}条；范围{values['minimum']}～{values['maximum']} {point['unit']}。未计算稳定窗口对比。", '']
    lines += ['## 文件体积', '', f"原始{storage['original_file_count']}份JSON共{storage['original_bytes']}字节，压缩证据包{storage['compressed_bytes']}字节；Agent入口JSON为{(destination / 'history-agent.json').stat().st_size}字节。", '',
        '此前联调原件与数据库未改写；后续取数工具将自动压缩归档并清理本轮逐日暂存JSON。', '']
    (destination / '整理后数据示例.md').write_text('\n'.join(lines), encoding='utf-8')
    print(json.dumps({'agent_input': str(destination / 'history-agent.json'), 'example': str(destination / '整理后数据示例.md'),
        'agent_input_bytes': (destination / 'history-agent.json').stat().st_size, 'storage': storage,
        'state_period_counts': {point['label']: point.get('state_period_count') for point in view['points']}}, ensure_ascii=False))


if __name__ == '__main__':
    main()