101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""IM 凭证规范化（独立版，不依赖客服系统 backend）。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
|
|
def _parse_tea_from_ls(origins: list) -> tuple[str, str]:
|
|
my_uid = ""
|
|
web_id = ""
|
|
ordered = sorted(origins or [], key=lambda o: 0 if "www.douyin.com" in (o.get("origin") or "") else 1)
|
|
for origin in ordered:
|
|
for entry in origin.get("localStorage") or []:
|
|
name = entry.get("name") or ""
|
|
if "tea_cache" not in name.lower():
|
|
continue
|
|
try:
|
|
parsed = json.loads(entry.get("value") or "{}")
|
|
except Exception:
|
|
continue
|
|
uid = str(parsed.get("user_unique_id") or "").strip()
|
|
wid = str(parsed.get("web_id") or "").strip()
|
|
if uid and wid and uid == wid and len(uid) > 12:
|
|
continue
|
|
if uid and uid.isdigit() and not my_uid:
|
|
my_uid = uid
|
|
if wid and wid.isdigit() and not web_id:
|
|
web_id = wid
|
|
if my_uid and web_id:
|
|
return my_uid, web_id or my_uid
|
|
return my_uid, web_id or my_uid
|
|
|
|
|
|
def _ws_device_id(url: str) -> str:
|
|
m = re.search(r"[?&]device_id=([^&\s]+)", url or "")
|
|
return unquote(m.group(1)) if m else ""
|
|
|
|
|
|
def normalize_storage_state_for_im(data: dict) -> dict:
|
|
if not isinstance(data, dict) or not isinstance(data.get("cookies"), list):
|
|
return data
|
|
data = dict(data)
|
|
origins = data.get("origins") or []
|
|
tea_uid, _tea_web_id = _parse_tea_from_ls(origins)
|
|
|
|
ws_url = str(data.get("frontier_ws_url") or "")
|
|
is_creator_ws = "aid=2906" in ws_url and "sdk_cert=" in ws_url
|
|
|
|
if is_creator_ws:
|
|
ws_dev = _ws_device_id(ws_url)
|
|
if ws_dev and ws_dev.isdigit():
|
|
data["my_uid"] = int(ws_dev)
|
|
my_uid = ws_dev
|
|
else:
|
|
my_uid = str(data.get("my_uid") or tea_uid or "")
|
|
if my_uid.isdigit():
|
|
data["my_uid"] = int(my_uid)
|
|
else:
|
|
my_uid = ""
|
|
elif tea_uid:
|
|
data["my_uid"] = int(tea_uid)
|
|
my_uid = tea_uid
|
|
else:
|
|
my_uid = str(data.get("my_uid") or "")
|
|
if my_uid.isdigit():
|
|
data["my_uid"] = int(my_uid)
|
|
else:
|
|
my_uid = ""
|
|
|
|
if ws_url and my_uid and not is_creator_ws:
|
|
ws_dev = _ws_device_id(ws_url)
|
|
token = ""
|
|
try:
|
|
token = parse_qs(urlparse(ws_url).query).get("token", [""])[0]
|
|
except Exception:
|
|
pass
|
|
looks_built = bool(token) and len(token) < 40
|
|
if ws_dev and ws_dev.isdigit() and ws_dev != my_uid and looks_built:
|
|
data.pop("frontier_ws_url", None)
|
|
data.pop("sdk_cert", None)
|
|
data.pop("ts_sign", None)
|
|
data.pop("frontier_ws_built", None)
|
|
|
|
for origin in origins:
|
|
ls = origin.get("localStorage") or []
|
|
fixed = False
|
|
for entry in ls:
|
|
if entry.get("name") == "web_runtime_security_uid":
|
|
fixed = True
|
|
val = str(entry.get("value") or "")
|
|
if not val.isdigit() and my_uid:
|
|
entry["value"] = my_uid
|
|
break
|
|
if not fixed and my_uid:
|
|
ls.append({"name": "web_runtime_security_uid", "value": my_uid})
|
|
origin["localStorage"] = ls
|
|
|
|
data["origins"] = origins
|
|
return data
|