398 lines
14 KiB
Python
398 lines
14 KiB
Python
"""抖音一键采集器 · 本地图形化页面服务（stdlib，无额外依赖）"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
from app_paths import ui_html_path
|
||
|
||
UI_FILE = ui_html_path()
|
||
DEFAULT_PORT = 8765
|
||
|
||
|
||
class CollectorState:
|
||
def __init__(self) -> None:
|
||
self._lock = threading.Lock()
|
||
self.reset()
|
||
|
||
def reset(self) -> None:
|
||
with self._lock:
|
||
self.phase = "waiting"
|
||
self.message = "正在启动…"
|
||
self.result = None
|
||
self.checks: list[dict[str, str]] = []
|
||
self.result_text = ""
|
||
self.my_uid = ""
|
||
self.ok_core = False
|
||
self.summary = ""
|
||
self.run_id = uuid.uuid4().hex[:8]
|
||
self.started_at = int(time.time())
|
||
self.port = DEFAULT_PORT
|
||
self.step = 1
|
||
self.logs: list[str] = []
|
||
|
||
def append_log(self, message: str) -> None:
|
||
line = (message or "").strip()
|
||
if not line:
|
||
return
|
||
with self._lock:
|
||
self.logs.append(line)
|
||
if len(self.logs) > 80:
|
||
self.logs = self.logs[-80:]
|
||
|
||
def set_step(self, step: int) -> None:
|
||
with self._lock:
|
||
self.step = max(1, min(4, step))
|
||
|
||
def set_waiting(self, message: str = "") -> None:
|
||
with self._lock:
|
||
self.phase = "waiting"
|
||
if message:
|
||
self.message = message
|
||
if self.step < 2:
|
||
self.step = 2
|
||
|
||
def set_collecting(self, message: str = "") -> None:
|
||
with self._lock:
|
||
self.phase = "collecting"
|
||
self.message = message or "正在采集 Cookie / 签名 / WebSocket…"
|
||
self.step = 3
|
||
|
||
def set_done(self, result: dict, checks: list[dict[str, str]], ok_core: bool, summary: str) -> None:
|
||
with self._lock:
|
||
self.phase = "done"
|
||
self.result = result
|
||
self.checks = checks
|
||
self.result_text = json.dumps(result, ensure_ascii=False, indent=2)
|
||
self.my_uid = str(result.get("my_uid") or "")
|
||
self.ok_core = ok_core
|
||
self.summary = summary
|
||
self.message = "采集完成"
|
||
self.step = 4
|
||
|
||
def set_error(self, message: str) -> None:
|
||
with self._lock:
|
||
self.phase = "error"
|
||
self.message = message
|
||
|
||
def snapshot(self) -> dict[str, Any]:
|
||
with self._lock:
|
||
return {
|
||
"phase": self.phase,
|
||
"message": self.message,
|
||
"checks": self.checks,
|
||
"result_text": self.result_text,
|
||
"my_uid": self.my_uid,
|
||
"ok_core": self.ok_core,
|
||
"summary": self.summary,
|
||
"run_id": self.run_id,
|
||
"started_at": self.started_at,
|
||
"port": self.port,
|
||
"step": self.step,
|
||
"logs": list(self.logs),
|
||
}
|
||
|
||
|
||
def append_log(message: str) -> None:
|
||
STATE.append_log(message)
|
||
|
||
|
||
STATE = CollectorState()
|
||
_SERVER: HTTPServer | None = None
|
||
_start_event: threading.Event | None = None
|
||
_shutdown_event = threading.Event()
|
||
|
||
|
||
def bind_start_event(event: threading.Event) -> None:
|
||
"""采集主流程注册：网页「开始采集」或自动就绪时触发。"""
|
||
global _start_event
|
||
_start_event = event
|
||
|
||
|
||
def request_start_collect() -> bool:
|
||
if _start_event is None:
|
||
return False
|
||
_start_event.set()
|
||
return True
|
||
|
||
|
||
def is_start_requested() -> bool:
|
||
return bool(_start_event and _start_event.is_set())
|
||
|
||
|
||
def request_shutdown() -> None:
|
||
_shutdown_event.set()
|
||
srv = _SERVER
|
||
if srv:
|
||
threading.Thread(target=srv.shutdown, daemon=True).start()
|
||
|
||
|
||
def wait_for_shutdown() -> None:
|
||
_shutdown_event.wait()
|
||
|
||
|
||
def _free_port(port: int) -> None:
|
||
"""关闭占用端口的旧采集器进程（避免网页显示上一次失败的错误）。"""
|
||
if sys.platform != "win32":
|
||
return
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
"powershell",
|
||
"-NoProfile",
|
||
"-ExecutionPolicy",
|
||
"Bypass",
|
||
"-Command",
|
||
(
|
||
f"Get-NetTCPConnection -LocalPort {port} -State Listen "
|
||
f"-ErrorAction SilentlyContinue | ForEach-Object {{ "
|
||
f"Stop-Process -Id $_.OwningProcess -Force -ErrorAction SilentlyContinue }}"
|
||
),
|
||
],
|
||
capture_output=True,
|
||
timeout=10,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _parse_web_protect(raw: str) -> dict:
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
obj = json.loads(str(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 _pick_ls(result: dict, exact: str, contains: tuple[str, ...] = ()) -> str:
|
||
for origin in result.get("origins") or []:
|
||
for entry in origin.get("localStorage") or []:
|
||
name = entry.get("name") or ""
|
||
val = entry.get("value") or ""
|
||
if name == exact and val:
|
||
return str(val)
|
||
lower = name.lower()
|
||
if val and contains and any(c in lower for c in contains):
|
||
return str(val)
|
||
return ""
|
||
|
||
|
||
def _extract_my_uid(result: dict) -> str:
|
||
uid = result.get("my_uid")
|
||
if uid and str(uid).isdigit():
|
||
return str(uid)
|
||
ws_url = str(result.get("frontier_ws_url") or "")
|
||
if "aid=2906" in ws_url:
|
||
try:
|
||
from urllib.parse import parse_qs, urlparse
|
||
dev = parse_qs(urlparse(ws_url).query).get("device_id", [""])[0]
|
||
if dev and str(dev).isdigit():
|
||
return str(dev)
|
||
except Exception:
|
||
pass
|
||
tea = _pick_ls(result, "tea_cache_tokens", ("tea_cache_tokens",))
|
||
if tea:
|
||
try:
|
||
uid = json.loads(tea).get("user_unique_id")
|
||
if uid and str(uid).isdigit():
|
||
return str(uid)
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def normalize_result(result: dict) -> dict:
|
||
try:
|
||
from credential_normalize import normalize_storage_state_for_im
|
||
return normalize_storage_state_for_im(result)
|
||
except Exception:
|
||
return result
|
||
|
||
|
||
def build_checks(result: dict) -> tuple[list[dict[str, str]], bool, str]:
|
||
result = normalize_result(result)
|
||
cookies = result.get("cookies") or []
|
||
names = {str(c.get("name", "")).lower() for c in cookies}
|
||
val_protect = _pick_ls(result, "security-sdk/s_sdk_sign_data_key/web_protect", ("web_protect",))
|
||
val_keys = _pick_ls(result, "security-sdk/s_sdk_crypt_sdk", ("crypt_sdk",))
|
||
my_uid = _extract_my_uid(result)
|
||
wp = _parse_web_protect(val_protect)
|
||
|
||
has_session = "sessionid" in names or "sessionid_ss" in names
|
||
has_protect = bool(val_protect)
|
||
has_keys = bool(val_keys)
|
||
has_uid = bool(my_uid)
|
||
has_wp_full = bool(wp.get("ticket") and wp.get("ts_sign") and (wp.get("client_cert") or wp.get("sdk_cert")))
|
||
has_ws = bool(result.get("frontier_ws_url"))
|
||
has_sdk_cert = bool(result.get("sdk_cert"))
|
||
token = ""
|
||
ws_url = str(result.get("frontier_ws_url") or "")
|
||
if ws_url:
|
||
try:
|
||
from urllib.parse import parse_qs, urlparse
|
||
token = parse_qs(urlparse(ws_url).query).get("token", [""])[0]
|
||
except Exception:
|
||
pass
|
||
ws_real = has_ws and (has_sdk_cert or len(token) >= 40)
|
||
|
||
def chk(ok: bool, ok_t: str, fail_t: str, critical: bool = True) -> dict[str, str]:
|
||
if ok:
|
||
return {"s": "ok", "t": "已包含 " + ok_t}
|
||
return {"s": "err" if critical else "warn", "t": fail_t}
|
||
|
||
checks = [
|
||
chk(has_session, "sessionid（登录态）", "缺 sessionid：请重新登录并在私信页停留后采集", True),
|
||
chk(has_uid, f"抖音 UID（{my_uid}）" if my_uid else "抖音 UID", "未提取到数字 UID", False),
|
||
chk(has_protect, "web_protect 签名", "缺 web_protect：请在私信页停留 5～10 秒后重采", True),
|
||
]
|
||
if has_protect:
|
||
checks.append(chk(has_wp_full, "web_protect 完整字段", "web_protect 不完整（缺 ticket/ts_sign/cert）", True))
|
||
checks.append(chk(has_keys, "crypt_sdk 密钥", "缺 crypt_sdk：请确认在已登录页面采集", True))
|
||
if has_ws and has_sdk_cert:
|
||
aid = "2906 创作者私信" if "aid=2906" in ws_url else "真实抓包"
|
||
checks.append({"s": "ok", "t": f"frontier WebSocket（{aid}，含 sdk_cert）"})
|
||
elif ws_real:
|
||
checks.append({"s": "ok", "t": "frontier WebSocket（真实抓包）"})
|
||
elif has_ws:
|
||
checks.append({"s": "warn", "t": "WS 无效（token 异常），请重新在私信页采集"})
|
||
elif has_session:
|
||
checks.append({"s": "warn", "t": "未捕获 WS：请在私信页等消息列表加载后重采"})
|
||
else:
|
||
checks.append({"s": "warn", "t": "未捕获 WS，请先完成登录再采集"})
|
||
|
||
ok_core = has_session and has_protect and has_keys and has_wp_full and has_uid
|
||
if ok_core:
|
||
summary = f"共 {len(cookies)} 条 Cookie，可直接复制导入系统。"
|
||
elif has_session and has_protect and has_keys:
|
||
summary = "核心凭证已有，建议核对红色项后导入；若 IM 异常请重新采集。"
|
||
else:
|
||
summary = "关键凭证缺失，请按提示重新登录采集，不要直接导入。"
|
||
|
||
return checks, ok_core, summary
|
||
|
||
|
||
class _Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, format: str, *args) -> None: # noqa: A003
|
||
pass
|
||
|
||
def _send(self, code: int, body: bytes, content_type: str, *, cors: bool = False) -> None:
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Cache-Control", "no-store")
|
||
if cors:
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_OPTIONS(self) -> None: # noqa: N802
|
||
self.send_response(204)
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.end_headers()
|
||
|
||
def do_GET(self) -> None: # noqa: N802
|
||
path = urlparse(self.path).path
|
||
if path in ("/", "/index.html"):
|
||
if UI_FILE.is_file():
|
||
self._send(200, UI_FILE.read_bytes(), "text/html; charset=utf-8")
|
||
else:
|
||
self._send(404, b"UI not found", "text/plain")
|
||
return
|
||
if path == "/api/status":
|
||
payload = json.dumps(STATE.snapshot(), ensure_ascii=False).encode("utf-8")
|
||
self._send(200, payload, "application/json; charset=utf-8", cors=True)
|
||
return
|
||
if path == "/api/ping":
|
||
self._send(200, b"ok", "text/plain; charset=utf-8", cors=True)
|
||
return
|
||
self._send(404, b"Not Found", "text/plain")
|
||
|
||
def do_POST(self) -> None: # noqa: N802
|
||
path = urlparse(self.path).path
|
||
if path == "/api/start":
|
||
ok = request_start_collect()
|
||
if ok:
|
||
STATE.set_collecting("收到开始信号，正在采集…")
|
||
append_log("用户点击：立即采集")
|
||
body = json.dumps({"ok": ok}, ensure_ascii=False).encode("utf-8")
|
||
self._send(200 if ok else 409, body, "application/json; charset=utf-8", cors=True)
|
||
return
|
||
if path == "/api/quit":
|
||
append_log("用户退出采集器")
|
||
request_shutdown()
|
||
body = json.dumps({"ok": True}, ensure_ascii=False).encode("utf-8")
|
||
self._send(200, body, "application/json; charset=utf-8", cors=True)
|
||
return
|
||
self._send(404, b"Not Found", "text/plain")
|
||
|
||
|
||
def start_server(port: int = DEFAULT_PORT) -> tuple[HTTPServer, str]:
|
||
global _SERVER
|
||
STATE.reset()
|
||
STATE.set_step(1)
|
||
append_log("正在启动本地界面…")
|
||
for attempt in range(10):
|
||
p = port + attempt
|
||
_free_port(p)
|
||
try:
|
||
srv = HTTPServer(("127.0.0.1", p), _Handler)
|
||
_SERVER = srv
|
||
STATE.port = p
|
||
url = f"http://127.0.0.1:{p}/"
|
||
thread = threading.Thread(target=srv.serve_forever, daemon=True)
|
||
thread.start()
|
||
return srv, url
|
||
except OSError:
|
||
continue
|
||
raise RuntimeError("无法启动本地页面服务（端口被占用）")
|
||
|
||
|
||
def open_browser(url: str) -> None:
|
||
try:
|
||
if sys.platform == "win32":
|
||
subprocess.Popen(
|
||
["cmd", "/c", "start", "", url],
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
close_fds=True,
|
||
)
|
||
return
|
||
except Exception:
|
||
pass
|
||
try:
|
||
import webbrowser
|
||
webbrowser.open(url)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def publish_result(result: dict) -> None:
|
||
normalized = normalize_result(result)
|
||
checks, ok_core, summary = build_checks(normalized)
|
||
STATE.set_done(normalized, checks, ok_core, summary)
|
||
|
||
|
||
def publish_error(message: str) -> None:
|
||
STATE.set_error(message)
|