261 lines
10 KiB
Python
261 lines
10 KiB
Python
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 "",
|
||
}
|