chore: 完成正式代码包结构
创建 app 包与 main.py 入口,将根目录 cdp.py 迁入 app/cdp.py,并保留最小 app.gui 占位入口。 修正 prototypes/demo.py 与 demo.bat,让原型脚本从项目根导入 app.cdp。 更新任务看板、当前状态和进度记录,标记 T-000 完成并记录本机 Python 解释器差异。
This commit is contained in:
+146
@@ -0,0 +1,146 @@
|
||||
#!/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 http_get(path):
|
||||
import requests
|
||||
|
||||
s = requests.Session()
|
||||
s.trust_env = False # 忽略环境代理
|
||||
return s.get(f"{BASE}{path}", timeout=10).json()
|
||||
|
||||
|
||||
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):
|
||||
"""在已打开的 tab 里找 URL 同时含 item_id 和 shopee.tw 的页面。"""
|
||||
for t in http_get("/json"):
|
||||
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(url):
|
||||
"""用 browser 级 Target.createTarget 新建 tab,返回其 page websocket。"""
|
||||
ver = http_get("/json/version")
|
||||
b = CDP(ver["webSocketDebuggerUrl"])
|
||||
try:
|
||||
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"):
|
||||
if t.get("id") == tid and t.get("webSocketDebuggerUrl"):
|
||||
return t["webSocketDebuggerUrl"]
|
||||
time.sleep(0.5)
|
||||
raise TimeoutError("等待新 tab websocket 超时")
|
||||
Reference in New Issue
Block a user