Wait for hydrated ChatGPT share messages

2026-07-08 02:25:33 +00:00
parent 11229e911f
commit 202c7a4466
2 changed files with 162 additions and 2 deletions
@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+import base64
+import hashlib
+import json
+import os
+import socket
+import struct
+import sys
+import urllib.parse
+import urllib.request
+
+
+CDP_ENDPOINT = os.environ.get("CDP_ENDPOINT", "http://127.0.0.1:9222")
+
+
+def get_page_ws_url():
+    with urllib.request.urlopen(f"{CDP_ENDPOINT}/json/list", timeout=30) as response:
+        targets = json.loads(response.read().decode("utf-8"))
+    pages = [target for target in targets if target.get("type") == "page"]
+    if not pages:
+        raise SystemExit("No CDP page target found.")
+    selected = next((page for page in pages if "chatgpt.com/share/" in page.get("url", "")), pages[0])
+    return selected["webSocketDebuggerUrl"], selected
+
+
+def websocket_connect(ws_url):
+    parsed = urllib.parse.urlparse(ws_url)
+    if parsed.scheme != "ws":
+        raise SystemExit(f"Only ws:// endpoints are supported, got {ws_url}")
+    host = parsed.hostname or "127.0.0.1"
+    port = parsed.port or 80
+    path = parsed.path
+    if parsed.query:
+        path += "?" + parsed.query
+
+    sock = socket.create_connection((host, port), timeout=30)
+    key = base64.b64encode(os.urandom(16)).decode("ascii")
+    request = (
+        f"GET {path} HTTP/1.1\r\n"
+        f"Host: {host}:{port}\r\n"
+        "Upgrade: websocket\r\n"
+        "Connection: Upgrade\r\n"
+        f"Sec-WebSocket-Key: {key}\r\n"
+        "Sec-WebSocket-Version: 13\r\n"
+        "\r\n"
+    ).encode("ascii")
+    sock.sendall(request)
+    response = b""
+    while b"\r\n\r\n" not in response:
+        response += sock.recv(4096)
+    header = response.split(b"\r\n\r\n", 1)[0].decode("iso-8859-1")
+    if " 101 " not in header.split("\r\n", 1)[0]:
+        raise SystemExit(f"WebSocket upgrade failed:\n{header}")
+    accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii")
+    if f"Sec-WebSocket-Accept: {accept}".lower() not in header.lower():
+        raise SystemExit("WebSocket accept header did not match.")
+    return sock
+
+
+def send_frame(sock, payload):
+    data = payload.encode("utf-8")
+    header = bytearray([0x81])
+    if len(data) < 126:
+        header.append(0x80 | len(data))
+    elif len(data) < 65536:
+        header.append(0x80 | 126)
+        header.extend(struct.pack("!H", len(data)))
+    else:
+        header.append(0x80 | 127)
+        header.extend(struct.pack("!Q", len(data)))
+    mask = os.urandom(4)
+    header.extend(mask)
+    masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(data))
+    sock.sendall(bytes(header) + masked)
+
+
+def recv_exact(sock, size):
+    chunks = []
+    remaining = size
+    while remaining:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            raise EOFError("WebSocket closed")
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
+def recv_frame(sock):
+    first, second = recv_exact(sock, 2)
+    opcode = first & 0x0F
+    masked = second & 0x80
+    length = second & 0x7F
+    if length == 126:
+        length = struct.unpack("!H", recv_exact(sock, 2))[0]
+    elif length == 127:
+        length = struct.unpack("!Q", recv_exact(sock, 8))[0]
+    mask = recv_exact(sock, 4) if masked else b""
+    payload = recv_exact(sock, length) if length else b""
+    if masked:
+        payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
+    if opcode == 0x8:
+        raise EOFError("WebSocket closed")
+    if opcode == 0x9:
+        return None
+    if opcode != 0x1:
+        return None
+    return payload.decode("utf-8")
+
+
+def call(sock, method, params=None, message_id=1):
+    send_frame(sock, json.dumps({"id": message_id, "method": method, "params": params or {}}))
+    while True:
+      message = recv_frame(sock)
+      if not message:
+          continue
+      parsed = json.loads(message)
+      if parsed.get("id") == message_id:
+          return parsed
+
+
+def main():
+    expression = sys.stdin.read().strip()
+    if not expression:
+        expression = "({ title: document.title, url: location.href, text: document.body.innerText.slice(0, 1000) })"
+
+    ws_url, target = get_page_ws_url()
+    sock = websocket_connect(ws_url)
+    try:
+        result = call(
+            sock,
+            "Runtime.evaluate",
+            {
+                "expression": expression,
+                "returnByValue": True,
+                "awaitPromise": True,
+                "timeout": 30000,
+            },
+        )
+    finally:
+        sock.close()
+
+    print(json.dumps({"target": target, "result": result}, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()