vent_agent_2.0/extensions/main_fan/realtime.py

"""安全监控及重大设备实时列表：保留原始字段并显式返回分页信息。"""
import asyncio
import json
import re

from pydantic import BaseModel, ConfigDict, Field

import config
from extensions.main_fan import gateway_queries

SENSOR_PATH = '/aqjk/node/getNodeRealtimeDetails'
EQUIPMENT_PATH = '/zdsb/sbnode/selectSbNodeList'


class RealtimePage(BaseModel):
    model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
    dept_id: str = Field(min_length=1, max_length=64, pattern=r'^[A-Za-z0-9_-]+$')
    page_num: int = Field(ge=1)
    page_size: int = Field(ge=1)


def _page(dept_id, page_num, page_size):
    page = RealtimePage(dept_id=dept_id, page_num=page_num, page_size=page_size)
    maximum = config.get_int('MAIN_FAN_REALTIME_MAX_PAGE_SIZE', 1000)
    if maximum <= 0 or page.page_size > maximum:
        raise ValueError(f'实时列表每页最多 {maximum} 条，请减小 page_size 并分页查询')
    return page


def _list_data(body, page):
    if not isinstance(body, (dict, list)):
        raise ValueError('实时列表未返回 JSON 对象或数组，请核对上游接口')
    container = body.get('data') if isinstance(body, dict) and 'data' in body else body
    rows, pointer = None, None
    if isinstance(body, dict) and 'rows' in body:
        rows, pointer, container = body['rows'], '/payload/rows', body
    elif isinstance(container, dict) and 'rows' in container:
        rows, pointer = container['rows'], '/payload/data/rows'
    elif isinstance(container, list) and all(isinstance(row, dict) for row in container):
        rows = container
        pointer = '/payload/data' if isinstance(body, dict) else '/payload'
    if pointer is not None and (not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows)):
        raise ValueError('实时列表 rows 必须是对象数组，不能把无效列表当成0条记录')
    total = container.get('total') if isinstance(container, dict) else None
    if total is None and isinstance(body, dict):
        total = body.get('total')
    total = int(str(total)) if re.fullmatch(r'\d+', str(total)) else None
    count = len(rows) if rows is not None else None
    has_more = ((page.page_num - 1) * page.page_size + count < total
                if total is not None and count is not None else None)
    next_page = (page.page_num + 1 if count and (has_more is True or
                 (has_more is None and count >= page.page_size)) else None)
    preview = rows[:3] if rows is not None else []
    preview_omitted = len(json.dumps(preview, ensure_ascii=False, allow_nan=False)) > 16000
    summary = {'page_num': page.page_num, 'page_size': page.page_size, 'row_count': count,
               'total': total, 'has_more': has_more, 'next_page': next_page,
               'rows_pointer': pointer, 'list_recognized': pointer is not None,
               'preview': [] if preview_omitted else preview, 'preview_omitted': preview_omitted}
    if rows is None:
        summary['next_action'] = '完整响应已保存，列表结构尚未识别；请读取 data_path 的 payload 核对字段，不能按0条记录处理。'
    elif count == 0 and has_more:
        summary['next_action'] = '本页为空但 total 表明仍有记录，请核对页码与上游分页状态，不能视为完整空列表。'
    return {'payload': body}, summary


async def _query(identifier, page, params, user, session, *, equipment):
    path = EQUIPMENT_PATH if equipment else SENSOR_PATH
    workflow, request, body = await gateway_queries.fetch(identifier, user, session, path, params)
    data, summary = await asyncio.to_thread(_list_data, body, page)
    if equipment:
        summary['dev_of_system'] = '30'
    return await asyncio.to_thread(gateway_queries.archive, workflow, user, session, request, body, data, summary,
        kind='equipment_realtime' if equipment else 'sensor_realtime',
        collection='equipment_queries' if equipment else 'sensor_queries',
        operation='查询重大设备实时列表' if equipment else '查询实时传感器列表', filename='realtime.json')


async def query_equipment(identifier, dept_id, page_num, page_size, user, session):
    page = _page(dept_id, page_num, page_size)
    params = {'pageNum': str(page.page_num), 'pageSize': str(page.page_size),
              'deptId': page.dept_id, 'devOfSystem': '30'}
    return await _query(identifier, page, params, user, session, equipment=True)


async def query_sensors(identifier, dept_id, page_num, page_size, sensor_type_list, user, session):
    page = _page(dept_id, page_num, page_size)
    if sensor_type_list is not None:
        if not isinstance(sensor_type_list, list) or len(sensor_type_list) > 128:
            raise ValueError('sensor_type_list 须为最多128项的编码数组')
        if any(not isinstance(value, str) or len(value) > 128 or
               not re.fullmatch(r'[A-Za-z0-9_-]+(?:,[A-Za-z0-9_-]+)*', value) for value in sensor_type_list):
            raise ValueError('传感器类型须为字符串编码，可保留接口示例中的逗号组合，不能猜类型编码')
    # 保留网关约定的键值对数组与索引参数；不擅自拆分上游可能有分组含义的逗号项。
    params = [['pageSize', page.page_size], ['pageNum', page.page_num], ['deptId', page.dept_id]]
    params.extend([f'sensorTypeList[{index}]', value] for index, value in enumerate(sensor_type_list or []))
    return await _query(identifier, page, params, user, session, equipment=False)