"""按测点及时间窗口取原始曲线，保存为本轮可引用的可信输入。"""
import asyncio

import config
from extensions.main_fan import gateway_queries

CURVE_PATH = '/aqjk/alarmcurve/getCurveByNodeId'


def _normalize(body, node_id, start, end):
    if not isinstance(body, dict) or body.get('success') is False or str(body.get('code')) not in {'0', '200'}:
        raise ValueError('曲线业务响应未成功，请检查网关及上游返回')
    data = body.get('data')
    if data == []:
        rows, metadata = [], []
    elif (isinstance(data, list) and len(data) == 2
          and isinstance(data[0], list) and isinstance(data[1], list)):
        rows, metadata = data
    else:
        raise ValueError('曲线响应应包含 [采样记录数组, 测点信息数组]，不能将未知结构当成空数据')
    maximum = config.get_int('MAIN_FAN_SERIES_MAX_SAMPLES', 200000)
    if maximum <= 0 or len(rows) > maximum:
        raise ValueError(f'单次返回最多接受 {maximum} 条采样，请缩小查询时间范围')
    for item in [*rows, *metadata]:
        if not isinstance(item, dict) or item.get('NodeID') != node_id:
            raise ValueError('曲线返回的测点编号与请求不一致，不能混用其他测点')
    samples, times = [], []
    for row in rows:
        if 'RealData' not in row or 'DataTime' not in row:
            raise ValueError('曲线采样缺少 RealData 或 DataTime，请核对实际接口字段')
        observed = gateway_queries.parse_time(row['DataTime'])
        times.append(observed)
        samples.append({'time': observed.isoformat(), 'value': row['RealData'],
                        'value_label': row.get('KgDesc'), 'status_code': row.get('StateID'),
                        'status_label': row.get('stateName')})
    # 原顺序、重复、缺值和窗口外记录全部保留；只提供质量计数，不替 Agent 选择分析样本。
    quality = {'duplicate_time_count': len(times) - len(set(times)),
               'out_of_order_count': sum(right < left for left, right in zip(times, times[1:])),
               'out_of_range_count': sum(not start <= value <= end for value in times),
               'null_value_count': sum(row['value'] is None for row in samples)}
    return {'metadata': metadata, 'samples': samples}, {
        'sample_count': len(samples), 'first_time': min(times).isoformat() if times else None,
        'last_time': max(times).isoformat() if times else None, 'quality': quality,
        'preview': samples[:3] + (samples[-1:] if len(samples) > 3 else [])}


async def query(identifier, node_id, dept_id, start_time, end_time, user, session):
    params, start, end = gateway_queries.curve_parameters(node_id, dept_id, start_time, end_time)
    workflow, request, body = await gateway_queries.fetch(identifier, user, session, CURVE_PATH, params)
    data, summary = await asyncio.to_thread(_normalize, body, params['nodeIds'], start, end)
    return await asyncio.to_thread(gateway_queries.archive, workflow, user, session, request, body,
        {'timezone': 'Asia/Shanghai', **data}, summary, kind='sensor_timeseries', collection='timeseries_queries',
        operation='查询测点历史时序', filename='samples.json')