vent_agent_2.0/extensions/main_fan/discovery.py

"""宿主固定发现测点：取齐分页、校验矿井、按真实类型选开停和风量，不调用模型。"""
from collections import defaultdict
import hashlib
import json
import re
import unicodedata

import config
from extensions.main_fan import exploration, realtime
from extensions.main_fan.batch_models import HistoryPoint, MinePlan, ShaftPlan
from extensions.main_fan.batch_tools import checked_plan
from extensions.main_fan.batch_context import scope_for
from runtime.python_sandbox import workspace_file


def fan_name(row):
    """只识别名称明确写出的风机号；不从nodeId或一级/二级推断编号。"""
    numbers = '一二三四五六七八九十'
    number = rf'([0-9]+(?![0-9])|[{numbers}]+(?![{numbers}]))'
    found = set()
    for key in ('devName', 'nodePlace', 'nodeTypeName'):
        text = unicodedata.normalize('NFKC', str(row.get(key) or ''))
        # 前置“1号主扇”和后置“主通1”都可识别；编号后紧跟“级”时是电机级数。
        patterns = (rf'主(?:通风机|通|扇)(?:风机)?\s*{number}(?!\s*级)',
                    rf'{number}\s*(?:号|#)\s*(?:主扇|主要通风机|主通风机|通风机|主风机|风机|风量)')
        for pattern in patterns:
            for value in re.findall(pattern, text):
                if len(value) == 1 and value in numbers:
                    value = str(numbers.index(value) + 1)
                if value.isdigit():
                    found.add(str(int(value)) + '号风机')
    return next(iter(found)) if len(found) == 1 else ''


def shaft_name(row):
    for key in ('shaftName', 'wellName'):
        if row.get(key):
            return str(row[key]).strip()
    # 只匹配明确的井名片段；单独“通风机”或风机号均不构成通风井身份。
    names = set()
    for key in ('devFixPos', 'nodePlace', 'devName'):
        text = str(row.get(key) or '')
        names.update(re.findall(r'(?:[东西南北中](?:翼)?[一二三四五0-9]*|[一二三四五0-9]+号)(?:回风立井|回风斜井|回风井|通风井|风井)', text))
    return next(iter(names)) if len(names) == 1 else ''


def plan_from_rows(mine, safety_rows, equipment_rows):
    groups, seen = defaultdict(list), {}
    for system, rows in (('safety', safety_rows), ('equipment', equipment_rows)):
        for row in rows:
            if not isinstance(row, dict):
                raise ValueError('实时列表包含非对象记录')
            selected = (str(row.get('sensorTypeId', '')) == '1010' if system == 'safety'
                        else str(row.get('nodeTypeNo', '')) == '30101')
            if not selected:
                continue
            for field, expected in (('deptId', mine['dept_id']), ('mineId', mine['mine_id'])):
                if row.get(field) is not None and str(row[field]) != str(expected):
                    raise ValueError('实时测点的' + field + '与本矿不一致')
            if not row.get('nodeId'):
                raise ValueError('目标类型测点缺少nodeId')
            point = HistoryPoint(system=system, node_id=str(row['nodeId']),
                role='fan_state' if system == 'safety' else 'flow', fan_id=fan_name(row),
                label=str((row.get('nodePlace') if system == 'safety' else row.get('nodeTypeName')) or ''),
                unit=str(row.get('sensorUnit') or ''), mapping_confirmed=False,
                mapping_reason='按接口原名称保留编号；一级/二级分别保留，跨系统物理绑定未确认')
            name = shaft_name(row)
            key = (system, point.node_id)
            signature = (name, point.model_dump())
            if key in seen:
                if seen[key] != signature:
                    raise ValueError('分页中同一测点的类型、名称或归属冲突：' + point.node_id)
                continue
            seen[key] = signature
            groups[name].append(point)
    if not groups:
        raise ValueError('本矿未取得安全监控1010开停点或重大设备30101风量点')
    shafts, limitations = [], []
    for name, points in sorted(groups.items()):
        fans = sorted({point.fan_id for point in points if point.fan_id})
        notes = ['风机号来自测点名称，跨系统仅按原编号作时间对照，未证明物理绑定。']
        if not name:
            notes.append('接口缺少明确井名，保留矿级待确认测点集合；不将1号、2号当作两个通风井。')
        if len(fans) != 2 or any(not point.fan_id for point in points):
            notes.append('部分风机编号或主备配对不足，不能据此宣称已覆盖全部主备系统。')
        shafts.append(ShaftPlan(shaft_id='shaft-' + hashlib.sha256(name.encode()).hexdigest()[:12] if name else 'unresolved-main',
            shaft_name=name or '主通风机测点集合（通风井归属待确认）',
            identity_status='confirmed' if name else 'candidate',
            identity_reason='实时列表明确井名原文：' + name if name else '两个实时列表未提供明确通风井归属',
            fan_pair=fans if len(fans) == 2 else [], points=points, limitations=notes))
        limitations.extend(notes)
    return MinePlan(coverage='partial', shafts=shafts, limitations=list(dict.fromkeys(limitations)))


async def discover(identifier, user, session):
    scope = scope_for(user, session)
    pages = {}
    size = min(100, config.get_int('MAIN_FAN_REALTIME_MAX_PAGE_SIZE', 1000))
    for system in ('safety', 'equipment'):
        rows, page, visited = [], 1, set()
        while page:
            if page in visited:
                raise ValueError('实时列表分页循环，不能声称已取齐')
            visited.add(page)
            if system == 'safety':
                receipt = await realtime.query_sensors(identifier, scope.mine['dept_id'], page, size, ['1010'], user, session)
            else:
                receipt = await realtime.query_equipment(identifier, scope.mine['dept_id'], page, size, user, session)
            body = json.loads(workspace_file(user, session, receipt['data_path']).read_text())['payload']
            if not isinstance(body, dict) or str(body.get('code')) not in {'0', '200'} or body.get('success') is False:
                raise ValueError('实时列表未返回成功业务响应：' + system)
            if not receipt['list_recognized'] or (receipt['has_more'] and not receipt['next_page']):
                raise ValueError('实时列表结构或分页不完整：' + system)
            container = body['rows'] if 'rows' in body else body.get('data')
            rows.extend(container['rows'] if isinstance(container, dict) else container)
            page = receipt['next_page']
        pages[system] = rows
    workflow = exploration._load(identifier, user, session)
    plan = checked_plan(workflow, user, session, plan_from_rows(scope.mine, pages['safety'], pages['equipment']))
    exploration._persist(workflow, user, session, '固定代码发现本矿测点', {'mode': 'fixed_code', 'plan': plan.model_dump(mode='json')})
    return plan