guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/service/registry.py

# -*- coding: utf-8 -*-
r"""端点登记表：**唯一需要维护的清单** —— 端点名 → (算法模块, 函数, 允许的查询参数)。

判据：只登记"能无参或仅带可选参数调用、返回可 JSON 化数据、不改盘上产物"的算法函数。
带写盘副作用的函数一律固定成只读参数（如 `curves.build_store(write=False)`）。
"""
from __future__ import annotations

import importlib
from typing import Any, Callable

# 端点 → dict(module=算法模块点号名, fn=函数名, args=允许的查询参数 → (类型, 转发名, 固定值/None))
#   · 类型: 'str' | 'int' | 'bool'
#   · 固定值不为 None 时，该参数**不接受外部传入**，恒用固定值（防止服务被用来写盘）
ENDPOINTS: dict[str, dict[str, Any]] = {
    'systems': dict(module='app_algorithmModel.app_algorithmModel_guanlan.taxonomy', fn='__systems__',
                    doc='判级维度字典（七系统 SYSTEMS + 温度通道→系统 TEMP2SYS）', args={}),
    'matrix': dict(module='app_algorithmModel.app_algorithmModel_guanlan.taxonomy', fn='system_matrix',
                   doc='七系统判级矩阵（与 /detail/api/fleet 的 sysmx 同源）',
                   args={}),          # span 见文件头实逮说明：不是窗口字符串，P10 不传
    'reconcile': dict(module='app_algorithmModel.app_algorithmModel_guanlan.taxonomy', fn='reconcile',
                      doc='判级与台账对账', args={}),
    'reliability': dict(module='app_algorithmModel.app_algorithmModel_guanlan.perf.reliability', fn='overview',
                        doc='可靠性总览（含可用率与故障统计）',
                        args={'since': ('str', 'since', None)}),   # span 是跨度对象，P10 不传
    'problems': dict(module='app_algorithmModel.app_algorithmModel_guanlan.perf.reliability',
                     fn='current_problems', doc='当前问题清单', args={}),
    # ★2026-09-29 实逮：`availability.build()` 返回 None（它只写盘、不返值）⇒ 不是接口，已撤。
    #   可用率对外只经 `availability_summary`（=`summary()`，返回结构化摘要）。
    'availability_summary': dict(module='app_algorithmModel.app_algorithmModel_guanlan.perf.availability',
                                 fn='summary', doc='可用率摘要',
                                 args={'month_from': ('str', 'month_from', None)}),
    'fusion': dict(module='app_algorithmModel.app_algorithmModel_guanlan.subsys.fusion', fn='fusion_table',
                   doc='融合面四源表', args={'all_turbines': ('bool', 'all_turbines', None)}),
    'handoff': dict(module='app_algorithmModel.app_algorithmModel_guanlan.subsys.fusion', fn='load_handoff',
                    doc='振动融合交接件（handoff_vibration_v2）', args={}),
    'pitch': dict(module='app_algorithmModel.app_algorithmModel_guanlan.subsys.pitch', fn='registry',
                  doc='变桨面（日粒度 + 零位三口径）', args={}),
    'curves': dict(module='app_algorithmModel.app_algorithmModel_guanlan.perf.curves', fn='build_store',
                   doc='功率/转速曲线仓（**只读**：write 固定 False）',
                   args={'write': ('bool', 'write', False)}),
    # ★2026-09-30 用户令（P12）：数据类端点的计算归算法服务（目标栈 = Python+FastAPI），
    #    Java 业务后端经 /algorithm 取数后按契约组装（`/api/maint_survey` 是首个）。
    'ont_chain': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.ontology_views',
                        fn='ont_chain', doc='本体链（fault/prev/tree/plan/doc 五路，详情层同形）',
                        args={'kind': ('str', 'kind', 'fault'), 'code': ('str', 'code', None),
                              'system': ('str', 'system', None), 'top': ('int', 'top', 10),
                              'keyword': ('str', 'keyword', None), 'dkind': ('str', 'dkind', None)}),
    'ont_list': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.ontology_views',
                       fn='ont_list_rows', doc='本体对象清单（详情层 /api/ont_list 同形）',
                       args={'type': ('str', 'type', None), 't': ('str', 't', None),
                             'limit': ('int', 'limit', 60)}),
    'ont_obj': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.ontology_views',
                      fn='ont_obj_bundle', doc='本体对象 + 链接（详情层 /api/ont_obj 同形）',
                      args={'id': ('str', 'id', None)}),
    'vibcms': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.fleet_views',
                    fn='vibcms', doc='windcms 评估报告转录（详情层 /api/vibcms 同形）', args={}),
    'reload': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.fleet_views',
                    fn='reload', doc='产物重载（返回旧/新指纹与文件数）',
                    args={'why': ('str', 'why', None)}),
    'curves_view': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.fleet_views',
                        fn='curves', doc='曲线视图（详情层 /api/curves 同形；注意与算法原生 curves 区分）',
                        args={'win': ('str', 'win', None)}),
    'fleet': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.fleet_views',
                   fn='fleet', doc='整场取数（10 个页签的取数源，详情层同形）',
                   args={'win': ('str', 'win', None)}),
    'maint_std': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.maint_views',
                    fn='maint_std', doc='运维标准四判据 + 成熟度评分（详情层同形）', args={}),
    'maint_framework': dict(module='app_algorithmModel.app_algorithmModel_guanlan.service.maint_views',
                          fn='maint_framework', doc='运维管理体系三维骨架（详情层同形）',
                          args={'lang': ('str', 'lang', None)}),
    'maint_survey': dict(module='app_ontology.app_ontology_guanlan.ontology.maintenance',
                         fn='survey', doc='维护面盘点（数据/机理两层的数据源清单与覆盖）', args={}),
    'curves_check': dict(module='app_algorithmModel.app_algorithmModel_guanlan.perf.curves', fn='physics_check',
                         doc='曲线物理性检查', args={}),
}

_cache: dict[str, Any] = {}


def _mod(name: str):
    return importlib.import_module(name)


def call(name: str, params: dict[str, Any]) -> Any:
    """按登记表取数（params 已由 HTTP 层按类型解析好）。"""
    spec = ENDPOINTS.get(name)
    if not spec:
        raise KeyError(f'未登记端点 {name!r}；可用: {", ".join(sorted(ENDPOINTS))}')
    m = _mod(spec['module'])
    if spec['fn'] == '__systems__':
        return {'SYSTEMS': list(m.SYSTEMS), 'TEMP2SYS': dict(m.TEMP2SYS)}
    fn: Callable = getattr(m, spec['fn'])
    kwargs: dict[str, Any] = {}
    for qname, (_t, fname, fixed) in (spec.get('args') or {}).items():
        val = fixed if fixed is not None else params.get(fname)
        if val is not None:
            kwargs[fname] = val
    return fn(**kwargs)


def spec_list() -> list[dict[str, Any]]:
    return [dict(name=k, module=v['module'], fn=v['fn'], doc=v['doc'],
                 params=sorted((v.get('args') or {}).keys())) for k, v in sorted(ENDPOINTS.items())]


__all__ = ['ENDPOINTS', 'call', 'spec_list']