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 "", }