259 lines
13 KiB
Python
259 lines
13 KiB
Python
import qrcode
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
from wechat_ipad.providers.server_864.base import Server864APIClientBase
|
||
|
||
|
||
class LoginMixin(Server864APIClientBase):
|
||
"""864 登录相关接口。"""
|
||
|
||
@staticmethod
|
||
def _normalize_online_text(value) -> str:
|
||
"""把 864 各种登录态字段压平成便于判断的字符串。"""
|
||
return str(value or "").strip().lower()
|
||
|
||
def _is_online_from_login_status_payload(self, data: dict | None) -> bool:
|
||
"""根据 `GetLoginStatus` 的返回结构判断当前是否在线。
|
||
|
||
设计说明：
|
||
1. 864 不同版本对在线态字段命名并不统一，可能是 `loginState/status/state/isLogin` 中的任意一种；
|
||
2. 这里集中做一次宽松识别，避免上层 runtime 到处散落同类判断；
|
||
3. 后续如果 864 新版本再补字段，只需要在这里扩展，不必改多处业务逻辑。
|
||
"""
|
||
payload = dict(data or {})
|
||
normalized_login_state = self._normalize_online_text(
|
||
self._pick_first(payload, "loginState", "LoginState", "status_text", "statusText")
|
||
)
|
||
raw_login_state = self._pick_first(payload, "loginState", "LoginState")
|
||
normalized_status = self._normalize_online_text(
|
||
self._pick_first(payload, "status", "Status", "state_text", "stateText")
|
||
)
|
||
state_value = self._pick_first(payload, "state", "State")
|
||
login_flag = self._pick_first(payload, "isLogin", "IsLogin", "online", "Online", "isOnline", "IsOnline")
|
||
login_err_msg = self._normalize_online_text(
|
||
self._pick_first(payload, "loginErrMsg", "LoginErrMsg", "msg", "message")
|
||
)
|
||
|
||
if normalized_login_state in {"online", "已登录", "在线"}:
|
||
return True
|
||
# 864 的 `GetLoginStatus` 在你当前这版 server 里会直接返回 `loginState: 1`：
|
||
# 1. 这不是扫码阶段的 `CheckLoginStatus.state`，而是服务端自身维护的登录态枚举；
|
||
# 2. 之前这里只识别到了字符串 `"online"`，导致明明已经在线却仍被判成未登录；
|
||
# 3. 这里把常见数字态一并纳入在线判定，避免后续资料拉取与前端显示被卡住。
|
||
try:
|
||
normalized_login_state_value = int(raw_login_state or 0)
|
||
except (TypeError, ValueError):
|
||
normalized_login_state_value = 0
|
||
if normalized_login_state_value in {1, 2}:
|
||
return True
|
||
if normalized_status in {"online", "已登录", "在线"}:
|
||
return True
|
||
if "在线状态良好" in login_err_msg or "账号在线" in login_err_msg:
|
||
return True
|
||
if isinstance(login_flag, bool):
|
||
return login_flag
|
||
if str(login_flag or "").strip().lower() in {"true", "1", "online"}:
|
||
return True
|
||
try:
|
||
normalized_state_value = int(state_value or 0)
|
||
except (TypeError, ValueError):
|
||
normalized_state_value = 0
|
||
return normalized_state_value in {1, 2}
|
||
|
||
def _extract_login_identity_from_status(self, data: dict | None) -> dict:
|
||
"""从 `GetLoginStatus` 返回中提取尽可能多的账号身份字段。"""
|
||
payload = dict(data or {})
|
||
return {
|
||
"wxid": str(self._pick_first(payload, "wxid", "Wxid", "UserName", "userName") or "").strip(),
|
||
"nickname": str(self._pick_first(payload, "nick_name", "nickName", "NickName", "nickname") or "").strip(),
|
||
"alias": str(self._pick_first(payload, "alias", "Alias", "wechatId", "WeChatId") or "").strip(),
|
||
"phone": str(self._pick_first(payload, "mobile", "Mobile", "phone", "Phone") or "").strip(),
|
||
"signature": str(self._pick_first(payload, "signature", "Signature") or "").strip(),
|
||
}
|
||
|
||
@staticmethod
|
||
def _normalize_login_way(login_way: str) -> str:
|
||
"""标准化 864 `GetLoginQrCodeNewX` 的 way 参数。"""
|
||
normalized_way = str(login_way or "").strip().lower()
|
||
return normalized_way if normalized_way in {"mac", "win", "harmony", "car", "watch"} else "mac"
|
||
|
||
def _extract_qr_response(self, data) -> tuple[str, str]:
|
||
"""从 864 的二维码返回结构中提取 uuid 与二维码地址。"""
|
||
qr_code_url = (
|
||
self._pick_first(data, "QrCodeUrl", "QRCodeUrl", "qrCodeUrl")
|
||
or self._pick_first(data, "QrUrl", "QRUrl", "qrUrl")
|
||
or self._pick_first(self._pick_first(data, "Qrcode", "QrCode", "qrcode") or {}, "Src", "src")
|
||
or ""
|
||
)
|
||
uuid = self._pick_first(data, "UUID", "Uuid", "uuid") or ""
|
||
if not uuid and qr_code_url:
|
||
# 864 的真实返回经常只给“二维码图片地址”，并不会直接带出 uuid：
|
||
# 1. 图床链接 query 中的 `data=http://weixin.qq.com/x/<uuid>` 才是真正扫码值；
|
||
# 2. Dashboard 当前以 uuid 作为刷新与历史记录主键，缺失后前端体验会明显变差；
|
||
# 3. 因此这里统一做一次反解，让 New / NewX 两种接口都复用同一份展示逻辑。
|
||
parsed_qs = parse_qs(urlparse(str(qr_code_url)).query)
|
||
scan_data = str((parsed_qs.get("data") or [""])[0] or "")
|
||
if "/x/" in scan_data:
|
||
uuid = scan_data.rsplit("/x/", 1)[-1].strip()
|
||
return str(uuid), str(qr_code_url)
|
||
|
||
async def get_qr_code(
|
||
self,
|
||
device_name: str = "",
|
||
device_id: str = "",
|
||
proxy=None,
|
||
print_qr: bool = False,
|
||
login_qr_api: str = "new_x",
|
||
login_way: str = "mac",
|
||
) -> tuple[str, str]:
|
||
"""获取 864 登录二维码。
|
||
|
||
说明：
|
||
1. 864 不依赖 855 的 `device_name/device_id` 入参，但保留参数签名以兼容上层调用；
|
||
2. `proxy` 当前仅保留兼容占位，后续如需补实际代理登录，可直接映射到 swagger 的 Proxy 字段；
|
||
3. 返回值继续保持 `(uuid, url)`，方便 Dashboard 与运行时共用同一套二维码展示逻辑。
|
||
"""
|
||
del device_name, device_id
|
||
proxy_value = ""
|
||
if proxy is not None:
|
||
proxy_value = getattr(proxy, "proxy", "") or ""
|
||
|
||
normalized_login_qr_api = str(login_qr_api or "new_x").strip().lower()
|
||
normalized_login_way = self._normalize_login_way(login_way)
|
||
|
||
if normalized_login_qr_api in {"harmony_api", "harmony", "harmony_login_api"}:
|
||
# `HarmonyLoginApi` 是 864 服务端单独暴露的一条鸿蒙二维码链路：
|
||
# 1. 用户当前希望在 Dashboard 上显式切换“鸿蒙专用二维码”和“标准 New 二维码”；
|
||
# 2. 该接口的请求模型与 `GetLoginQrCodeNew` 一致，仍然只需要 `Proxy/Check`；
|
||
# 3. 因此这里单独加一个模式键，不去复用 `new_x + way=harmony`，避免两条链路在运维上混淆。
|
||
data = await self._request_data(
|
||
"post",
|
||
"/login/HarmonyLoginApi",
|
||
json_body={"Proxy": proxy_value, "Check": False},
|
||
timeout=30,
|
||
)
|
||
uuid, qr_url = self._extract_qr_response(data)
|
||
elif normalized_login_qr_api in {"new_x", "x", "newx"}:
|
||
try:
|
||
# NewX 是当前 864 联调里更完整的一条登录链路：
|
||
# 1. 它支持 `Way` 指定登录端形态，兼容更多 server 变体；
|
||
# 2. 用户实测也验证该接口可正常返回二维码；
|
||
# 3. 若某些旧版 864 仍未提供 NewX，则自动回退到旧接口，避免新配置直接打挂启动。
|
||
data = await self._request_data(
|
||
"post",
|
||
"/login/GetLoginQrCodeNewX",
|
||
json_body={"Proxy": proxy_value, "Check": False, "Way": normalized_login_way},
|
||
timeout=30,
|
||
)
|
||
uuid, qr_url = self._extract_qr_response(data)
|
||
except Exception:
|
||
data = await self._request_data(
|
||
"post",
|
||
"/login/GetLoginQrCodeNew",
|
||
json_body={"Proxy": proxy_value, "Check": False},
|
||
timeout=30,
|
||
)
|
||
uuid, qr_url = self._extract_qr_response(data)
|
||
elif normalized_login_qr_api in {"new", "legacy_new"}:
|
||
data = await self._request_data(
|
||
"post",
|
||
"/login/GetLoginQrCodeNew",
|
||
json_body={"Proxy": proxy_value, "Check": False},
|
||
timeout=30,
|
||
)
|
||
uuid, qr_url = self._extract_qr_response(data)
|
||
else:
|
||
data = await self._request_data(
|
||
"post",
|
||
"/login/GetLoginQrCodeNew",
|
||
json_body={"Proxy": proxy_value, "Check": False},
|
||
timeout=30,
|
||
)
|
||
uuid, qr_url = self._extract_qr_response(data)
|
||
|
||
if print_qr and uuid:
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"http://weixin.qq.com/x/{uuid}")
|
||
qr.make(fit=True)
|
||
qr.print_ascii()
|
||
|
||
return str(uuid), str(qr_url)
|
||
|
||
async def check_login_status(self) -> tuple[bool, dict]:
|
||
"""检查当前二维码登录状态。"""
|
||
data = await self._request_data("get", "/login/CheckLoginStatus", timeout=20)
|
||
normalized = dict(data or {})
|
||
# 864 扫码后可能不会直接变成 online，而是先返回一个安全验证链接：
|
||
# 1. 该字段在 swagger / 实机联调里都存在，但大小写并不稳定；
|
||
# 2. 上层 runtime 只关心统一字段名，因此这里先做一层归一化；
|
||
# 3. 这样 Dashboard 只需要消费 `verification_url`，不用感知各 server 的原始差异。
|
||
verification_url = self._pick_first(
|
||
normalized,
|
||
"VerificationUrl",
|
||
"VerificationURL",
|
||
"verificationUrl",
|
||
"verification_url",
|
||
)
|
||
if verification_url:
|
||
normalized["verification_url"] = str(verification_url).strip()
|
||
# 864 的扫码状态字段在不同版本里语义并不完全一致：
|
||
# 1. 上层运行时需要知道原始 state，才能区分“等待扫码 / 安全验证 / 终态缓存未清理”等场景；
|
||
# 2. 因此这里除了保留旧的 `state` 字段外，再显式放一份 `raw_state` 供 runtime 做分支；
|
||
# 3. 同时把昵称、头像这些已扫描用户信息一并透出，便于 Dashboard 展示最新状态。
|
||
state = int(normalized.get("state", 0) or 0)
|
||
normalized["raw_state"] = state
|
||
if self._pick_first(normalized, "nick_name", "nickName", "NickName"):
|
||
normalized["nick_name"] = str(
|
||
self._pick_first(normalized, "nick_name", "nickName", "NickName") or ""
|
||
).strip()
|
||
if self._pick_first(normalized, "head_img_url", "headImgUrl", "HeadImgUrl"):
|
||
normalized["head_img_url"] = str(
|
||
self._pick_first(normalized, "head_img_url", "headImgUrl", "HeadImgUrl") or ""
|
||
).strip()
|
||
login_state = str(normalized.get("loginState", "") or "").strip().lower()
|
||
return state == 2 or login_state == "online", normalized
|
||
|
||
async def get_init_status(self) -> bool:
|
||
"""检查 server 侧初始化是否完成。"""
|
||
data = await self._request_data("get", "/login/GetInItStatus", timeout=15)
|
||
return bool(data)
|
||
|
||
async def awaken_login(self, wxid: str = "") -> dict:
|
||
"""触发 864 的唤醒登录。"""
|
||
del wxid
|
||
return await self._request_data("post", "/login/WakeUpLogin", timeout=30)
|
||
|
||
async def get_login_status(self, auto_login: bool = True) -> dict:
|
||
"""获取 864 在线状态。"""
|
||
data = await self._request_data(
|
||
"get",
|
||
"/login/GetLoginStatus",
|
||
params={"autoLogin": str(bool(auto_login)).lower()},
|
||
timeout=20,
|
||
)
|
||
normalized = dict(data or {})
|
||
# 把最常用的在线态判断和身份字段提前归一化：
|
||
# 1. 这样上层只要消费 `is_online` / `wxid` / `nickname` 这些统一键，不必感知原始 swagger 字段差异；
|
||
# 2. 与 `CheckLoginStatus` 一样，Dashboard 和 runtime 都能共享同一份兼容结果；
|
||
# 3. 也方便后续把 864 的不同 server 版本收敛到更薄的一层 provider 适配。
|
||
normalized["is_online"] = self._is_online_from_login_status_payload(normalized)
|
||
normalized.update(self._extract_login_identity_from_status(normalized))
|
||
return normalized
|
||
|
||
async def log_out(self) -> bool:
|
||
"""退出当前 864 登录态。"""
|
||
try:
|
||
# 不同 864 版本里退出接口命名存在差异：
|
||
# 1. 当前项目早期接的是 `/login/LogOutRequest`；
|
||
# 2. 用户本地 864 源码中实际注册的是 `/login/LogOut`；
|
||
# 3. 因此这里优先尝试较新的 `/LogOut`，失败后再回退到旧路径，降低版本切换成本。
|
||
await self._request_data("get", "/login/LogOut", timeout=15)
|
||
except Exception:
|
||
await self._request_data("get", "/login/LogOutRequest", timeout=15)
|
||
return True
|