312 lines
13 KiB
Python
312 lines
13 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:
|
||
return False
|
||
parsed = urlparse(url)
|
||
host = (parsed.hostname or "").lower()
|
||
query = parse_qs(parsed.query)
|
||
fpid = (query.get("fpid") or [""])[0]
|
||
legacy = host == "frontier-im.douyin.com" and "token" in query and fpid == "9"
|
||
browser_frontier = (
|
||
"frontier" in host
|
||
and host.endswith("zijieapi.com")
|
||
and "access_key" in query
|
||
and "device_id" in query
|
||
and fpid == "9"
|
||
)
|
||
return legacy or browser_frontier
|
||
|
||
|
||
@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 是否已由账号资料 UID 等可靠来源核验。query/user 的 user_uid
|
||
# 不是所有账号的 IM UID，不能据此覆盖已采集/已同步的 my_uid。
|
||
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 里可能已过期的）
|
||
# 账号级公网出口配置来自 accounts 表，不写回 im_session_data，避免网络配置
|
||
# 与登录凭证重复存储。egress_source_ip 是当前服务器探测出的本地绑定地址。
|
||
egress_public_ip: str = ""
|
||
egress_source_ip: str = ""
|
||
egress_auto_attempts: int = 1
|
||
|
||
@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()
|
||
|
||
# 先整段扫描 localStorage，收集字段（避免遍历顺序导致取值不确定）。
|
||
# 关键背景：新版抖音 web 端 __tea_cache_tokens_6383 的 user_unique_id 实际存的是
|
||
# web_id（如 7678646545793812008），并非账号 UID；而 web_runtime_security_uid
|
||
# 才是账号真实 UID（如 2609567359568155）。混合登录态下若把 tea 的 user_unique_id
|
||
# 当 my_uid，会导致 device_id != my_uid，IM 发送被安全网关 KICK。
|
||
ls_sec_uid = "" # web_runtime_security_uid（最可靠的账号 UID 来源）
|
||
ls_web_id = "" # 第一个 tea 条目的 web_id/user_unique_id
|
||
ls_tea_pairs = [] # [(user_unique_id, web_id), ...] 按出现顺序
|
||
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:
|
||
try:
|
||
parsed = json.loads(value)
|
||
if isinstance(parsed, dict):
|
||
wid = str(parsed.get("web_id") or "")
|
||
uid = str(parsed.get("user_unique_id") or "")
|
||
if not ls_web_id:
|
||
ls_web_id = wid or uid
|
||
ls_tea_pairs.append((uid, wid))
|
||
except Exception:
|
||
pass
|
||
if name == "web_runtime_security_uid":
|
||
v = str(value or "")
|
||
if v.isdigit() and not ls_sec_uid:
|
||
ls_sec_uid = v
|
||
|
||
# web_id：extra 显式值 > localStorage tea
|
||
if not web_id:
|
||
web_id = ls_web_id
|
||
|
||
# my_uid 优先级：extra/顶层 > web_runtime_security_uid（真实账号 UID）>
|
||
# tea 的 user_unique_id（仅当与自身 web_id 不同才可信，避免误取 web_id）
|
||
if not my_uid and ls_sec_uid:
|
||
my_uid = int(ls_sec_uid)
|
||
if not my_uid:
|
||
for uid, wid in ls_tea_pairs:
|
||
if uid.isdigit() and not (wid and uid == wid):
|
||
my_uid = int(uid)
|
||
break
|
||
|
||
# device_id 优先级：extra/cookies > web_runtime_security_uid（与账号 UID 绑定）
|
||
if not device_id:
|
||
if ls_sec_uid:
|
||
device_id = ls_sec_uid
|
||
elif my_uid:
|
||
device_id = str(my_uid)
|
||
|
||
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
|
||
|
||
# 最终一致性收敛：protobuf/frontier 的 device_id 优先取 session.device_id
|
||
# （resolve_proto_device_id），若凭证里残留旧设备号（如 www 域
|
||
# web_runtime_security_uid），发送时 device_id != my_uid 会被安全网关
|
||
# 判为设备指纹异常 -> decision=KICK。my_uid 此时已是权威账号 UID，
|
||
# 不一致时以 my_uid 收敛 device_id。
|
||
if my_uid and device_id and str(device_id) != str(my_uid):
|
||
device_id = str(my_uid)
|
||
|
||
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,
|
||
"uid_verified": self.uid_verified,
|
||
"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),
|
||
uid_verified=bool(data.get("uid_verified", False)),
|
||
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 "",
|
||
}
|