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)}),
    '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']