cmshoppe/app/cdp.py

#!/usr/bin/env python3
# app/cdp.py —— 共享 CDP 底座：连接 Chrome、定位/打开商品 tab、执行 JS、模拟拖拽。
# 被正式模块和 prototypes/demo.py 复用。
#
# 依赖：pip install websocket-client requests

import os

# CDP 局域网直连，绝不走代理（WSL 里 *_proxy 指向 :1080 会导致超时）。
for _k in ("ALL_PROXY", "all_proxy", "HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"):
    os.environ.pop(_k, None)

import json
import time
import threading

CDP_HOST = os.environ.get("CDP_HOST", "127.0.0.1:9222")
BASE = f"http://{CDP_HOST}"


def _base(host=None):
    return f"http://{host or CDP_HOST}"


def http_get(path, host=None):
    import requests

    s = requests.Session()
    s.trust_env = False  # 忽略环境代理
    return s.get(f"{_base(host)}{path}", timeout=10).json()


def close_tab(target_id, host=None):
    """Close a Chrome target/page by target id."""

    if not target_id:
        return False
    import requests

    s = requests.Session()
    s.trust_env = False
    response = s.get(f"{_base(host)}/json/close/{target_id}", timeout=10)
    return response.ok


def activate_tab(target_id, host=None):
    """Bring an existing Chrome target/page to the foreground."""

    if not target_id:
        return False
    ver = http_get("/json/version", host=host)
    b = CDP(ver["webSocketDebuggerUrl"])
    try:
        b.send("Target.activateTarget", {"targetId": target_id})
    finally:
        b.close()
    return True


class CDP:
    """单个 target 的 CDP 客户端：命令同步、事件回调异步。"""

    def __init__(self, ws_url):
        from websocket import create_connection

        # suppress_origin=True：不发 Origin 头，避免 Chrome 未开 --remote-allow-origins 时 403。
        self.ws = create_connection(ws_url, max_size=None, suppress_origin=True)
        self._id = 0
        self._res = {}
        self._cond = threading.Condition()
        self._handlers = []
        self._stop = False
        threading.Thread(target=self._loop, daemon=True).start()

    def _loop(self):
        while not self._stop:
            try:
                raw = self.ws.recv()
            except Exception:
                break
            if not raw:
                continue
            m = json.loads(raw)
            if "id" in m:
                with self._cond:
                    self._res[m["id"]] = m
                    self._cond.notify_all()
            elif "method" in m:
                for h in self._handlers:
                    try:
                        h(m)
                    except Exception:
                        pass

    def on(self, fn):
        self._handlers.append(fn)

    def send(self, method, params=None, timeout=30):
        self._id += 1
        mid = self._id
        self.ws.send(json.dumps({"id": mid, "method": method, "params": params or {}}))
        end = time.time() + timeout
        with self._cond:
            while mid not in self._res:
                left = end - time.time()
                if left <= 0:
                    raise TimeoutError(f"CDP {method} 超时")
                self._cond.wait(left)
            m = self._res.pop(mid)
        if "error" in m:
            raise RuntimeError(f"{method}: {m['error'].get('message')}")
        return m.get("result", {})

    def ev(self, expr, by_value=True):
        r = self.send("Runtime.evaluate", {"expression": expr, "returnByValue": by_value})
        if "exceptionDetails" in r:
            raise RuntimeError("JS 异常: " + r["exceptionDetails"].get("text", ""))
        return r.get("result", {})

    def val(self, expr):
        return self.ev(expr).get("value")

    def object_id(self, expr):
        return self.ev(expr, by_value=False).get("objectId")

    def drag(self, x0, y0, x1, y1, steps=34):
        """按住源点 → 多步移动到目标点 → 释放，驱动自定义拖拽排序库。"""
        me = "Input.dispatchMouseEvent"
        self.send(me, {"type": "mouseMoved", "x": x0, "y": y0})
        self.send(me, {"type": "mousePressed", "x": x0, "y": y0, "button": "left", "buttons": 1, "clickCount": 1})
        for dx in (3, 6, 10):  # 小幅抖动，越过拖拽启动阈值
            self.send(me, {"type": "mouseMoved", "x": x0 + dx, "y": y0, "button": "left", "buttons": 1})
            time.sleep(0.03)
        for i in range(1, steps + 1):
            x = x0 + (x1 - x0) * i / steps
            y = y0 + (y1 - y0) * i / steps
            self.send(me, {"type": "mouseMoved", "x": x, "y": y, "button": "left", "buttons": 1})
            time.sleep(0.025)
        for _ in range(10):  # 在目标点停留，让排序库吸附
            self.send(me, {"type": "mouseMoved", "x": x1, "y": y1, "button": "left", "buttons": 1})
            time.sleep(0.05)
        self.send(me, {"type": "mouseReleased", "x": x1, "y": y1, "button": "left", "buttons": 0, "clickCount": 1})

    def close(self):
        self._stop = True
        try:
            self.ws.close()
        except Exception:
            pass


def find_product_tab(item_id, host=None):
    """在已打开的 tab 里找 URL 同时含 item_id 和 shopee.tw 的页面。"""
    for t in http_get("/json", host=host):
        if t.get("type") != "page":
            continue
        url = t.get("url") or ""
        if item_id in url and "shopee.tw" in url:
            return t
    return None


def create_tab_info(url, host=None, background=False):
    """用 browser 级 Target.createTarget 新建 tab，返回其 page target 信息。"""
    ver = http_get("/json/version", host=host)
    b = CDP(ver["webSocketDebuggerUrl"])
    try:
        params = {"url": url}
        if background:
            params["background"] = True
        try:
            tid = b.send("Target.createTarget", params)["targetId"]
        except RuntimeError:
            if not background:
                raise
            tid = b.send("Target.createTarget", {"url": url})["targetId"]
    finally:
        b.close()
    end = time.time() + 15
    while time.time() < end:
        for t in http_get("/json", host=host):
            if t.get("id") == tid and t.get("webSocketDebuggerUrl"):
                return t
        time.sleep(0.5)
    raise TimeoutError("等待新 tab websocket 超时")


def create_tab(url, host=None):
    """用 browser 级 Target.createTarget 新建 tab，返回其 page websocket。"""

    return create_tab_info(url, host=host)["webSocketDebuggerUrl"]