40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
"""PocketBase：校验用户令牌与 live_allowed；供 /start /restart 可选强制。"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any, Tuple
|
||
|
||
import httpx
|
||
|
||
PB_URL = os.environ.get("WEB2_PB_URL", "https://pocketbase.hackrobot.cn").rstrip("/")
|
||
REQUIRE_PB = os.environ.get("WEB2_REQUIRE_PB_AUTH", "").strip().lower() in ("1", "true", "yes")
|
||
|
||
|
||
async def pb_auth_refresh_and_validate(token: str) -> Tuple[bool, str, dict[str, Any]]:
|
||
"""
|
||
使用 auth-refresh 校验令牌并拉取最新用户记录。
|
||
返回 (ok, error_message, record)
|
||
"""
|
||
t = (token or "").strip()
|
||
if not t:
|
||
return False, "缺少 PocketBase 授权（请在手机端登录会员账号）", {}
|
||
url = f"{PB_URL}/api/collections/users/auth-refresh"
|
||
try:
|
||
async with httpx.AsyncClient(timeout=20.0) as client:
|
||
r = await client.post(url, headers={"Authorization": f"Bearer {t}"})
|
||
except Exception as e:
|
||
return False, f"PocketBase 不可达: {e}", {}
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
return False, "PocketBase 响应无效", {}
|
||
if r.status_code != 200:
|
||
msg = str(data.get("message") or "授权无效或已过期")
|
||
return False, msg, {}
|
||
rec = data.get("record") or {}
|
||
if not isinstance(rec, dict):
|
||
return False, "用户记录异常", {}
|
||
if rec.get("live_allowed") is False:
|
||
return False, "账号已被管理员停用远程直播权限", {}
|
||
return True, "", rec
|