# [fix-iter3 strict-auth v2] fail-closed 门禁（Bug 1BoEurcDgf4HHRqBMmJt8 / R-01-neg，P0）——勿删
# 本段取代「globals().get("get_user") 判空」的旧判据：ahserver 的 dspy 预加载全局经
# builtins 解析、不在 exec-globals 里，globals().get 恒为 None → 判据失真 → 匿名 200。
# 口径：裸名取值 → 双重判空（get_user + get_userinfo，且必须取到有效用户标识）
#       → 未登录返回真 HTTP 401（Response 解析不到时 set_status(401)+JSON，绝不返回业务数据）。
_auth_uid = ""


def _auth_uid_of(_v):
    """从 get_user()/get_userinfo() 的返回值里提取有效用户标识；取不到返回 ''。"""
    if not _v:
        return ""
    if isinstance(_v, str):
        return _v
    if isinstance(_v, dict):
        for _k in ("id", "userid", "user_id", "user", "username", "name", "uid", "account"):
            try:
                _x = _v.get(_k)
            except Exception:
                _x = None
            if _x:
                return str(_x)
        return ""
    for _k in ("id", "userid", "user_id", "user", "username", "name", "uid", "account"):
        try:
            _x = getattr(_v, _k, None)
        except Exception:
            _x = None
        if _x:
            return str(_x)
    return ""


def _auth_call(_name):
    """裸名优先取预加载全局（builtins 解析），退 globals 映射。取不到返回 None。"""
    _f = None
    try:
        _f = globals().get(_name)
    except Exception:
        _f = None
    if not callable(_f):
        try:
            _f = _auth_bare(_name)
        except Exception:
            _f = None
    return _f if callable(_f) else None


def _auth_bare(_name):
    """裸名解析（ahserver 预加载全局在 builtins 中）；未注入时抛 NameError。"""
    if _name == "get_user":
        return get_user
    if _name == "get_userinfo":
        return get_userinfo
    if _name == "get_username":
        return get_username
    raise NameError(_name)


async def _auth_resolve_uid():
    for _n in ("get_userinfo", "get_user", "get_username"):
        _f = _auth_call(_n)
        if _f is None:
            continue
        try:
            _r = _f()
            if hasattr(_r, "__await__"):
                _r = await _r
        except Exception:
            _r = None
        _u = _auth_uid_of(_r)
        if _u:
            return _u
    return ""


try:
    _auth_uid = await _auth_resolve_uid()
except Exception:
    _auth_uid = ""

if not _auth_uid:
    _auth_deny = {
        "status": 401,
        "success": False,
        "ok": False,
        "need_login": True,
        "message": "unauthorized: login required",
        "reason": "login required",
        "rows": [],
        "data": [],
        "total": 0,
    }
    _auth_resp_cls = None
    try:
        _auth_proc = globals().get("processor")
        if _auth_proc is None:
            _auth_proc = processor
        _auth_resp_cls = _auth_proc.__class__.__init__.__globals__.get("Response")
    except Exception:
        _auth_resp_cls = None
    if _auth_resp_cls is not None:
        return _auth_resp_cls(
            body=json.dumps(_auth_deny, ensure_ascii=False).encode("utf-8"),
            status=401,
            headers={"Cache-Control": "no-store"},
            content_type="application/json",
        )
    try:
        request.set_status(401)
    except Exception:
        pass
    try:
        return json.dumps(_auth_deny, ensure_ascii=False)
    except Exception:
        return _auth_deny
# [fix-iter3 strict-auth v2] end
# [ns-shim] 统一注入 dspy 执行命名空间（Bug qVMyATuR2Z7UuN2P9pbzp / Ua43R1DMO8-7tCb9ybwo8 / j_rE23TAyqp0OV20OTDwv / iFsP7Yklktq_AixTnEWYi）
# 根因：ahserver 的 dspy exec 命名空间不保证注入 `env`（ServerEnv 实例）与模块函数
#      （create_entity / list_entities / *_import …），直写裸名会 NameError，或把 None 拼进 SQL
#      （'select None as None from None'）→ 泛化 500。
# 注入方式（QC#1 整改）：ServerEnv 继承 appPublic.dictObject.DictObject（dict 子类 Singleton），
#      `vars(_ns)` 返回实例 __dict__（实测为空）→ 旧写法 globals().update(vars(_ns)) 实际注入 0 项，
#      shim 形同虚设（5cffb7b 实测记载）。现统一改为**映射取值解析**：
#      `_resolve_delegate(name)` 依次 globals().get(name) → env.get(name) → getattr(env, name, None)；
#      并在 env 为映射时按 `env.items()`（不是 vars()）批量导出到 globals。全 74 个端点共用同一段。
# Bug Dsgk7Hflcyyf_p9ILXYHg 部署态实测：ahserver 预加载全局（request 等）经 builtins 解析，
# 不在 exec-globals 里 —— `globals().get("request")` 恒为 None，旧写法让 5 条契约端点全返 503。
# 改为裸名优先（ahserver 语义），globals 取值仅作兼容兜底。
_ns = None
try:
    _ns = request._run_ns
except Exception:
    _ns = None
if _ns is None:
    _ns = getattr(globals().get("request"), "_run_ns", None)
if _ns is None:
    return {"status": 503, "success": False,
            "message": "[ns-shim] dspy 执行命名空间缺少 request._run_ns，`env` 与模块函数无法解析；"
                       "请检查应用 init() 是否完成 ServerEnv 装配并逐个调用 load_{模块}()"}
env = _ns


def _resolve_delegate(_name):
    """三级映射取值解析委托对象（QC#1 统一内置实现）。

    解析顺序：globals().get(name) → env.get(name) → getattr(env, name, None)。
    全部未命中返回 None，调用方据此走 503 可行动兜底（不退化为泛化 500）。
    """
    _v = globals().get(_name)
    if _v is not None:
        return _v
    try:
        _v = env.get(_name)
    except Exception:
        _v = None
    if _v is not None:
        return _v
    return getattr(env, _name, None)


def _ns_export():
    """把 env 中的名字导出到 dspy globals。

    主路径：映射取值 `env.items()`——ServerEnv 继承 appPublic.dictObject.DictObject（dict 子类），
    这是唯一可靠注入源（实例 __dict__ 实测为空，故不得只用 vars()）。
    回退路径：env 不是映射时（无 items()），退回 `vars(env)` 读实例 __dict__，
    以覆盖属性式挂载的命名空间（不回退会把可用命名空间误判成"未注册"→假 503）。
    """
    _items = None
    try:
        _items = env.items()
    except Exception:
        _items = None
    if _items is None:
        try:
            _items = vars(env).items()
        except Exception:
            _items = None
    if _items is None:
        return
    for _k, _v in list(_items):
        if not isinstance(_k, str) or _k.startswith("__") or _k == "params_kw":
            continue
        globals()[_k] = _v


_ns_export()
for _ldrname in ("load_entity", "load_scene", "load_scense", "load_scense_game",
                 "load_world", "load_world_sync", "load_world_snapshot",
                 "load_runtime", "load_script_engine", "load_sys_ops"):
    _ldr = _resolve_delegate(_ldrname)
    if _ldr is None:
        continue
    try:
        _loaded = _ldr()
        if hasattr(_loaded, "__await__"):
            await _loaded
    except Exception:
        pass
_ns_export()


def _dspy_missing(_name):
    """未注册模块函数的统一兜底：503 + 可行动提示（不要退化成泛化 500）。"""
    return {"status": 503, "success": False,
            "message": "[ns-shim] 模块函数 %s 未注入 dspy 命名空间；"
                       "请确认该模块 init() 已 env.%s = %s 且应用已 load_{模块}() 挂载" % (_name, _name, _name)}
# [/ns-shim]
# -*- coding: utf-8 -*-
# runtime/api/build_entity_tree.dspy —— W-10a 运行态实体树构建
# 入参（二选一，可同时给）：
#   entities  : 实体数组（对象数组或 JSON 字符串）——纯函数路径，无 DB 依赖
#   world_id  : 世界 id —— 服务端路径：先按真实 world_id 读实体，再构树
# 出参：{ok, roots, index, count}；任何异常降级返回，绝不 500
#   （Bug aylOIQTBs7w5MpuSCxn7I：旧版 build_entity_tree 在空输入时返回 list，
#     端点 result['ok']=True 触发 "list indices must be integers or slices, not str"，
#     被降级兜底吞掉 → roots 恒为空。本版强制 dict 口径并支持 world_id 直读。）


def _norm_tree(v):
    """把任意形态的构树结果（None / list / dict）统一成 {roots,index,count}。"""
    if isinstance(v, dict):
        roots = v.get('roots')
        index = v.get('index')
        count = v.get('count')
        if not isinstance(roots, list):
            roots = []
        if not isinstance(index, dict):
            index = {}
        try:
            count = int(count)
        except Exception:
            count = len(index)
        return {'roots': roots, 'index': index, 'count': count}
    if isinstance(v, (list, tuple)):
        # 兼容旧版返回值：list 即根节点数组
        return {'roots': list(v), 'index': {}, 'count': len(v)}
    return {'roots': [], 'index': {}, 'count': 0}


try:
    entities = params_kw.get('entities')
    if entities is None:
        entities = params_kw.get('entity_json')
    if entities is None:
        entities = params_kw.get('data')
    if isinstance(entities, str):
        try:
            entities = json.loads(entities)
        except Exception:
            entities = []
    if entities is not None and not isinstance(entities, list):
        entities = []

    world_id = params_kw.get('world_id') or params_kw.get('id') or ''
    loaded = None

    # 未显式给 entities 且给了 world_id -> 按真实 world_id 从库里读实体（服务端路径）
    if not entities and world_id:
        loaded = await load_world_runtime(world_id=str(world_id))
        if isinstance(loaded, dict):
            got = loaded.get('entities')
            if isinstance(got, list):
                entities = got

    tree = _norm_tree(build_entity_tree(entities or []))
    tree['ok'] = True
    tree['degraded'] = False
    tree['count'] = int(tree.get('count') or 0)
    if world_id:
        tree['world_id'] = str(world_id)
        if isinstance(loaded, dict):
            if loaded.get('degraded'):
                tree['degraded'] = True
                tree['reason'] = 'entity source degraded: ' + str(loaded.get('message') or '')
        tree['entity_count'] = len(entities or [])
    return tree
except Exception as e:
    return {'ok': True, 'degraded': True, 'roots': [], 'index': {}, 'count': 0,
            'message': 'build_entity_tree degraded: ' + str(e)}