"""按日取数，压缩证据并整理观测状态区间；整机判识仍由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