scense/app/_archive_disabled_20261028_151505/ns_wiring.py

# -*- coding: utf-8 -*-
"""宿主侧命名空间注入接线（T4 dspy 命名空间符号注入族）。

【为什么在宿主侧还有一层】ahserver 的 .dspy exec 命名空间由框架在每次请求构造，
模块 init() 的 `env.xxx = xxx` 只有在**宿主已调用 load_{模块}()** 时才生效；基础模块
（rbac / appbase / bricks）是 referenced_modules（只读，禁改源码），其端点若依赖
`env.<handler>` 而该符号未挂/为 None，表现就是 `TypeError: 'NoneType' object is not callable`
（R42 集群⑤）。宿主侧做两件确定性的事：

  1) `install_module_shim(env)` —— 把业务模块的统一 dispatch 入口挂到 ServerEnv
     （scense.ns_shim.install；scense_demo 复用同一入口，不重复实现）。
  2) `backfill_base_handlers(env)` —— 对只读基础模块，用包内真实存在且可调用的函数
     补齐 `env.<name>`（**只 setattr 运行时命名空间，不改基础模块源码**）；
     包内也不存在的符号不伪造，交由端点侧 ns-shim 给 `symbol-missing:<op>` 结构化 503。

接线点：应用入口 init() 在全部 load_*() 之后调用本模块 `wire_ns_shim(env)`，并把
`env.NS_SHIM_WIRING` 落一条自证字符串，供部署态核验。
"""
import importlib

NS_WIRING_VERSION = "t4-host-wiring-1"

# 只读基础模块的 handler 候选（包内存在才补；不存在不伪造）
BASE_HANDLER_CANDIDATES = (
    ("rbac", ("find_unauth_files", "list_path_roles", "list_roles", "list_permissions")),
    ("appbase", ("get_code", "get_appcodes_kv", "get_appcodes", "get_svgicon")),
    ("bricks", ("list_tables", "tableinfo", "channels")),
)


def _get(env, name):
    try:
        return getattr(env, name, None)
    except Exception:
        return None


def _set(env, name, value):
    try:
        setattr(env, name, value)
        return True
    except Exception:
        pass
    try:
        env[name] = value
        return True
    except Exception:
        return False


def install_module_shim(env):
    """业务模块统一 dispatch 入口（scense 域）。返回已挂载入口名清单。"""
    installed = []
    if _get(env, "dspy_ns_call") is not None:
        installed.append("dspy_ns_call(pre-existing)")
    try:
        from scense.ns_shim import install as _install_scense
        _install_scense(env)
        installed.append("scense.ns_shim.install")
    except Exception as exc:  # 模块未装/未加载都不是致命错误，但要留痕
        installed.append("scense.ns_shim.install:FAIL:%s" % exc)
    return installed


def backfill_base_handlers(env):
    """只读基础模块 handler 补齐。返回 {module: [补齐符号]}。"""
    report = {}
    for modpath, names in BASE_HANDLER_CANDIDATES:
        try:
            mod = importlib.import_module(modpath)
        except Exception:
            report[modpath] = []
            continue
        fixed = []
        for name in names:
            fn = getattr(mod, name, None)
            if fn is None or not callable(fn):
                continue
            if _get(env, name) is None:
                if _set(env, name, fn):
                    fixed.append(name)
        report[modpath] = fixed
    return report


def matrix(env):
    """注入生效矩阵（部署态可经 /scense/api 侧或脚本读取）。"""
    rows = []
    try:
        from scense.ns_shim import ns_shim_matrix
        rows = ns_shim_matrix(env).get("rows", [])
    except Exception:
        rows = []
    base = {}
    for modpath, names in BASE_HANDLER_CANDIDATES:
        base[modpath] = {
            name: {
                "env_has": _get(env, name) is not None,
                "module_has": _module_has(modpath, name),
            } for name in names
        }
    return {"version": NS_WIRING_VERSION, "ops": rows, "base_handlers": base}


def _module_has(modpath, name):
    try:
        mod = importlib.import_module(modpath)
    except Exception:
        return False
    return callable(getattr(mod, name, None))


def wire_ns_shim(env):
    """入口接线：install + backfill + 自证。永不抛异常（启动期不能让接线阻断服务）。"""
    out = {"version": NS_WIRING_VERSION, "installed": [], "backfilled": {}, "error": None}
    try:
        out["installed"] = install_module_shim(env)
    except Exception as exc:
        out["error"] = "install:" + str(exc)
    try:
        out["backfilled"] = backfill_base_handlers(env)
    except Exception as exc:
        out["error"] = (out["error"] or "") + "|backfill:" + str(exc)
    summary = "; ".join(
        ["installed=" + ",".join(out["installed"])] +
        ["%s=%s" % (k, ",".join(v) or "-") for k, v in out["backfilled"].items()]
    )
    try:
        env.NS_SHIM_WIRING = NS_WIRING_VERSION + " " + summary
        env.ns_shim_matrix = lambda: matrix(env)  # noqa: E731  部署态自检入口
    except Exception:
        pass
    return out