915 lines
32 KiB
Python
915 lines
32 KiB
Python
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from urllib.parse import parse_qs, unquote, urlparse
|
||
|
||
from douyin_collector_bootstrap import browser_launch_kwargs, fix_playwright_env
|
||
|
||
fix_playwright_env()
|
||
|
||
from app_paths import app_root, clear_browser_ready, is_gui_mode, launch_splash, signal_browser_ready, use_incognito_ui
|
||
from douyin_collector_server import (
|
||
STATE,
|
||
append_log,
|
||
bind_start_event,
|
||
open_browser,
|
||
publish_error,
|
||
publish_result,
|
||
start_server,
|
||
wait_for_shutdown,
|
||
)
|
||
|
||
try:
|
||
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError
|
||
except Exception as exc:
|
||
msg = f"缺少 Playwright：{exc}"
|
||
if is_gui_mode():
|
||
publish_error(msg + "\n\n请运行 fix_browser.bat 或 setup.bat")
|
||
wait_for_shutdown()
|
||
else:
|
||
print("缺少 Playwright，请先安装：")
|
||
print(" pip install playwright")
|
||
print(" python -m playwright install chromium")
|
||
print(f"\n错误：{exc}")
|
||
input("\n按回车退出...")
|
||
raise SystemExit(1)
|
||
|
||
|
||
class _UiLogWriter:
|
||
"""GUI 模式下把 print 同步到网页日志。"""
|
||
|
||
def __init__(self, mirror) -> None:
|
||
self._mirror = mirror
|
||
|
||
def write(self, text: str) -> None:
|
||
if not text:
|
||
return
|
||
if self._mirror:
|
||
self._mirror.write(text)
|
||
for line in text.splitlines():
|
||
if line.strip():
|
||
append_log(line.strip())
|
||
|
||
def flush(self) -> None:
|
||
if self._mirror:
|
||
self._mirror.flush()
|
||
|
||
|
||
def _install_ui_logging() -> None:
|
||
sys.stdout = _UiLogWriter(getattr(sys, "__stdout__", None))
|
||
sys.stderr = _UiLogWriter(getattr(sys, "__stderr__", None))
|
||
|
||
CHAT_URL = "https://creator.douyin.com/creator-micro/data/following/chat"
|
||
OUT_FILE = app_root() / "douyin_storage_state.json"
|
||
|
||
BROWSER_ARGS = [
|
||
"--disable-blink-features=AutomationControlled",
|
||
"--incognito",
|
||
]
|
||
|
||
WS_HOOK_INIT = """
|
||
(() => {
|
||
function __dyIsImWs(u) {
|
||
u = String(u || '');
|
||
if (!u || !/^wss?:/i.test(u)) return false;
|
||
if (!/[?&]token=/.test(u)) return false;
|
||
if (/frontier/i.test(u)) return true;
|
||
if (/zijieapi\\.com/i.test(u)) return true;
|
||
return /\\/ws\\/v2/i.test(u) && /douyin/i.test(u);
|
||
}
|
||
function __dySaveWs(u) {
|
||
if (!__dyIsImWs(u)) return;
|
||
let a = [];
|
||
try { a = JSON.parse(sessionStorage.getItem('__dy_captured_ws') || '[]'); } catch (e) {}
|
||
if (a.indexOf(u) < 0) a.unshift(u);
|
||
try { sessionStorage.setItem('__dy_captured_ws', JSON.stringify(a.slice(0, 8))); } catch (e) {}
|
||
}
|
||
if (window.__dyWsHookInstalled) return;
|
||
window.__dyWsHookInstalled = 1;
|
||
const O = window.WebSocket;
|
||
if (!O) return;
|
||
try {
|
||
const origSend = O.prototype.send;
|
||
O.prototype.send = function () {
|
||
try { __dySaveWs(this.url); } catch (e) {}
|
||
return origSend.apply(this, arguments);
|
||
};
|
||
} catch (e) {}
|
||
function W(u, p) {
|
||
try { __dySaveWs(String(u || '')); } catch (e) {}
|
||
return arguments.length > 1 ? new O(u, p) : new O(u);
|
||
}
|
||
W.prototype = O.prototype;
|
||
Object.keys(O).forEach(k => { try { W[k] = O[k]; } catch (e) {} });
|
||
['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED'].forEach(k => { if (k in O) W[k] = O[k]; });
|
||
window.WebSocket = W;
|
||
try {
|
||
const perfObs = new PerformanceObserver(list => {
|
||
list.getEntries().forEach(e => __dySaveWs(e.name));
|
||
});
|
||
perfObs.observe({ entryTypes: ['resource'] });
|
||
(performance.getEntries() || []).forEach(e => __dySaveWs(e.name));
|
||
} catch (e) {}
|
||
})();
|
||
"""
|
||
|
||
HARVEST_WS_JS = """
|
||
() => {
|
||
const found = [];
|
||
function add(u) {
|
||
u = String(u || '');
|
||
if (!u || !/^wss?:/i.test(u) || !/[?&]token=/.test(u)) return;
|
||
if (!(/frontier/i.test(u) || /zijieapi\\.com/i.test(u) || (/\\/ws\\/v2/i.test(u) && /douyin|creator/i.test(u)))) return;
|
||
if (found.indexOf(u) < 0) found.unshift(u);
|
||
}
|
||
function scanText(t) {
|
||
t = String(t || '');
|
||
const re = /wss?:\\/\\/[^\\s"'`,)\\]}]+/gi;
|
||
let m;
|
||
while ((m = re.exec(t)) !== null) add(m[0]);
|
||
}
|
||
try { JSON.parse(sessionStorage.getItem('__dy_captured_ws') || '[]').forEach(add); } catch (e) {}
|
||
try { (performance.getEntries() || []).forEach(e => add(e.name)); } catch (e) {}
|
||
try {
|
||
for (let i = 0; i < localStorage.length; i++) scanText(localStorage.getItem(localStorage.key(i)));
|
||
for (let j = 0; j < sessionStorage.length; j++) scanText(sessionStorage.getItem(sessionStorage.key(j)));
|
||
} catch (e) {}
|
||
return found;
|
||
}
|
||
"""
|
||
|
||
CHAT_SCROLL_JS = """
|
||
() => {
|
||
try { window.scrollTo(0, document.body.scrollHeight); } catch (e) {}
|
||
try {
|
||
const nodes = document.querySelectorAll('[class*="list"], [class*="chat"], [class*="message"]');
|
||
nodes.forEach(n => { try { n.dispatchEvent(new Event('scroll', { bubbles: true })); } catch (e) {} });
|
||
} catch (e) {}
|
||
}
|
||
"""
|
||
|
||
LS_KEYS = [
|
||
"security-sdk/s_sdk_crypt_sdk",
|
||
"security-sdk/s_sdk_sign_data_key/web_protect",
|
||
"tea_cache_tokens",
|
||
"web_runtime_security_uid",
|
||
"security-sdk/s_sdk_cert_key",
|
||
"security-sdk/s_sdk_sign_data_key/token",
|
||
]
|
||
|
||
|
||
def cookie_header(cookies):
|
||
return "; ".join(
|
||
f"{c.get('name')}={c.get('value')}"
|
||
for c in cookies
|
||
if c.get("name") and c.get("value") is not None
|
||
)
|
||
|
||
|
||
def parse_web_protect(raw):
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
obj = json.loads(raw)
|
||
if isinstance(obj, str):
|
||
obj = json.loads(obj)
|
||
data = obj.get("data") if isinstance(obj, dict) else None
|
||
if isinstance(data, str):
|
||
return json.loads(data)
|
||
if isinstance(data, dict):
|
||
return data
|
||
return obj if isinstance(obj, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def get_query(url, key):
|
||
try:
|
||
value = parse_qs(urlparse(url).query).get(key, [""])[0]
|
||
return unquote(value) if value else ""
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def pick_local_storage(ls, exact, contains=()):
|
||
if ls.get(exact):
|
||
return ls[exact]
|
||
for key, value in ls.items():
|
||
lower = key.lower()
|
||
if value and any(item in lower for item in contains):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def parse_jsonish(raw):
|
||
if not raw:
|
||
return None
|
||
try:
|
||
obj = json.loads(raw)
|
||
if isinstance(obj, str):
|
||
try:
|
||
return json.loads(obj)
|
||
except Exception:
|
||
return obj
|
||
return obj
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def find_security_value(ls, expected_fields, key_contains=()):
|
||
for key, value in ls.items():
|
||
lower = key.lower()
|
||
if key_contains and not any(item in lower for item in key_contains):
|
||
continue
|
||
obj = parse_jsonish(value)
|
||
candidates = []
|
||
if isinstance(obj, dict):
|
||
data = obj.get("data")
|
||
data_obj = parse_jsonish(data) if isinstance(data, str) else data
|
||
if isinstance(data_obj, dict):
|
||
candidates.append(data_obj)
|
||
candidates.append(obj)
|
||
for item in candidates:
|
||
if all(item.get(field) for field in expected_fields):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def build_web_protect_from_token(raw):
|
||
obj = parse_jsonish(raw)
|
||
if not isinstance(obj, dict):
|
||
return ""
|
||
data = obj.get("data")
|
||
data_obj = parse_jsonish(data) if isinstance(data, str) else data
|
||
if not isinstance(data_obj, dict):
|
||
return ""
|
||
if not (data_obj.get("ticket") and data_obj.get("ts_sign") and data_obj.get("client_cert")):
|
||
return ""
|
||
return json.dumps({
|
||
"data": json.dumps({
|
||
"ticket": data_obj.get("ticket"),
|
||
"ts_sign": data_obj.get("ts_sign"),
|
||
"client_cert": data_obj.get("client_cert"),
|
||
"create_time": int(time.time()),
|
||
}, ensure_ascii=False)
|
||
}, ensure_ascii=False)
|
||
|
||
|
||
def normalize_tea_key(ls):
|
||
raw = pick_local_storage(ls, "tea_cache_tokens", ["tea_cache_tokens"])
|
||
if raw:
|
||
uid, _ = _parse_tea_uid(raw)
|
||
if uid:
|
||
return raw
|
||
for key in ("__tea_cache_tokens_6383", "__tea_cache_tokens_1661", "__tea_cache_tokens_5231", "__tea_cache_tokens_2906"):
|
||
raw = ls.get(key)
|
||
if not raw:
|
||
continue
|
||
uid, _ = _parse_tea_uid(str(raw))
|
||
if uid:
|
||
return str(raw)
|
||
return ""
|
||
|
||
|
||
def _parse_tea_obj(raw):
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
obj = json.loads(raw)
|
||
return obj if isinstance(obj, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _parse_tea_uid(raw: str) -> tuple[str, str]:
|
||
"""返回 (user_unique_id, web_id)，忽略伪造条目（user_unique_id == web_id）。"""
|
||
obj = _parse_tea_obj(raw)
|
||
uid = str(obj.get("user_unique_id") or "").strip()
|
||
wid = str(obj.get("web_id") or "").strip()
|
||
if uid and wid and uid == wid and len(uid) > 12:
|
||
return "", wid
|
||
return uid, wid
|
||
|
||
|
||
def extract_my_uid_from_ls(ls: dict) -> str:
|
||
"""IM 数字 UID：优先 tea_cache_tokens，跳过 creator 侧伪造的 __tea_cache_tokens_*。"""
|
||
priority_keys = (
|
||
"tea_cache_tokens",
|
||
"__tea_cache_tokens_6383",
|
||
"__tea_cache_tokens_1661",
|
||
"__tea_cache_tokens_5231",
|
||
"__tea_cache_tokens_2906",
|
||
)
|
||
for key in priority_keys:
|
||
raw = ls.get(key)
|
||
if not raw:
|
||
continue
|
||
uid, _ = _parse_tea_uid(raw)
|
||
if uid and uid.isdigit():
|
||
return uid
|
||
for key, raw in ls.items():
|
||
if "tea_cache" not in key.lower() or not raw:
|
||
continue
|
||
uid, _ = _parse_tea_uid(str(raw))
|
||
if uid and uid.isdigit():
|
||
return uid
|
||
return ""
|
||
|
||
|
||
def extract_web_id_from_ls(ls: dict) -> str:
|
||
for key in ("tea_cache_tokens", "__tea_cache_tokens_6383", "__tea_cache_tokens_1661", "__tea_cache_tokens_5231", "__tea_cache_tokens_2906"):
|
||
raw = ls.get(key)
|
||
if not raw:
|
||
continue
|
||
_, wid = _parse_tea_uid(raw)
|
||
uid, _ = _parse_tea_uid(raw)
|
||
if wid and wid.isdigit():
|
||
return wid
|
||
if uid and uid.isdigit():
|
||
return uid
|
||
return ""
|
||
|
||
|
||
def extract_creator_uid(ls: dict) -> str:
|
||
"""创作者中心私信页 UID（aid=2906 场景，与 WS device_id 对齐）。"""
|
||
import re
|
||
for key in ("__tea_cache_tokens_1661", "__tea_cache_tokens_2906", "tea_cache_tokens"):
|
||
raw = ls.get(key)
|
||
if not raw:
|
||
continue
|
||
uid, _ = _parse_tea_uid(str(raw))
|
||
if uid and uid.isdigit():
|
||
return uid
|
||
slardar = str(ls.get("SLARDARdouyin_creator") or "")
|
||
m = re.search(r"userId[^0-9]{0,8}(\d{6,})", slardar)
|
||
if m:
|
||
return m.group(1)
|
||
for value in ls.values():
|
||
if not isinstance(value, str):
|
||
continue
|
||
m2 = re.search(r'"uuid"\s*:\s*"(\d{6,})"', value)
|
||
if m2:
|
||
return m2.group(1)
|
||
return ""
|
||
|
||
|
||
def resolve_canonical_uid(www_ls: dict, creator_ls: dict, ws_urls: list) -> tuple[str, str]:
|
||
"""创作者私信 WS（aid=2906）优先：UID 与 WS device_id 对齐。"""
|
||
best_ws = pick_best_ws_url(ws_urls)
|
||
creator_uid = extract_creator_uid(creator_ls)
|
||
www_uid = extract_my_uid_from_ls(www_ls)
|
||
|
||
if best_ws and "aid=2906" in best_ws:
|
||
dev = ws_device_id(best_ws)
|
||
my_uid = dev if dev.isdigit() else (creator_uid or www_uid)
|
||
else:
|
||
my_uid = www_uid or creator_uid
|
||
if not my_uid and best_ws:
|
||
dev = ws_device_id(best_ws)
|
||
if dev.isdigit():
|
||
my_uid = dev
|
||
|
||
web_id = extract_web_id_from_ls(www_ls) or extract_web_id_from_ls(creator_ls)
|
||
return my_uid, web_id
|
||
|
||
|
||
def is_im_ws_url(url: str) -> bool:
|
||
if not url or "token=" not in url:
|
||
return False
|
||
if "frontier-im.douyin.com" in url:
|
||
return True
|
||
if "zijieapi.com" in url and ("/ws/v2" in url or "frontier" in url.lower()):
|
||
return True
|
||
return "/ws/v2" in url and ("douyin" in url or "creator" in url)
|
||
|
||
|
||
def ws_rank(url: str) -> int:
|
||
"""创作者私信页 WS（aid=2906 + sdk_cert）优先级最高。"""
|
||
if not url:
|
||
return 0
|
||
score = 1
|
||
if "sdk_cert=" in url:
|
||
score += 25
|
||
if "aid=2906" in url or "device_platform=douyin_pc" in url:
|
||
score += 20
|
||
if is_real_captured_ws(url):
|
||
score += 5
|
||
if "frontier-im.douyin.com" in url:
|
||
score += 3
|
||
if "aid=6383" in url:
|
||
score -= 5
|
||
return score
|
||
|
||
|
||
def pick_best_ws_url(ws_urls: list) -> str:
|
||
candidates = [u for u in (ws_urls or []) if is_im_ws_url(u)]
|
||
if not candidates:
|
||
return ""
|
||
candidates.sort(key=ws_rank, reverse=True)
|
||
return candidates[0]
|
||
|
||
|
||
def ws_device_id(url: str) -> str:
|
||
return get_query(url, "device_id")
|
||
|
||
|
||
def merge_ws_urls(target: list, incoming: list) -> None:
|
||
for url in incoming or []:
|
||
if not url or not is_im_ws_url(url):
|
||
continue
|
||
if url not in target:
|
||
target.append(url)
|
||
target.sort(key=ws_rank, reverse=True)
|
||
|
||
|
||
def is_real_captured_ws(url: str) -> bool:
|
||
"""真实抓包 WS 的 token 是编码串，不是裸 sessionid。"""
|
||
if not is_im_ws_url(url):
|
||
return False
|
||
token = get_query(url, "token")
|
||
sid_like = len(token) < 40 and token.replace("_", "").replace("-", "").isalnum()
|
||
return not sid_like
|
||
|
||
|
||
def attach_frontier_ws(result: dict, ws_urls: list, my_uid: str, web_id: str) -> str:
|
||
chosen = pick_best_ws_url(ws_urls)
|
||
|
||
if not chosen:
|
||
print(" 未捕获 frontier WS（请在创作者私信页等消息列表加载后重采）")
|
||
return ""
|
||
|
||
if not is_real_captured_ws(chosen) and "sdk_cert=" not in chosen:
|
||
print(" 警告：抓到的 WS 无 sdk_cert，可能无法实时收消息")
|
||
|
||
ws_dev = ws_device_id(chosen)
|
||
if ws_dev and ws_dev.isdigit():
|
||
if "aid=2906" in chosen:
|
||
result["my_uid"] = int(ws_dev)
|
||
my_uid = ws_dev
|
||
elif not my_uid:
|
||
result["my_uid"] = int(ws_dev)
|
||
|
||
result["frontier_ws_url"] = chosen
|
||
result["ws_source"] = "creator_chat" if "aid=2906" in chosen else "douyin_web"
|
||
sdk_cert = get_query(chosen, "sdk_cert")
|
||
ts_sign = get_query(chosen, "ts_sign")
|
||
if sdk_cert:
|
||
result["sdk_cert"] = sdk_cert
|
||
if ts_sign:
|
||
result["ts_sign"] = ts_sign
|
||
print(f" 已选用 WS（aid={'2906 创作者私信' if 'aid=2906' in chosen else '其他'}，device_id={ws_dev or '?'}）")
|
||
return chosen
|
||
|
||
|
||
def normalize_cookie(cookie):
|
||
same_site = cookie.get("sameSite") or "Lax"
|
||
if same_site not in ("Strict", "Lax", "None"):
|
||
same_site = "Lax"
|
||
return {
|
||
"name": cookie["name"],
|
||
"value": cookie.get("value", ""),
|
||
"domain": cookie.get("domain") or ".douyin.com",
|
||
"path": cookie.get("path") or "/",
|
||
"expires": int(cookie.get("expires") or -1),
|
||
"httpOnly": bool(cookie.get("httpOnly", False)),
|
||
"secure": bool(cookie.get("secure", True)),
|
||
"sameSite": same_site,
|
||
}
|
||
|
||
|
||
def add_cookie_if_missing(cookies, name, value):
|
||
if not value or any(c.get("name") == name for c in cookies):
|
||
return
|
||
cookies.append({
|
||
"name": name,
|
||
"value": value,
|
||
"domain": ".douyin.com",
|
||
"path": "/",
|
||
"expires": int(time.time()) + 60 * 60 * 24 * 180,
|
||
"httpOnly": True,
|
||
"secure": True,
|
||
"sameSite": "None",
|
||
})
|
||
|
||
|
||
def complete_session_cookies(cookies):
|
||
values = {c.get("name"): c.get("value", "") for c in cookies}
|
||
sid = values.get("sessionid") or values.get("sessionid_ss") or values.get("sid_tt")
|
||
guard = values.get("sid_guard")
|
||
if not sid and guard:
|
||
try:
|
||
sid = unquote(guard).split("|")[0].strip()
|
||
except Exception:
|
||
sid = guard.split("|")[0].strip()
|
||
add_cookie_if_missing(cookies, "sessionid", sid)
|
||
add_cookie_if_missing(cookies, "sessionid_ss", sid)
|
||
|
||
|
||
async def wait_for_login(context, page):
|
||
STATE.set_waiting("正在打开无痕浏览器…")
|
||
print("\n=== 抖音 IM 凭证采集（无痕窗口）===")
|
||
print("① 切换到「② 抖音登录」标签页扫码")
|
||
print("② 登录后自动采集，回到「① 凭证采集」标签页复制\n")
|
||
|
||
start_event = threading.Event()
|
||
bind_start_event(start_event)
|
||
|
||
logged_in_at = None
|
||
chat_at = None
|
||
navigated_chat = False
|
||
start = time.time()
|
||
|
||
async def on_chat_page() -> bool:
|
||
url = page.url or ""
|
||
return "/following/chat" in url or CHAT_URL in url
|
||
|
||
async def readiness_score() -> int:
|
||
score = 0
|
||
cookies = await context.cookies(["https://www.douyin.com", "https://creator.douyin.com"])
|
||
names = {c.get("name") for c in cookies}
|
||
if "sessionid" in names or "sessionid_ss" in names or "sid_guard" in names:
|
||
score += 10
|
||
if await on_chat_page():
|
||
score += 5
|
||
try:
|
||
ls = await page.evaluate(
|
||
"""
|
||
() => {
|
||
try {
|
||
const wp = localStorage.getItem('security-sdk/s_sdk_sign_data_key/web_protect');
|
||
const crypt = localStorage.getItem('security-sdk/s_sdk_crypt_sdk');
|
||
return { wp: !!wp, crypt: !!crypt };
|
||
} catch (e) { return { wp: false, crypt: false }; }
|
||
}
|
||
"""
|
||
)
|
||
if ls.get("wp"):
|
||
score += 15
|
||
if ls.get("crypt"):
|
||
score += 10
|
||
except Exception:
|
||
pass
|
||
return score
|
||
|
||
while True:
|
||
cookies = await context.cookies(["https://www.douyin.com", "https://creator.douyin.com"])
|
||
names = {c.get("name") for c in cookies}
|
||
logged_in = "sessionid" in names or "sessionid_ss" in names or "sid_guard" in names or "sid_tt" in names
|
||
|
||
if logged_in:
|
||
if logged_in_at is None:
|
||
logged_in_at = time.time()
|
||
print("✓ 已检测到登录")
|
||
if not navigated_chat:
|
||
navigated_chat = True
|
||
STATE.set_waiting("已登录，正在打开私信页…")
|
||
print("正在打开创作者私信页…")
|
||
try:
|
||
await page.goto(CHAT_URL, wait_until="domcontentloaded", timeout=45000)
|
||
await page.wait_for_timeout(2000)
|
||
except Exception:
|
||
pass
|
||
|
||
if await on_chat_page():
|
||
if chat_at is None:
|
||
chat_at = time.time()
|
||
score = await readiness_score()
|
||
waited = time.time() - chat_at
|
||
if score >= 35:
|
||
print("\n凭证已就绪，开始自动采集…")
|
||
STATE.set_collecting("正在采集凭证…")
|
||
return
|
||
if waited >= 18 and score >= 15:
|
||
print("\n开始采集（请确保私信页消息列表已出现）…")
|
||
STATE.set_collecting("正在采集凭证…")
|
||
return
|
||
remain = max(0, 18 - int(waited))
|
||
STATE.set_waiting(
|
||
f"已登录 ✓ 等待私信页…（约 {remain} 秒后自动采集，可点「① 凭证采集」页按钮）"
|
||
)
|
||
else:
|
||
STATE.set_waiting("已登录 ✓ 正在跳转私信页…")
|
||
else:
|
||
STATE.set_waiting("请切换到「② 抖音登录」标签页扫码")
|
||
|
||
if start_event.is_set():
|
||
print("\n收到开始采集信号…")
|
||
STATE.set_collecting("正在采集凭证…")
|
||
return
|
||
|
||
if time.time() - start > 600:
|
||
print("\n等待超时，尝试采集当前状态…")
|
||
STATE.set_collecting("等待超时，尝试采集当前状态…")
|
||
return
|
||
|
||
await asyncio.sleep(0.8)
|
||
|
||
|
||
async def collect(ui_url: str = ""):
|
||
launch_kw = browser_launch_kwargs()
|
||
incognito_ui = use_incognito_ui()
|
||
async with async_playwright() as pw:
|
||
browser = await pw.chromium.launch(
|
||
headless=False,
|
||
args=BROWSER_ARGS,
|
||
**launch_kw,
|
||
)
|
||
signal_browser_ready()
|
||
context = await browser.new_context(viewport={"width": 1280, "height": 900})
|
||
await context.add_init_script(WS_HOOK_INIT)
|
||
ws_urls: list[str] = []
|
||
ui_page = None
|
||
|
||
print("已以无痕模式打开浏览器（关闭后不会保留登录记录）")
|
||
|
||
if ui_url and incognito_ui:
|
||
ui_page = await context.new_page()
|
||
await ui_page.goto(f"{ui_url}?run={STATE.run_id}", wait_until="domcontentloaded")
|
||
try:
|
||
await ui_page.evaluate("document.title = '① 凭证采集'")
|
||
except Exception:
|
||
pass
|
||
append_log("已打开「凭证采集」标签页")
|
||
print(f"凭证页：{ui_url}")
|
||
|
||
page = await context.new_page()
|
||
try:
|
||
await page.evaluate("document.title = '② 抖音登录'")
|
||
except Exception:
|
||
pass
|
||
|
||
def on_websocket(ws):
|
||
url = ws.url
|
||
if is_im_ws_url(url):
|
||
merge_ws_urls(ws_urls, [url])
|
||
print("已捕获 Frontier WS")
|
||
|
||
def bind_page(p):
|
||
p.on("websocket", on_websocket)
|
||
|
||
bind_page(page)
|
||
context.on("page", bind_page)
|
||
|
||
async def harvest_ws(p) -> None:
|
||
try:
|
||
found = await p.evaluate(HARVEST_WS_JS)
|
||
if found:
|
||
merge_ws_urls(ws_urls, found)
|
||
except Exception:
|
||
pass
|
||
|
||
async def wait_for_ws_on_page(p, label: str, wait_ms: int = 60000, reload_at_ms: int = 20000) -> None:
|
||
print(f"正在 {label} 等待 WebSocket 连接…")
|
||
STATE.set_collecting(f"正在 {label} 等待 WebSocket…")
|
||
elapsed = 0
|
||
reloaded = False
|
||
step = 500
|
||
ideal_rank = 46 # aid=2906 + sdk_cert + 真实 token
|
||
min_rank = 28 # aid=2906 + 真实 token（无 sdk_cert 也可先收下）
|
||
while elapsed < wait_ms:
|
||
await harvest_ws(p)
|
||
try:
|
||
await p.evaluate(CHAT_SCROLL_JS)
|
||
except Exception:
|
||
pass
|
||
best = pick_best_ws_url(ws_urls)
|
||
if best:
|
||
r = ws_rank(best)
|
||
if r >= ideal_rank:
|
||
print(f" 已捕获创作者私信 WS（aid=2906，含 sdk_cert）")
|
||
return
|
||
if r >= min_rank and elapsed >= 8000:
|
||
print(f" 已捕获创作者私信 WS（aid=2906，继续等待 sdk_cert… score={r}）")
|
||
if not reloaded and elapsed >= reload_at_ms and (not best or ws_rank(best) < min_rank):
|
||
reloaded = True
|
||
print(" 尚未抓到 WS，刷新私信页重试…")
|
||
try:
|
||
await p.reload(wait_until="domcontentloaded", timeout=30000)
|
||
await p.wait_for_timeout(3000)
|
||
except Exception:
|
||
pass
|
||
await p.wait_for_timeout(step)
|
||
elapsed += step
|
||
best = pick_best_ws_url(ws_urls)
|
||
if best and ws_rank(best) >= min_rank:
|
||
print(f" 使用已抓到的 WS（score={ws_rank(best)}）")
|
||
elif best:
|
||
print(f" 警告：WS 质量偏低（score={ws_rank(best)}），建议私信页刷新后重采")
|
||
|
||
await page.goto(CHAT_URL, wait_until="domcontentloaded")
|
||
try:
|
||
await page.bring_to_front()
|
||
except Exception:
|
||
pass
|
||
append_log("请在本窗口「② 抖音登录」标签页扫码")
|
||
await wait_for_login(context, page)
|
||
|
||
# 用户按 Enter 后：确保在私信页，再抓 WS
|
||
try:
|
||
cur = page.url or ""
|
||
if CHAT_URL not in cur:
|
||
print("正在跳转到创作者私信页…")
|
||
await page.goto(CHAT_URL, wait_until="domcontentloaded", timeout=30000)
|
||
await page.wait_for_timeout(3000)
|
||
except PlaywrightTimeoutError:
|
||
pass
|
||
await wait_for_ws_on_page(page, "创作者私信页")
|
||
|
||
try:
|
||
if CHAT_URL not in (page.url or ""):
|
||
await page.goto(CHAT_URL, wait_until="domcontentloaded", timeout=30000)
|
||
await page.wait_for_timeout(2000)
|
||
except Exception:
|
||
pass
|
||
await harvest_ws(page)
|
||
|
||
print("正在读取 localStorage / Cookie…")
|
||
STATE.set_collecting("正在读取签名凭证…")
|
||
|
||
async def read_local_storage_inline():
|
||
try:
|
||
return await page.evaluate("""
|
||
() => {
|
||
const out = {};
|
||
for (let i = 0; i < localStorage.length; i++) {
|
||
const k = localStorage.key(i);
|
||
out[k] = localStorage.getItem(k);
|
||
}
|
||
return out;
|
||
}
|
||
""")
|
||
except Exception:
|
||
return {}
|
||
|
||
async def read_local_storage(target_url):
|
||
try:
|
||
await page.goto(target_url, wait_until="domcontentloaded", timeout=30000)
|
||
await page.wait_for_timeout(1500)
|
||
return await read_local_storage_inline()
|
||
except Exception:
|
||
return {}
|
||
|
||
creator_ls = await read_local_storage_inline()
|
||
if not creator_ls:
|
||
creator_ls = await read_local_storage(CHAT_URL)
|
||
www_ls = await read_local_storage("https://www.douyin.com/")
|
||
ls = {**creator_ls, **www_ls}
|
||
|
||
cookies = await context.cookies(["https://www.douyin.com", "https://creator.douyin.com"])
|
||
cookies = [normalize_cookie(c) for c in cookies if ".douyin.com" in c.get("domain", "") or "douyin.com" in c.get("domain", "")]
|
||
complete_session_cookies(cookies)
|
||
|
||
web_protect_raw = pick_local_storage(ls, "security-sdk/s_sdk_sign_data_key/web_protect", ["web_protect"])
|
||
if not web_protect_raw:
|
||
token_raw = find_security_value(ls, ["ticket", "ts_sign", "client_cert"], ["sign_data_key", "token"])
|
||
web_protect_raw = build_web_protect_from_token(token_raw) or token_raw
|
||
crypt_raw = pick_local_storage(ls, "security-sdk/s_sdk_crypt_sdk", ["crypt_sdk"])
|
||
if not crypt_raw:
|
||
crypt_raw = find_security_value(ls, ["ec_privateKey", "ec_publicKey"], ["crypt", "sdk"])
|
||
tea_raw = normalize_tea_key(www_ls) or normalize_tea_key(creator_ls)
|
||
my_uid, web_id = resolve_canonical_uid(www_ls, creator_ls, ws_urls)
|
||
device_id = my_uid or web_id
|
||
|
||
local_storage_entries = []
|
||
if crypt_raw:
|
||
local_storage_entries.append({"name": "security-sdk/s_sdk_crypt_sdk", "value": str(crypt_raw)})
|
||
if web_protect_raw:
|
||
local_storage_entries.append({"name": "security-sdk/s_sdk_sign_data_key/web_protect", "value": str(web_protect_raw)})
|
||
if tea_raw:
|
||
local_storage_entries.append({"name": "tea_cache_tokens", "value": str(tea_raw)})
|
||
elif my_uid:
|
||
local_storage_entries.append({
|
||
"name": "tea_cache_tokens",
|
||
"value": json.dumps({"user_unique_id": my_uid, "web_id": web_id or my_uid}, ensure_ascii=False),
|
||
})
|
||
if device_id and str(device_id).isdigit():
|
||
local_storage_entries.append({"name": "web_runtime_security_uid", "value": str(device_id)})
|
||
|
||
user_agent = await page.evaluate("navigator.userAgent")
|
||
origins = [{"origin": "https://www.douyin.com", "localStorage": local_storage_entries}]
|
||
if creator_ls:
|
||
creator_only = []
|
||
for key, value in creator_ls.items():
|
||
if key in ls and ls[key] == value and not any(e["name"] == key for e in local_storage_entries):
|
||
creator_only.append({"name": key, "value": str(value)})
|
||
if creator_only:
|
||
origins.append({"origin": "https://creator.douyin.com", "localStorage": creator_only})
|
||
|
||
result = {
|
||
"cookies": cookies,
|
||
"origins": origins,
|
||
"user_agent": user_agent,
|
||
}
|
||
|
||
if my_uid:
|
||
result["my_uid"] = int(my_uid)
|
||
elif web_id and web_id.isdigit():
|
||
result["my_uid"] = int(web_id)
|
||
|
||
attach_frontier_ws(result, ws_urls, my_uid, web_id)
|
||
my_uid = str(result.get("my_uid") or my_uid or "")
|
||
|
||
wp = parse_web_protect(web_protect_raw)
|
||
names = {c["name"] for c in cookies}
|
||
ws_url = result.get("frontier_ws_url") or ""
|
||
ws_captured = bool(ws_url and is_real_captured_ws(ws_url))
|
||
ws_has_cert = bool(result.get("sdk_cert"))
|
||
print("\n采集检查：")
|
||
print(" sessionid:", "OK" if ("sessionid" in names or "sessionid_ss" in names) else "缺失")
|
||
print(" web_protect:", "OK" if web_protect_raw else "缺失")
|
||
print(" crypt_sdk:", "OK" if crypt_raw else "缺失")
|
||
print(" web_protect 完整:", "OK" if (wp.get("ticket") and wp.get("ts_sign") and (wp.get("client_cert") or wp.get("sdk_cert"))) else "缺失")
|
||
if ws_captured and ws_has_cert:
|
||
aid_hint = "2906 创作者私信" if "aid=2906" in ws_url else "其他"
|
||
print(f" frontier_ws: OK（真实抓包，含 sdk_cert，{aid_hint}）")
|
||
elif ws_captured:
|
||
print(" frontier_ws:", "OK（真实抓包，无 sdk_cert）")
|
||
elif ws_url:
|
||
print(" frontier_ws:", "警告：WS 疑似无效，请重采")
|
||
else:
|
||
print(" frontier_ws:", "缺失（请在私信页等消息列表加载后重采）")
|
||
print(" my_uid:", my_uid or web_id or "缺失")
|
||
if my_uid and web_id and my_uid != web_id:
|
||
print(" web_id:", web_id, "(与 my_uid 不同，属正常)")
|
||
|
||
OUT_FILE.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
publish_result(result)
|
||
print(f"\n已生成：{OUT_FILE}")
|
||
if ui_page:
|
||
try:
|
||
await ui_page.bring_to_front()
|
||
await ui_page.evaluate("document.title = '① 凭证采集 · 已完成'")
|
||
except Exception:
|
||
pass
|
||
append_log("采集完成，请在本窗口「① 凭证采集」标签页复制")
|
||
print("请在本窗口「凭证采集」标签页复制 JSON。")
|
||
else:
|
||
print("已在浏览器打开采集结果页，可一键复制凭证 JSON。")
|
||
|
||
if incognito_ui and ui_page:
|
||
loop = asyncio.get_running_loop()
|
||
await loop.run_in_executor(None, wait_for_shutdown)
|
||
|
||
await context.close()
|
||
await browser.close()
|
||
|
||
|
||
def main() -> None:
|
||
gui = is_gui_mode()
|
||
if gui:
|
||
_install_ui_logging()
|
||
|
||
try:
|
||
clear_browser_ready()
|
||
_, ui_url = start_server()
|
||
if not use_incognito_ui():
|
||
if os.environ.get("SPLASH_LAUNCHED") != "1":
|
||
launch_splash()
|
||
open_browser(f"{ui_url}?run={STATE.run_id}")
|
||
append_log("请在弹出的 Chrome 扫码登录")
|
||
else:
|
||
append_log("正在打开无痕浏览器…")
|
||
asyncio.run(collect(ui_url))
|
||
except KeyboardInterrupt:
|
||
print("\n已取消")
|
||
publish_error("用户取消采集")
|
||
if not gui:
|
||
input("\n按回车退出…")
|
||
return
|
||
except Exception as exc:
|
||
msg = str(exc)
|
||
if "Executable doesn't exist" in msg or "playwright install" in msg.lower():
|
||
from app_paths import is_frozen
|
||
|
||
if is_frozen():
|
||
fix_hint = "请运行同目录「修复浏览器.bat」，或 DouyinIMCollector.exe --install-browser"
|
||
else:
|
||
fix_hint = f"请运行：{app_root() / 'fix_browser.bat'}"
|
||
msg = f"Playwright 找不到浏览器。\n\n{fix_hint}"
|
||
print(f"\n采集失败：{msg}")
|
||
publish_error(msg)
|
||
if gui:
|
||
wait_for_shutdown()
|
||
else:
|
||
input("\n按回车退出…")
|
||
return
|
||
|
||
if gui and not use_incognito_ui():
|
||
append_log("采集流程结束，可复制凭证后点「退出」")
|
||
wait_for_shutdown()
|
||
elif gui and use_incognito_ui():
|
||
pass
|
||
else:
|
||
phase = STATE.snapshot().get("phase")
|
||
if phase == "done":
|
||
print("\n✓ 采集完成！请在浏览器页面复制凭证后关闭本窗口。")
|
||
input("\n按回车关闭…")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|