OGMatrix
commented
1 year ago already did it but i have to implementing it in your code.
so basically you need the websocket
def get_board(self, access_token_in):
logger.debug("Connecting and obtaining board images")
while True:
try:
ws = create_connection(
"wss://gql-realtime-2.reddit.com/query",
origin="https://garlic-bread.reddit.com",
)
break
except Exception:
logger.error(
"Failed to connect to websocket, trying again in 30 seconds..."
)
time.sleep(30)
ws.send(
json.dumps(
{
"type": "connection_init",
"payload": {"Authorization": "Bearer " + access_token_in},
}
)
)
while True:
try:
msg = ws.recv()
except WebSocketConnectionClosedException as e:
logger.error(e)
continue
if msg is None:
logger.error("Reddit failed to acknowledge connection_init")
exit()
if msg.startswith('{"type":"connection_ack"}'):
logger.debug("Connected to WebSocket server")
break
logger.debug("Obtaining Canvas information")
ws.send(
json.dumps(
{
"id": "1",
"type": "start",
"payload": {
"variables": {
"input": {
"channel": {
"teamOwner": "GARLICBREAD",
"category": "CONFIG",
}
}
},
"extensions": {},
"operationName": "configuration",
"query": "subscription configuration($input: SubscribeInput!) {\n subscribe(input: $input) {\n id\n ... on BasicMessage {\n data {\n __typename\n ... on ConfigurationMessageData {\n colorPalette {\n colors {\n hex\n index\n __typename\n }\n __typename\n }\n canvasConfigurations {\n index\n dx\n dy\n __typename\n }\n canvasWidth\n canvasHeight\n __typename\n }\n }\n __typename\n }\n __typename\n }\n}\n",
},
}
)
)
while True:
canvas_payload = json.loads(ws.recv())
if canvas_payload["type"] == "data":
canvas_details = canvas_payload["payload"]["data"]["subscribe"]["data"]
logger.debug("Canvas config: {}", canvas_payload)
break
canvas_sockets = []
canvas_count = len(canvas_details["canvasConfigurations"])
for i in range(0, canvas_count):
canvas_sockets.append(2 + i)
logger.debug("Creating canvas socket {}", canvas_sockets[i])
ws.send(
json.dumps(
{
"id": str(2 + i),
"type": "start",
"payload": {
"variables": {
"input": {
"channel": {
"teamOwner": "GARLICBREAD",
"category": "CANVAS",
"tag": str(i),
}
}
},
"extensions": {},
"operationName": "replace",
"query": """subscription replace($input: SubscribeInput!) {
subscribe(input: $input) {
id
... on BasicMessage {
data {
__typename
... on FullFrameMessageData {
__typename
name
timestamp
}
... on DiffFrameMessageData {
__typename
name
currentTimestamp
previousTimestamp
}
}
}
}
}""",
},
}
)
)
imgs = []
logger.debug("A total of {} canvas sockets opened", len(canvas_sockets))
while len(canvas_sockets) > 0:
temp = json.loads(ws.recv())
logger.debug("Waiting for WebSocket message")
print(temp)
if temp["type"] == "data":
logger.debug("Received WebSocket data type message")
msg = temp["payload"]["data"]["subscribe"]
print(msg)
if msg["data"]["__typename"] == "FullFrameMessageData":
logger.debug("Received full frame message")
img_id = int(temp["id"])
logger.debug("Image ID: {}", img_id)
if img_id in canvas_sockets:
logger.debug("Getting image: {}", msg["data"]["name"])
try:
imgs.append(
[
img_id,
Image.open(
BytesIO(
requests.get(
msg["data"]["name"],
stream=True,
proxies=proxy.get_random_proxy(
self, name=None
),
).content
)
),
]
)
except Exception:
print("Error with image")
canvas_sockets.remove(img_id)
logger.debug(
"Canvas sockets remaining: {}", len(canvas_sockets)
)
for i in range(0, canvas_count - 1):
ws.send(json.dumps({"id": str(2 + i), "type": "stop"}))
ws.close()
new_img_width = (
max(map(lambda x: x["dx"], canvas_details["canvasConfigurations"])) +
canvas_details["canvasWidth"]
)
logger.debug("New image width: {}", new_img_width)
new_img_height = (
max(map(lambda x: x["dy"], canvas_details["canvasConfigurations"])) +
canvas_details["canvasHeight"]
)
logger.debug("New image height: {}", new_img_height)
new_img = Image.new("RGB", (new_img_width, new_img_height))
for idx, img in enumerate(sorted(imgs, key=lambda x: x[0])):
logger.debug("Adding image (ID {}): {}", img[0], img[1])
dx_offset = int(canvas_details["canvasConfigurations"][idx]["dx"])
dy_offset = int(canvas_details["canvasConfigurations"][idx]["dy"])
new_img.paste(img[1], (dx_offset, dy_offset))
return new_imgthis is a working code do something with it and implementing it @michaelshumshum
tryptech
michaelshumshum
Alvaro2x2
hedihadi
Hedwig7s
Someduckwithahat
NFKRZZ
cereal-kitty
get_boardmethod inmanagement.pywith the new method.that's pretty much it. the request payload is the same as before. the JWT bearer token continues to use the API token. use the developer tools in your browser to observe the network behaviour.
the person who gets the script working again, i will provide you with my 50 accounts from last year.