vent_agent_2.0/extensions/main_fan/history.py

"""按日取数，压缩证据并整理观测状态区间；整机判识仍由Agent完成。"""
import asyncio
from datetime import timedelta
import hashlib
import json
import time

import config
from obs import get_logger
from extensions import get_service
from extensions.registry import load_extensions
from extensions.main_fan import equipment, exploration, gateway_queries, history_compaction, history_processing, timeseries
from extensions.main_fan.batch_context import scope_for
from extensions.main_fan.source_policy import source_policy, select_history_points
from runtime.python_sandbox import digest, workspace_file
from db import main_fan_store, db_conn

logger = get_logger(__name__)


def daily_windows(cutoff, days):
    end = gateway_queries.parse_time(cutoff)
    if not 1 <= days <= 40:
        raise ValueError('回查天数须为1至40天')
    return [(end - timedelta(days=index + 1), end - timedelta(days=index)) for index in range(days)]


async def query(identifier, user, session):
    started = time.monotonic()
    scope = scope_for(user, session)
    if scope.shaft is None:
        raise ValueError('请先完成矿级测点发现并由宿主派发通风井任务')
    selected_points = select_history_points(scope.shaft.points)
    if not selected_points:
        raise ValueError('开停以安全监控为准，重大设备只取风量；当前计划没有可用于本轮分析的测点')
    points = {(point.system, point.node_id): point for point in selected_points}
    ignored_points = [point.model_dump() for point in scope.shaft.points
                      if (point.system, point.node_id) not in points]
    workflow = exploration._load(identifier, user, session)
    exploration.validate_inputs(workflow, user, session)
    context = workflow['context']
    index = context.setdefault('history_index', {})
    if not index:
        # 同批次重新分析时允许宿主导入同矿、同井、同时段、同计划的原件，保持分析自主且不重拉数据。
        for row in db_conn.query('SELECT id FROM main_fan_workflows WHERE user_name=? AND session_id=? AND id<>? ORDER BY created_at DESC LIMIT 20',
                (user, session, identifier)):
            previous = main_fan_store.workflow(row['id'], user, session)['context']
            batch = previous.get('task_input', {}).get('measurements', {}).get('batch', {})
            if (batch.get('batch_id') == scope.batch_id and batch.get('cutoff') == scope.cutoff
                    and batch.get('lookback_days') == scope.lookback_days
                    and batch.get('shaft_plan') == scope.shaft.model_dump(mode='json') and previous.get('history_index')):
                index.update(previous['history_index'])
                for record in index.values():
                    context.setdefault('query_exports', {}).update(record['files'])
                context['history_reused_from_workflow'] = row['id']
                break
    gateway = get_service('shaanxi_gateway')
    spec = load_extensions().business('main_fan')
    windows = daily_windows(scope.cutoff, scope.lookback_days)
    maximum_hours = config.get_int('MAIN_FAN_SERIES_MAX_HOURS', 24)
    if maximum_hours <= 0:
        raise ValueError('历史查询窗口上限须大于0')
    # 每日窗口仍受既有接口上限约束，配置小于24小时时自动细分。
    segments = []
    for point in points.values():
        for start, end in windows:
            right = end
            while right > start:
                left = max(start, right - timedelta(hours=min(24, maximum_hours)))
                segments.append((point, left, right))
                right = left
    semaphore = asyncio.Semaphore(max(1, min(4, config.get_int('MAIN_FAN_HISTORY_CONCURRENCY', 2))))
    lock = asyncio.Lock()

    def query_key(point, start, end):
        material = [point.system, scope.mine['dept_id'], point.node_id, start.isoformat(), end.isoformat()]
        return hashlib.sha256(json.dumps(material).encode()).hexdigest()[:32]

    active_keys = {query_key(*segment) for segment in segments}
    logger.info('历史取数开始', extra={'kv': {'event': 'history.started', 'workflow_id': identifier,
        'query_count': len(segments), 'point_count': len(points), 'lookback_days': scope.lookback_days}})

    async def fetch(point, start, end):
        key = query_key(point, start, end)
        previous = index.get(key)
        if previous and previous['status'] == 'completed':
            expected = previous['files']
            if all(workspace_file(user, session, path).is_file()
                   and digest(workspace_file(user, session, path)) == sha for path, sha in expected.items()):
                return
        prefix = '/workspace/exploration/' + identifier + '/history/' + key
        record = {'key': key, 'system': point.system, 'node_id': point.node_id,
                  'role': point.role, 'fan_id': point.fan_id, 'label': point.label, 'unit': point.unit,
                  'mapping_confirmed': point.mapping_confirmed, 'start_time': start.isoformat(),
                  'end_time': end.isoformat(), 'status': 'failed', 'files': {}}
        async with semaphore:
            path = timeseries.CURVE_PATH if point.system == 'safety' else equipment.CURVE_PATH
            params, _, _ = gateway_queries.curve_parameters(point.node_id, scope.mine['dept_id'],
                start.isoformat(), end.isoformat(), node_key='nodeIds' if point.system == 'safety' else 'nodeId')
            request = {'method': 'GET', 'path': path, 'params': params}
            for attempt in range(3):
                query_started = time.monotonic()
                try:
                    body = await gateway.proxy(spec.gateway.agent_code, path, params=params)
                    if point.system == 'safety':
                        data, summary = timeseries._normalize(body, point.node_id, start, end)
                    else:
                        data, summary = equipment._curve_data(body, point.node_id)
                    for name, value in (('request.json', request), ('response.json', body), ('data.json', data)):
                        virtual = prefix + '/' + name
                        local = workspace_file(user, session, virtual)
                        exploration._write(local, value)
                        record['files'][virtual] = digest(local)
                    record.update(status='completed', summary=summary, data_path=prefix + '/data.json',
                                  python_path=(prefix + '/data.json').removeprefix('/workspace/'))
                    record.pop('error', None)
                    logger.debug('历史日段取数完成', extra={'kv': {'event': 'history.query.completed',
                        'system': point.system, 'node_id': point.node_id, 'start': start.isoformat(),
                        'end': end.isoformat(), 'attempt': attempt + 1,
                        'duration_ms': round((time.monotonic() - query_started) * 1000)}})
                    break
                except asyncio.CancelledError:
                    raise
                except (ValueError, OSError, RuntimeError) as exc:
                    logger.warning('历史日段取数失败', extra={'kv': {'event': 'history.query.failed',
                        'system': point.system, 'node_id': point.node_id, 'start': start.isoformat(),
                        'end': end.isoformat(), 'attempt': attempt + 1, 'error_type': type(exc).__name__}})
                    record['error'] = type(exc).__name__ + ': ' + str(exc)[:300]
                    if attempt == 2 or getattr(exc, 'retryable', True) is False:
                        break
                    await asyncio.sleep(.25 * (attempt + 1))
        async with lock:
            index[key] = record
            # 不为每个日窗口增加 Agent 操作节点；独立原件、指纹和进度仍逐段持久化。
            context.setdefault('query_exports', {}).update(record['files'])
            main_fan_store.update_workflow(identifier, context=context)
            finished = sum(key in index for key in active_keys)
            if scope.progress and (finished % 10 == 0 or finished == len(segments)):
                scope.progress({'stage': 'history', 'dept_id': scope.mine['dept_id'],
                    'shaft_id': scope.shaft.shaft_id, 'finished': finished, 'total': len(segments)})

    await asyncio.gather(*(fetch(*segment) for segment in segments))
    active = [index[key] for key in active_keys]
    logger.info('历史取数结束', extra={'kv': {'event': 'history.finished', 'workflow_id': identifier,
        'query_count': len(active), 'failed_count': sum(row['status'] != 'completed' for row in active),
        'duration_ms': round((time.monotonic() - started) * 1000)}})
    records = list(index.values())
    resolve = lambda path: workspace_file(user, session, path)
    archive_path = '/workspace/exploration/' + identifier + '/history-evidence.zip'
    records, archive_files, cleanup, storage = await asyncio.to_thread(
        history_compaction.compact_evidence, records, archive_path, resolve)
    for old in index.values():
        for path in old['files']:
            context.setdefault('query_exports', {}).pop(path, None)
    context['query_exports'].update(archive_files)
    context['history_index'] = {row['key']: row for row in records}
    # 先保存指向已校验ZIP的索引，再清理本工作流暂存JSON；旧批次原件不删除。
    main_fan_store.update_workflow(identifier, context=context)
    staging = '/workspace/exploration/' + identifier + '/history/'
    directories = set()
    for path in cleanup:
        if path.startswith(staging):
            local = resolve(path)
            local.unlink(missing_ok=True)
            directories.add(local.parent)
    for directory in directories:
        try:
            directory.rmdir()
        except OSError:
            pass
    # 旧档案原件和缓存索引保留，只让当前数据源计划的日段参与覆盖统计及Agent摘要。
    records = [row for row in records if row['key'] in active_keys]
    failed = [row for row in records if row['status'] != 'completed']
    manifest = {'cutoff': scope.cutoff, 'lookback_days': scope.lookback_days,
        'source_policy': source_policy(),
        'ignored_plan_points': ignored_points,
        'reused_from_workflow': context.get('history_reused_from_workflow', ''),
        'storage': storage,
        'point_count': len(points), 'expected_segments': len(segments),
        'completed_segments': len(records) - len(failed), 'failed_segments': len(failed),
        'coverage_complete': bool(points) and not failed and len(records) == len(segments),
        'points': [point.model_dump() for point in points.values()], 'segments': records}
    def derive():
        shapes, equipment_counts = [], {}
        with history_compaction.RecordReader(resolve) as reader:
            for point in points.values():
                if point.system != 'equipment':
                    continue
                recognized_count, unrecognized, example = 0, 0, None
                for record in records:
                    if record['system'] != point.system or record['node_id'] != point.node_id or record['status'] != 'completed':
                        continue
                    payload = reader.data(record)['payload']
                    _, summary = equipment._curve_data(payload, point.node_id)
                    if summary['sample_count'] is None:
                        unrecognized += 1
                    else:
                        recognized_count += summary['sample_count']
                    if example is None or (not example['sample_count'] and summary.get('sample_count')):
                        example = {'system': point.system, 'node_id': point.node_id, 'role': point.role,
                            'sample_count': summary['sample_count'], 'archive_path': record['archive']['path'],
                            'data_member': record['archive']['members']['data']['name'],
                            'structure': summary['data_structure']}
                equipment_counts[point.node_id] = {'role': point.role, 'recognized_samples': recognized_count,
                    'unrecognized_windows': unrecognized}
                if example:
                    shapes.append(example)
            return shapes, equipment_counts, history_compaction.summarize(manifest, reader)
    shapes, equipment_counts, details = await asyncio.to_thread(derive)
    details_path = '/workspace/exploration/' + identifier + '/history-state-details.json'
    agent_path = '/workspace/exploration/' + identifier + '/history-agent.json'
    details['storage'] = storage
    details['source_policy'] = source_policy()
    details['mine'] = {key: scope.mine[key] for key in ('mine_id', 'dept_id', 'mine_name')}
    details['shaft'] = {'shaft_id': scope.shaft.shaft_id, 'shaft_name': scope.shaft.shaft_name, 'identity_status': scope.shaft.identity_status}
    with history_compaction.RecordReader(resolve) as reader:
        compact = await asyncio.to_thread(history_processing.prepare, details, manifest, reader, context['policy'], details_path)
    exploration._write(resolve(details_path), details)
    context['query_exports'][details_path] = digest(resolve(details_path))
    compact = history_processing.agent_input(compact)
    exploration._write(resolve(agent_path), compact)
    context['query_exports'][agent_path] = digest(resolve(agent_path))
    manifest.update(equipment_sample_counts=equipment_counts, data_shapes=shapes, agent_data_path=agent_path)
    virtual = '/workspace/exploration/' + identifier + '/history-manifest.json'
    local = workspace_file(user, session, virtual)
    exploration._write(local, manifest)
    context.setdefault('query_exports', {})[virtual] = digest(local)
    context['history_manifest'] = manifest
    receipt = {key: manifest[key] for key in ('cutoff', 'lookback_days', 'point_count',
               'expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')}
    receipt.update(ok=True, source_policy=source_policy(), manifest_path=virtual, python_path=virtual.removeprefix('/workspace/'),
        agent_data_path=agent_path, agent_python_path=agent_path.removeprefix('/workspace/'),
        evidence_archive_path=storage.get('archive_path'), equipment_sample_counts=equipment_counts,
        state_interval_counts={row['node_id']: len(row['state_intervals']) for row in details['points'] if row['role'] == 'fan_state'},
        note='先读精简的agent_data_path：states为开停区间，flows为风量概况，switch_candidates为切换候选及前后风量。开停区间为同状态观测合并；time_range不是切换耗时；无效风量差为null。质量和计算明细仅在需要时读取details_path。')
    context.setdefault('history_queries', []).append(receipt)
    exploration._persist(workflow, user, session, '按日回查开停与风量', receipt)
    logger.info('切换候选及前后参考风量已写入Agent输入', extra={'kv': {
        'event': 'history.prepared', 'workflow_id': identifier,
        'preprocessing_version': history_processing.VERSION,
        'switch_candidate_count': details['preprocessing']['switch_candidate_count'],
        'agent_data_path': agent_path}})
    return receipt