scense/modules/scense/wwwroot/api/list_scense.dspy

# [fix-iter17 auth-gate v3] fail-closed 登录态门禁 —— 勿删
# 归属 Bug：OmWRbpwPZRpNe9r2hkKWC / uabpjOGelxJTknG9vqO9V / G-l57of8enoipnJDg17Uz
# 口径：本块必须是文件第一条可执行语句（在命名空间 shim / 业务分支之前），任何匿名请求
#       （无有效会话）一律返回真 HTTP 401，**不返回任何业务数据**；Response 类解析不到时
#       raise（→500）硬失败，绝不降级为 200——R26~R27 的两次放行均源于「判据失真 + 降级返回」。
def _r17_uid_of(_v):
    if not _v:
        return ""
    _BAD = ("", "none", "null", "anonymous", "anonymoususer", "anon", "0")
    if isinstance(_v, str):
        _s = _v.strip()
        return "" if _s.lower() in _BAD else _s
    if isinstance(_v, (list, tuple, set, bytes, bytearray)):
        return ""
    _KEYS = ("id", "userid", "user_id", "user", "username", "name", "uid", "account")
    if isinstance(_v, dict):
        for _k in _KEYS:
            try:
                _x = _v.get(_k)
            except Exception:
                _x = None
            if _x:
                _s = str(_x).strip()
                if _s.lower() not in _BAD:
                    return _s
        return ""
    for _k in _KEYS:
        try:
            _x = getattr(_v, _k, None)
        except Exception:
            _x = None
        if _x:
            _s = str(_x).strip()
            if _s.lower() not in _BAD:
                return _s
    return ""


async def _r17_uid():
    """取登录用户标识；取不到返回 ''。裸名解析（ahserver 预加载全局在 builtins）。"""
    for _n in ("get_userinfo", "get_user", "get_username"):
        try:
            if _n == "get_userinfo":
                _f = get_userinfo
            elif _n == "get_user":
                _f = get_user
            else:
                _f = get_username
        except Exception:
            continue
        try:
            _r = _f()
            if hasattr(_r, "__await__"):
                _r = await _r
        except Exception:
            continue
        _u = _r17_uid_of(_r)
        if _u:
            return _u
    return ""


_r17_uid_val = ""
try:
    _r17_uid_val = await _r17_uid()
except Exception:
    _r17_uid_val = ""

if not _r17_uid_val:
    _r17_deny = {
        "status": 401,
        "success": False,
        "ok": False,
        "need_login": True,
        "message": "unauthorized: login required",
        "rows": [],
        "data": [],
        "total": 0,
    }
    _r17_proc = None
    try:
        _r17_proc = processor
    except Exception:
        _r17_proc = globals().get("processor")
    _r17_rc = None
    try:
        if _r17_proc is not None:
            _r17_rc = _r17_proc.__class__.__init__.__globals__.get("Response")
    except Exception:
        _r17_rc = None
    if _r17_rc is None:
        raise RuntimeError(
            "[fix-iter17 auth-gate v3] 未登录且无法解析 aiohttp Response：门禁 hard-fail"
            "（宁 500 不放行，绝不返回业务数据）"
        )
    return _r17_rc(
        body=json.dumps(_r17_deny, ensure_ascii=False).encode("utf-8"),
        status=401,
        headers={"Cache-Control": "no-store", "X-Scene-Auth": "denied"},
        content_type="application/json",
    )
# [fix-iter17 auth-gate v3] end

# [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]
"""GET /scense/api/list_scense.dspy —— 展示会话列表（验收契约路径）。

QC#4 整改（硬编码空数组桩）：原实现是一个**恒返回常量空结果的桩**
（`return json.dumps({'ok': True, 'total': 0, 'rows': [], 'note': ...})`），
不读任何数据源，任何调用方拿到的都是空列表——它让 `/scense/api/list_scense.dspy`
从「500 fpath is None」变成了「200 但永远空」，属于用桩掩盖契约缺失，不可接受。

scense 是游戏**展示前端**模块，自身不持有数据表（无 models/ 无 json/），
会话数据的权威持有者是 scense_game 模块的 `game_session` 表。
scense 模块自己的同源 list 实现是 `scense.core.game_list_sessions(user_id)`：
经 `scense/init.py` 的 `load_scense()` 注册到 ServerEnv（`env.game_list_sessions = ...`），
内部走 `_get_dbname()` 取库名 + `sor.R('game_session', {...})` 按 updated_at 倒序取最近 20 条。

故本文件改为**委托该同源实现**，并按登录用户隔离（多用户不串数据），
不再返回任何硬编码常量。
"""
try:
    uid = ''
    try:
        info = get_userinfo()
    except Exception:
        info = None
    if isinstance(info, dict):
        uid = info.get('id') or info.get('userid') or info.get('user') or ''
    if not uid:
        try:
            uid = get_user() or ''
        except Exception:
            uid = ''
    rows = await game_list_sessions(uid)
    rows = rows or []
    return json.dumps({'ok': True, 'total': len(rows), 'rows': rows},
                      ensure_ascii=False, default=str)
except Exception as e:
    return json.dumps({'ok': False, 'total': 0, 'rows': [], 'error': str(e)},
                      ensure_ascii=False)