ksjsb/tools/captcha_capture.py

"""抓 captcha 协议的 config 响应体 + bgPic/cutPic 图片 + captchaSession（无需拖动）。

config 在 iframe 加载时自动 POST，所以不开滑块也能拿到 schema。
落盘 out/captcha_cap/: config.json, bgPic.jpg, cutPic.png, session.txt, full_timeline.json
"""
from __future__ import annotations
import argparse, json, re, subprocess, sys, time, shutil, tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))
OUT = ROOT / "out" / "captcha_cap"


def fresh_error_url(mobile, profile, base):
    cmd = [sys.executable, "-m", "tools.sms_login_cli", "--mobile", mobile,
           "--code", "000000", "--base-url", base,
           "--transport", "okhttp4-android10"]
    print("[cap] 取新鲜 error_url ...")
    p = subprocess.run(cmd, capture_output=True, text=True, cwd=str(ROOT))
    out = p.stdout + p.stderr
    m = re.search(r'https://app\.m\.kuaishou\.com/verify/captcha\.html\?[^"\s\\]+', out)
    if not m:
        print(out[-1500:]); raise SystemExit("[cap] 未提取到 error_url")
    return m.group(0)


def capture(error_url, wait):
    from core.captcha_assist import _find_system_chromium, _free_local_port, _wait_for_cdp_endpoint
    from playwright.sync_api import sync_playwright
    import subprocess as sp
    OUT.mkdir(parents=True, exist_ok=True)
    browser = _find_system_chromium(); port = _free_local_port()
    prof = tempfile.mkdtemp(prefix="ksjsb-cap-")
    proc = sp.Popen([browser, f"--remote-debugging-port={port}",
                     "--remote-debugging-address=127.0.0.1", f"--user-data-dir={prof}",
                     "--no-first-run", "--no-default-browser-check",
                     "--window-size=430,920", "about:blank"], stdout=sp.DEVNULL, stderr=sp.DEVNULL)
    blobs = {}
    timeline = []
    try:
        ep = _wait_for_cdp_endpoint(port, timeout=20)
        with sync_playwright() as pw:
            b = pw.chromium.connect_over_cdp(ep, timeout=20000)
            page = b.contexts[0].pages[0] if b.contexts[0].pages else b.contexts[0].new_page()

            def on_response(resp):
                url = str(resp.url)
                # 调试: 记录所有非静态资源的响应
                if not re.search(r'\.(js|css|png|jpg|gif|woff|ico)(\?|$)', url) and "log/collect" not in url and "radar" not in url:
                    timeline.append({"url": url[:200], "status": resp.status, "ct": (resp.headers or {}).get("content-type", "")})
                if any(k in url for k in ("/sliding/config", "/sliding/bgPic", "/sliding/cutPic")):
                    ct = (resp.headers or {}).get("content-type", "")
                    timeline.append({"url": url, "status": resp.status, "ct": ct})
                    try:
                        body = resp.body()
                    except Exception as e:
                        timeline[-1]["err"] = str(e); return
                    if "config" in url:
                        try:
                            data = json.loads(body)
                        except Exception:
                            data = {"_raw": body.decode("utf-8", "replace")[:2000]}
                        (OUT / "config.json").write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
                        blobs["config"] = True
                        print(f"[cap] config JSON ({len(body)}B) -> keys: {list(data.keys()) if isinstance(data, dict) else type(data)}")
                    elif "bgPic" in url:
                        (OUT / "bgPic.jpg").write_bytes(body); blobs["bgPic"] = len(body)
                        print(f"[cap] bgPic {len(body)}B")
                    elif "cutPic" in url:
                        (OUT / "cutPic.png").write_bytes(body); blobs["cutPic"] = len(body)
                        print(f"[cap] cutPic {len(body)}B")

            page.on("response", on_response)
            print(f"[cap] goto {error_url[:80]}...")
            page.goto(error_url, wait_until="domcontentloaded", timeout=30000)
            # iframe 里的 config 请求要等 iframe 加载完
            for _ in range(wait * 2):
                if blobs.get("config") and blobs.get("bgPic") and blobs.get("cutPic"):
                    break
                time.sleep(0.5)
            try:
                page.screenshot(path=str(OUT / "page.png"), full_page=True)
            except Exception as e:
                print("[cap] 截图失败", e)
            # 抓 captchaSession（从 iframe url 或 config 请求体）
            try:
                for fr in page.frames:
                    if "captchaSession=" in fr.url:
                        cs = re.search(r'captchaSession=([^&]+)', fr.url)
                        if cs: (OUT / "session.txt").write_text(cs.group(1), encoding="utf-8"); print("[cap] session saved")
            except Exception as e:
                print("[cap] session 抓取失败", e)
    finally:
        if proc.poll() is None:
            proc.terminate()
            try: proc.wait(timeout=3)
            except sp.TimeoutExpired: proc.kill()
        shutil.rmtree(prof, ignore_errors=True)
    (OUT / "timeline.json").write_text(json.dumps(timeline, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"[cap] done. blobs={blobs}  dir={OUT}")
    return blobs


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--mobile", default="17666663175")
    ap.add_argument("--url", default="")
    ap.add_argument("--device-profile", default="out/sms_device_latest.json")
    ap.add_argument("--base-url", default="https://az1-api.ksapisrv.com")
    ap.add_argument("--wait", type=int, default=10)
    a = ap.parse_args()
    url = a.url or fresh_error_url(a.mobile, a.device_profile, a.base_url)
    print("[cap] error_url =", url[:90])
    capture(url, a.wait)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())