更新
This commit is contained in:
Your Name
@@ -0,0 +1,260 @@
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional
|
||||
from urllib.parse import parse_qs, unquote, urlparse
|
||||
|
||||
IM_TOKEN_COOKIES = ("sessionid", "sessionid_ss")
|
||||
|
||||
|
||||
def is_frontier_ws_url(url: str) -> bool:
|
||||
"""判断是否为抖音 IM frontier 长连接地址。
|
||||
|
||||
真实抓包里 host 可能是 frontier-im.douyin.com，也可能是
|
||||
frontierNN-normal.zijieapi.com 这类内部别名，二者都要认。
|
||||
"""
|
||||
if not url or "token=" not in url:
|
||||
return False
|
||||
return "frontier-im.douyin.com" in url or ("frontier" in url and "zijieapi.com" in url)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DouyinImSession:
|
||||
"""抖音 IM 直连所需会话信息（从 Cookie + 浏览器抓包获得）"""
|
||||
|
||||
cookies: dict = field(default_factory=dict)
|
||||
ws_urls: list = field(default_factory=list)
|
||||
device_id: str = ""
|
||||
web_id: str = ""
|
||||
user_agent: str = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||||
)
|
||||
keys_str: str = ""
|
||||
web_protect_str: str = ""
|
||||
my_uid: int = 0
|
||||
# my_uid 是否已用 query/user 接口核验过（采集端推断的 my_uid 可能取错 tea_cache id）
|
||||
uid_verified: bool = False
|
||||
conv_meta: dict = field(default_factory=dict)
|
||||
# 方案 A：直接复用浏览器抓到的真实 frontier 连接凭证（绕开我们自己推导 token/access_key 不准的问题）
|
||||
sdk_cert: str = "" # bd-ticket-guard 客户端证书（frontier sdk_cert / HTTP client-cert）
|
||||
frontier_ts_sign: str = "" # 抓包得到的新鲜 ts_sign（覆盖 web_protect 里可能已过期的）
|
||||
|
||||
@classmethod
|
||||
def from_storage_state(cls, data: dict, extra: Optional[dict] = None) -> "DouyinImSession":
|
||||
extra = extra or {}
|
||||
cookies = {}
|
||||
cookie_items = data.get("cookies", [])
|
||||
priority_names = set(IM_TOKEN_COOKIES)
|
||||
|
||||
def pick_best_cookie(name: str) -> str:
|
||||
matches = [
|
||||
c for c in cookie_items
|
||||
if c.get("name") == name and c.get("value")
|
||||
]
|
||||
if not matches:
|
||||
return ""
|
||||
matches.sort(
|
||||
key=lambda c: (
|
||||
0 if ".douyin.com" in (c.get("domain") or "") else 1,
|
||||
-len(c.get("value") or ""),
|
||||
)
|
||||
)
|
||||
return matches[0].get("value") or ""
|
||||
|
||||
for item in cookie_items:
|
||||
name = item.get("name")
|
||||
if not name or any(c in name for c in "()[]{}'\"\n \t\\"):
|
||||
continue
|
||||
val = item.get("value") or ""
|
||||
if name in priority_names:
|
||||
best = pick_best_cookie(name)
|
||||
if best:
|
||||
cookies[name] = best
|
||||
elif name not in cookies or len(val) > len(cookies.get(name, "")):
|
||||
cookies[name] = val
|
||||
|
||||
device_id = extra.get("device_id") or cookies.get("device_id") or ""
|
||||
web_id = extra.get("web_id") or ""
|
||||
|
||||
keys_str = extra.get("keys_str") or ""
|
||||
web_protect_str = extra.get("web_protect_str") or ""
|
||||
|
||||
# my_uid 优先级：浏览器实时采集(extra) > storage_state 顶层(凭证采集工具手填/抓取)。
|
||||
# 顶层 my_uid 让导入的明文数字 UID 直接生效，避免后端用加密 uid_tt 解析失败而回退联网查询。
|
||||
def _as_uid(v) -> int:
|
||||
try:
|
||||
return int(str(v).strip())
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
my_uid = _as_uid(extra.get("my_uid")) or _as_uid(data.get("my_uid"))
|
||||
user_agent = str(extra.get("user_agent") or data.get("user_agent") or "").strip()
|
||||
|
||||
if not device_id or not web_id or not keys_str or not web_protect_str or not my_uid:
|
||||
for origin in data.get("origins", []):
|
||||
for entry in origin.get("localStorage", []):
|
||||
name = entry.get("name", "")
|
||||
value = entry.get("value", "")
|
||||
if not value:
|
||||
continue
|
||||
if name == "security-sdk/s_sdk_crypt_sdk" and not keys_str:
|
||||
keys_str = value
|
||||
if name == "security-sdk/s_sdk_sign_data_key/web_protect" and not web_protect_str:
|
||||
web_protect_str = value
|
||||
if "tea_cache_tokens" in name and not web_id:
|
||||
try:
|
||||
parsed = json.loads(value)
|
||||
web_id = str(
|
||||
parsed.get("web_id")
|
||||
or parsed.get("user_unique_id")
|
||||
or ""
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
if name == "web_runtime_security_uid" and not device_id:
|
||||
if str(value or "").isdigit():
|
||||
device_id = value
|
||||
if "tea_cache_tokens" in name and not my_uid:
|
||||
try:
|
||||
parsed = json.loads(value)
|
||||
uid = parsed.get("user_unique_id")
|
||||
if uid and str(uid).isdigit():
|
||||
my_uid = int(uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not my_uid:
|
||||
for item in data.get("cookies", []):
|
||||
cname = item.get("name") or ""
|
||||
if cname in ("uid_tt", "uid_tt_ss") and item.get("value"):
|
||||
try:
|
||||
my_uid = int(item.get("value"))
|
||||
break
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
|
||||
if not device_id and my_uid:
|
||||
device_id = str(my_uid)
|
||||
elif not device_id and web_id:
|
||||
device_id = web_id
|
||||
|
||||
ws_urls = list(extra.get("ws_urls") or [])
|
||||
|
||||
# 方案 A：凭证采集工具可携带浏览器抓到的真实 frontier 连接（含 token/sdk_cert/ts_sign）。
|
||||
frontier_ws_url = str(
|
||||
extra.get("frontier_ws_url") or data.get("frontier_ws_url") or ""
|
||||
).strip()
|
||||
sdk_cert = str(extra.get("sdk_cert") or data.get("sdk_cert") or "").strip()
|
||||
frontier_ts_sign = str(
|
||||
extra.get("frontier_ts_sign")
|
||||
or data.get("frontier_ts_sign")
|
||||
or data.get("ts_sign")
|
||||
or ""
|
||||
).strip()
|
||||
if is_frontier_ws_url(frontier_ws_url):
|
||||
# 真实抓包 URL 优先，放在最前面
|
||||
ws_urls = [frontier_ws_url] + [u for u in ws_urls if u != frontier_ws_url]
|
||||
# 从真实 URL 里补抽 sdk_cert / ts_sign（用户只贴了 URL 时）。
|
||||
# 注意：不能用 parse_qs（它会把 + 解成空格，毁掉 base64 证书），用 unquote。
|
||||
def _q(url: str, key: str) -> str:
|
||||
m = re.search(rf"[?&]{re.escape(key)}=([^&\s]+)", url)
|
||||
return unquote(m.group(1)) if m else ""
|
||||
if not sdk_cert:
|
||||
sdk_cert = _q(frontier_ws_url, "sdk_cert")
|
||||
if not frontier_ts_sign:
|
||||
frontier_ts_sign = _q(frontier_ws_url, "ts_sign")
|
||||
|
||||
return cls(
|
||||
cookies=cookies,
|
||||
ws_urls=ws_urls,
|
||||
device_id=str(device_id or ""),
|
||||
web_id=str(web_id or ""),
|
||||
keys_str=keys_str,
|
||||
web_protect_str=web_protect_str,
|
||||
my_uid=my_uid,
|
||||
user_agent=user_agent if user_agent else cls.user_agent,
|
||||
sdk_cert=sdk_cert,
|
||||
frontier_ts_sign=frontier_ts_sign,
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"cookies": self.cookies,
|
||||
"ws_urls": self.ws_urls,
|
||||
"device_id": self.device_id,
|
||||
"web_id": self.web_id,
|
||||
"user_agent": self.user_agent,
|
||||
"keys_str": self.keys_str,
|
||||
"web_protect_str": self.web_protect_str,
|
||||
"my_uid": self.my_uid,
|
||||
"conv_meta": self.conv_meta,
|
||||
"sdk_cert": self.sdk_cert,
|
||||
"frontier_ts_sign": self.frontier_ts_sign,
|
||||
"saved_at": time.time(),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "DouyinImSession":
|
||||
if not data:
|
||||
return cls()
|
||||
return cls(
|
||||
cookies=data.get("cookies") or {},
|
||||
ws_urls=data.get("ws_urls") or [],
|
||||
device_id=str(data.get("device_id") or ""),
|
||||
web_id=str(data.get("web_id") or ""),
|
||||
user_agent=data.get("user_agent") or cls.user_agent,
|
||||
keys_str=str(data.get("keys_str") or ""),
|
||||
web_protect_str=str(data.get("web_protect_str") or ""),
|
||||
my_uid=int(data.get("my_uid") or 0),
|
||||
conv_meta=dict(data.get("conv_meta") or {}),
|
||||
sdk_cert=str(data.get("sdk_cert") or ""),
|
||||
frontier_ts_sign=str(data.get("frontier_ts_sign") or ""),
|
||||
)
|
||||
|
||||
def cookie_header(self) -> str:
|
||||
parts = []
|
||||
for name, value in self.cookies.items():
|
||||
if name and value is not None:
|
||||
parts.append(f"{name}={value}")
|
||||
return "; ".join(parts)
|
||||
|
||||
def has_login(self) -> bool:
|
||||
login_keys = {
|
||||
"sessionid",
|
||||
"sessionid_ss",
|
||||
"sid_tt",
|
||||
"sid_guard",
|
||||
"passport_auth_status",
|
||||
"odin_tt",
|
||||
}
|
||||
return any(self.cookies.get(k) for k in login_keys)
|
||||
|
||||
def can_direct_im(self) -> bool:
|
||||
"""是否具备 Cookie 直连 IM 的最低条件（无需浏览器）"""
|
||||
return self.has_login() and bool(
|
||||
self.cookies.get("sessionid") or self.cookies.get("sessionid_ss")
|
||||
)
|
||||
|
||||
def frontier_ws_url(self) -> Optional[str]:
|
||||
"""仅返回 IM frontier 地址，忽略浏览器抓到的 bytelink 等无关 WS。"""
|
||||
for url in self.ws_urls:
|
||||
if is_frontier_ws_url(url):
|
||||
return url
|
||||
return None
|
||||
|
||||
def sanitize_ws_urls(self) -> None:
|
||||
self.ws_urls = [url for url in self.ws_urls if is_frontier_ws_url(url)]
|
||||
|
||||
def common_params(self) -> dict[str, str]:
|
||||
return {
|
||||
"aid": "6383",
|
||||
"app_name": "douyin_web",
|
||||
"device_platform": "webapp",
|
||||
"channel": "channel_pc_web",
|
||||
"pc_client_type": "1",
|
||||
"version_code": "170400",
|
||||
"version_name": "17.4.0",
|
||||
"device_id": self.device_id or self.web_id or "",
|
||||
"webid": self.web_id or self.device_id or "",
|
||||
}
|
||||
Reference in New Issue
Block a user