Esse é o request: curl 'https://poe.com/api/gql_PO...
Créé le : 19 août 2025
Répondu en utilisant GPT-5 Thinking par Chat01
Créé le : 19 août 2025
Répondu en utilisant GPT-5 Thinking par Chat01
Esse é o request: curl 'https://poe.com/api/gql_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-queryname: useMentionedBotsInputsQuery'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: cf8a79b3dc7b7229af1239d644a87dd1'
-H 'poe-tchannel: poe-chan105-8888-eubnxknfqzvcppphsprm'
-H 'poegraphql: 1'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/GPT-5'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '{"queryName":"useMentionedBotsInputsQuery","variables":{"chatCode":"","hasChat":false,"fullChatInput":"O","currentPartialHandle":""},"extensions":{"hash":"6fb81c76162cce33770988fd687dd93feb357cb4c1ae82a6340279659dcc28aa"}}' ;
curl 'https://poe.com/api/gql_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-queryname: useMentionedBotsInputsQuery'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: c7d7d766647f5ac02349392861e63e07'
-H 'poe-tchannel: poe-chan105-8888-eubnxknfqzvcppphsprm'
-H 'poegraphql: 1'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/GPT-5'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '{"queryName":"useMentionedBotsInputsQuery","variables":{"chatCode":"","hasChat":false,"fullChatInput":"Oi","currentPartialHandle":""},"extensions":{"hash":"6fb81c76162cce33770988fd687dd93feb357cb4c1ae82a6340279659dcc28aa"}}' ;
curl 'https://api.sprig.com/sdk/1/visitors/8c3e0461-99e9-4da5-8d11-05e9bb71d923/events'
-X 'OPTIONS'
-H 'Accept: /'
-H 'Accept-Language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'Access-Control-Request-Headers: authorization,content-type,sprig-modules,userleap-platform,x-ul-environment-id,x-ul-installation-method,x-ul-sdk-version,x-ul-user-id,x-ul-visitor-id'
-H 'Access-Control-Request-Method: POST'
-H 'Connection: keep-alive'
-H 'Origin: https://poe.com'
-H 'Referer: https://poe.com/'
-H 'Sec-Fetch-Dest: empty'
-H 'Sec-Fetch-Mode: cors'
-H 'Sec-Fetch-Site: cross-site'
-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0' ;
curl 'https://api.sprig.com/sdk/1/visitors/8c3e0461-99e9-4da5-8d11-05e9bb71d923/events'
-H 'Accept: /'
-H 'Accept-Language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'Authorization: Bearer tokenNotNeeded'
-H 'Connection: keep-alive'
-H 'Content-Type: application/json'
-H 'Origin: https://poe.com'
-H 'Referer: https://poe.com/'
-H 'Sec-Fetch-Dest: empty'
-H 'Sec-Fetch-Mode: cors'
-H 'Sec-Fetch-Site: cross-site'
-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sprig-modules: replay'
-H 'userleap-platform: web'
-H 'x-ul-environment-id: OFiT3eT6FKrU'
-H 'x-ul-installation-method: web-npm'
-H 'x-ul-sdk-version: 2.32.6'
-H 'x-ul-user-id: 3059850752'
-H 'x-ul-visitor-id: 8c3e0461-99e9-4da5-8d11-05e9bb71d923'
--data-raw '{"event":"send_chat_clicked","metadata":{"url":"https://poe.com/GPT-5"}}' ;
curl 'https://poe.com/api/gql_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-queryname: sendMessageMutation'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: 676d4909e2c4331c47c204f6893bf640'
-H 'poe-tchannel: poe-chan105-8888-eubnxknfqzvcppphsprm'
-H 'poegraphql: 0'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/GPT-5'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '{"queryName":"sendMessageMutation","variables":{"chatId":null,"bot":"GPT-5","query":"Oi","source":{"sourceType":"chat_input","chatInputMetadata":{"useVoiceRecord":false}},"clientNonce":"dEnlY0Z67mkuUuRU","sdid":"48ec8f8e-87b1-41bd-8217-7536bfea2651","attachments":[],"existingMessageAttachmentsIds":[],"shouldFetchChat":true,"referencedMessageId":null,"parameters":null,"fileHashJwts":[]},"extensions":{"hash":"c877ecaabad9cc7bf70f5aef44aafc65b0ac9b4b13abcc15464aee443d482f41"}}' ;
curl 'https://qph.cf2.poecdn.net/main-thumb-pb-3045-50-wkomexlqbcxpravoniszumuijaodcygr.jpeg'
-H 'accept: image/avif,image/webp,image/apng,image/svg+xml,image/,/;q=0.8'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-b '__cf_bm=TZvBxs82ho.JRstbSGgimG9pKlVWFcHXyO9uZGln354-1755600519-1.0.1.1-HirGMWdBUzTZJd9gpF2Ke5wxjjuw3RFUhsOwAsk5T5eM9cXdV5bpIWrBciMXS0qd.rios4y.6bq0UihCUtFwB1khThNqk1.sI.BQyLhgavk'
-H 'priority: i'
-H 'referer: https://poe.com/'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: image'
-H 'sec-fetch-mode: no-cors'
-H 'sec-fetch-site: cross-site'
-H 'sec-fetch-storage-access: active'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0' ;
curl 'https://www.facebook.com/tr/?id=1611526079610489&ev=PageView&dl=https%3A%2F%2Fpoe.com%2Fchat%2Fs66jwoc3zessgej3vr&rl=https%3A%2F%2Fpoe.com%2Fgoogle_callback%3Fstate%3DeyJwZXJtcyI6ICIiLCAiY3NyZiI6ICI4OGM2YmJmYmZjZmJkZDdkMmJmODhjNDUwYzUyNTdhMiIsICJwbGF0Zm9ybSI6ICJ3ZWIiLCAidGFyZ2V0IjogIiIsICJjdXJyZW50X3VybCI6ICIvbG9naW4ifQ%253D%253D%26code%3D4%252F0AVMBsJg6fRHYiE4joFTI5lezisZWwTy2F9LZhXiE0uBcJ4PKxIHO0SyiQMj44ORy36WEuQ%26scope%3Demail%2Bprofile%2Bhttps%253A%252F%252Fwww.googleapis.com%252Fauth%252Fuserinfo.profile%2Bhttps%253A%252F%252Fwww.googleapis.com%252Fauth%252Fuserinfo.email%2Bopenid%26authuser%3D0%26prompt%3Dconsent%26didAttemptSignupLogin%3Dtrue&if=false&ts=1755602317310&sw=1600&sh=900&v=2.9.224&r=stable&ec=1&o=4126&fbp=fb.1.1755600520633.6733147262899197&ler=other&pmd$$title$$=GPT-5%20-%20Poe&plt=848.8000000000466&it=1755601483098&coo=false&rqm=GET' -H 'Referer: https://poe.com/' ;
curl 'https://www.facebook.com/privacy_sandbox/pixel/register/trigger/?id=1611526079610489&ev=PageView&dl=https%3A%2F%2Fpoe.com%2Fchat%2Fs66jwoc3zessgej3vr&rl=https%3A%2F%2Fpoe.com%2Fgoogle_callback%3Fstate%3DeyJwZXJtcyI6ICIiLCAiY3NyZiI6ICI4OGM2YmJmYmZjZmJkZDdkMmJmODhjNDUwYzUyNTdhMiIsICJwbGF0Zm9ybSI6ICJ3ZWIiLCAidGFyZ2V0IjogIiIsICJjdXJyZW50X3VybCI6ICIvbG9naW4ifQ%253D%253D%26code%3D4%252F0AVMBsJg6fRHYiE4joFTI5lezisZWwTy2F9LZhXiE0uBcJ4PKxIHO0SyiQMj44ORy36WEuQ%26scope%3Demail%2Bprofile%2Bhttps%253A%252F%252Fwww.googleapis.com%252Fauth%252Fuserinfo.profile%2Bhttps%253A%252F%252Fwww.googleapis.com%252Fauth%252Fuserinfo.email%2Bopenid%26authuser%3D0%26prompt%3Dconsent%26didAttemptSignupLogin%3Dtrue&if=false&ts=1755602317310&sw=1600&sh=900&v=2.9.224&r=stable&ec=1&o=4126&fbp=fb.1.1755600520633.6733147262899197&ler=other&pmd$$title$$=GPT-5%20-%20Poe&plt=848.8000000000466&it=1755601483098&coo=false&rqm=FGET' -H 'Referer: https://poe.com/' ;
curl 'https://poe.com/apple-touch-icon.png'
-H 'accept: image/avif,image/webp,image/apng,image/svg+xml,image/,/;q=0.8'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'if-modified-since: Fri, 31 Jan 2025 10:05:04 GMT'
-H 'if-none-match: W/"19e9-194bbd23880"'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: image'
-H 'sec-fetch-mode: no-cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0' ;
curl 'https://psc2.cf2.poecdn.net/assets/_next/static/chunks/6881.20ae643cfd0cb2fe.js'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-b '__cf_bm=AOUty9I4rOnHMfOiiOjY_Lxg0loeK0X5vSvj9zwBxlc-1755602317-1.0.1.1-g7na2fFZmiWbA57thskkVgn4F2G78YpqBRkOjyMzlUuUA8N8Eck.MnX7hOsdsrkMwwAhecoU5ZOFDYE_3JIYFPGAv3tvTVwVMU5IiVDvWng'
-H 'referer: https://poe.com/'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: script'
-H 'sec-fetch-mode: no-cors'
-H 'sec-fetch-site: cross-site'
-H 'sec-fetch-storage-access: active'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0' ;
curl 'https://poe.com/api/gql_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-queryname: ChatLastMessageAreaQuery'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: 155534d0c4c15d8cac2e34f758bc7e60'
-H 'poe-tchannel: poe-chan105-8888-eubnxknfqzvcppphsprm'
-H 'poegraphql: 1'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '{"queryName":"ChatLastMessageAreaQuery","variables":{"messageNodeId":"TWVzc2FnZTo0MjUwMDY5MTI5MzY=","chatNodeId":"Q2hhdDoxMjQ4MDkxNTYw"},"extensions":{"hash":"dcf4b7a90ee6728a544c4225bdb531b6ad33ab2a8e0e545680406f3ba115fb67"}}' ;
curl 'https://poe.com/api/gql_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-queryname: useChatLastMessageImpressionTracker_ChatMarkAsSeen_Mutation'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: 18c7d02ce7516116888858a9bc8a6d4b'
-H 'poe-tchannel: poe-chan105-8888-eubnxknfqzvcppphsprm'
-H 'poegraphql: 1'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '{"queryName":"useChatLastMessageImpressionTracker_ChatMarkAsSeen_Mutation","variables":{"chatId":1248091560},"extensions":{"hash":"dcfc17acd2ec0bbd1525fd786ead7689d8c6201c59300c31f2d5d335984689db"}}' ;
curl 'https://qph.cf2.poecdn.net/main-thumb-pb-5902949-200-tqimwapupgzbprlyuevlrrtjjztgxwjx.jpeg'
-H 'accept: image/avif,image/webp,image/apng,image/svg+xml,image/,/*;q=0.8'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-b '__cf_bm=AOUty9I4rOnHMfOiiOjY_Lxg0loeK0X5vSvj9zwBxlc-1755602317-1.0.1.1-g7na2fFZmiWbA57thskkVgn4F2G78YpqBRkOjyMzlUuUA8N8Eck.MnX7hOsdsrkMwwAhecoU5ZOFDYE_3JIYFPGAv3tvTVwVMU5IiVDvWng'
-H 'priority: i'
-H 'referer: https://poe.com/'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: image'
-H 'sec-fetch-mode: no-cors'
-H 'sec-fetch-site: cross-site'
-H 'sec-fetch-storage-access: active'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0' ;
curl 'https://qph.cf2.poecdn.net/main-thumb-pb-6015-200-ppgahqisjenazkillisfowssvnmpknwg.jpeg'
-H 'sec-ch-ua-platform: "Windows"'
-H 'Referer: https://poe.com/'
-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0' ;
curl 'https://qph.cf2.poecdn.net/main-thumb-pb-5576465-200-rbkdokveoijnssmrtxehspndcuuhttyr.jpeg'
-H 'sec-ch-ua-platform: "Windows"'
-H 'Referer: https://poe.com/'
-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0' ;
curl 'https://qph.cf2.poecdn.net/main-thumb-pb-5176044-200-ixsrhozcvbrhrvwgdxghshtrqpjjashv.jpeg'
-H 'sec-ch-ua-platform: "Windows"'
-H 'Referer: https://poe.com/'
-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0' ;
curl 'https://poe.com/api/receive_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: 1206d995887ae6b242a14090c31c5684'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '[{"category":"poe/statsd_event","data":{"category":"count","key":"poe.message.nonce_e2e.added","value":1,"extra_data":{"socketState":"open","channel":"poe-chan105-8888-eubnxknfqzvcppphsprm"},"timeout":false,"hidden":false}},{"category":"poe/statsd_event","data":{"category":"count","key":"poe.message.nonce_e2e.cleared","value":1,"extra_data":{"socketState":"open","channel":"poe-chan105-8888-eubnxknfqzvcppphsprm","lastPongMessageReceivedTime":834566.5000000001,"elapsedTimeFromLastPongMessage":362.79999999981374,"timeSpent":517.5999999998603,"hidden":false}}},{"category":"poe/statsd_event","data":{"key":"poe.message.mutation.sendMessageMutation.http_status.200","value":1,"category":"count","extra_data":{}}},{"category":"poe/statsd_event","data":{"key":"poe.speed.web_vitals.next_render","value":167.60000000009313,"category":"time","path":"/[handle]","extra_data":{"initial_page":"/[handle]","page":"/[handle]"}}},{"category":"poe/statsd_event","data":{"key":"poe.speed.web_vitals.next_route_change_to_render","value":20.399999999906868,"category":"time","path":"/[handle]","extra_data":{"initial_page":"/[handle]","page":"/[handle]"}}},{"category":"poe/page_load","data":{"url":"/chat/s66jwoc3zessgej3vr","pathname":"/[handle]","navigationType":"SPA","timingData":{"pageLoad":198.69999999995343},"previousUrl":"/GPT-5"}},{"category":"poe/statsd_event","data":{"category":"count","key":"poe.message.heartbeat.complete","value":1,"extra_data":{"messageId":"425006912936"}}},{"category":"poe/bot_response_speed","data":{"bot_name":"gpt_o4_alpha","time_to_first_typing_indicator":747,"time_to_first_subscription_response":816,"time_to_full_bot_response":2097,"full_response_length":30,"full_response_word_count":6,"human_message_id":425006897576,"bot_message_id":425006912936,"chat_id":"OPTIMISTIC_CHAT_207626171","bot_response_status":"success"}},{"category":"poe/statsd_event","data":{"category":"count","key":"poe.job.heartbeat.complete","value":1,"extra_data":{"jobId":"6581991840","heartbeatHistory":"sendMessage:not_started, jobStarted:not_started, jobUpdated:not_started, jobUpdated:completed","source":"jobUpdated","jobState":"completed"}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":476,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":0}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":476,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":1}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":476,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":2}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":476,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":3}}}]' ;
curl 'https://poe.com/api/receive_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: db7c490626fbbc500e9532a1d3a6ef3d'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '[{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":1125,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":4}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":1315,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":5}}},{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.item","value":1465,"extra_data":{"followupAction":"SendMessageFollowupAction","messageId":425006912936,"index":6}}}]' ;
curl 'https://poe.com/api/receive_POST'
-H 'accept: /'
-H 'accept-language: pt-BR,pt;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6'
-H 'content-type: application/json'
-b 'p-b=fnWI711UP20Z-5hhB4qmXA%3D%3D; _gcl_au=1.1.1600208785.1755600520; _fbp=fb.1.1755600520633.6733147262899197; poe-tchannel-channel=poe-chan105-8888-eubnxknfqzvcppphsprm; __cf_bm=EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog; cf_clearance=thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w; OptanonConsent=isGpcEnabled=0&datestamp=Tue+Aug+19+2025+08%3A04%3A43+GMT-0300+(Hor%C3%A1rio+Padr%C3%A3o+de+Bras%C3%ADlia)&version=202405.1.0&browserGpcFlag=0&isIABGlobal=false&hosts=&landingPath=NotLandingPage&groups=C0001%3A1%2CC0003%3A1%2CC0004%3A1&AwaitingReconsent=false&geolocation=BR%3BSC; OptanonAlertBoxClosed=2025-08-19T11:04:43.244Z'
-H 'origin: https://poe.com'
-H 'poe-formkey: 16318c91334ce2978bb3b21e7c18d684'
-H 'poe-revision: e2acc7025b43e08e88164ba8105273f37fbeaa26'
-H 'poe-tag-id: 01e7cea6e3f98b837e6e863ad6a67399'
-H 'priority: u=1, i'
-H 'referer: https://poe.com/chat/s66jwoc3zessgej3vr'
-H 'sec-ch-ua: "Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"'
-H 'sec-ch-ua-mobile: ?0'
-H 'sec-ch-ua-platform: "Windows"'
-H 'sec-fetch-dest: empty'
-H 'sec-fetch-mode: cors'
-H 'sec-fetch-site: same-origin'
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
--data-raw '[{"category":"poe/statsd_event","data":{"category":"time","key":"poe.speed.product.follow_up_action.full_list","value":1465,"extra_data":{"messageId":425006912936}}}]' esse é o meu código: import asyncio
import json
import random
import re
import string
from typing import Optional, Dict, Any, List, Tuple
import urllib.parse
import requests
import websockets
from playwright.async_api import async_playwright
P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b)
P_LAT_COOKIE = "" # opcional
CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w"
CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog"
P_SB = ""
POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm"
BOT_NAME = "GPT-5" # slug visível na URL
PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar
POE_URL_ROOT = "https://poe.com"
POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST"
RECV_URL = f"{POE_URL_ROOT}/api/receive_POST"
BASE_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0",
"Accept": "/",
"Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8",
"Origin": POE_URL_ROOT,
"Referer": f"{POE_URL_ROOT}/{BOT_NAME}",
"Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Content-Type": "application/json",
"poegraphql": "0",
}
def get_random_nonce(n: int = 16) -> str:
return "".join(random.choices(string.ascii_letters + string.digits, k=n))
def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str:
return "; ".join(f"{c.name}={c.value}" for c in jar)
def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]:
for c in jar:
name = c.name.lower()
if "tchannel" in name and "channel" in name:
return c.value
return None
def ws_url_from_tchannel(channel: str) -> str:
m = re.search(r"chan(\d+)-", channel)
if not m:
raise ValueError(f"tchannel inesperado: {channel}")
shard = m.group(1)
return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}"
def parse_cookie_header_and_set(session: requests.Session, cookie_header: str):
"""
Lê um header "Cookie: a=b; c=d" capturado do browser e aplica os cookies no session jar.
Útil para garantir tchannel-cookie e demais sinalizadores.
"""
if not cookie_header:
return
parts = [p.strip() for p in cookie_header.split(";") if "=" in p]
for part in parts:
try:
k, v = part.split("=", 1)
k = k.strip()
v = v.strip()
# domain e path padrão para poe.com
session.cookies.set(k, v, domain="poe.com", path="/")
except Exception:
continue
SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str]) -> Dict[str, Optional[str]]:
"""
1) Abre a página do bot com Playwright com a sua sessão.
2) Captura a request real de sendMessageMutation (payload + headers).
- FULFILL para não enviar de verdade (fake 200), mas guarda uma cópia exata.
3) Coleta formkey, revision, tag-id, sdid e tchannel da MESMA sessão/build.
4) Tenta ler o hash dos bundles JS como fallback.
Retorna um dicionário com tudo que precisamos para fazer um POST válido.
"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
text# Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey a partir da função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas revision_meta = None tag_id_meta = None try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: pass try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: pass # sdid vindo do localStorage (nem sempre existe; o payload-base é a fonte da verdade) sdid_local = None try: sdid_local = await page.evaluate(""" (() => { try { for (const k of Object.keys(localStorage)) { if (/sdid/i.test(k)) { const v = localStorage.getItem(k); if (v && v.length >= 10) return v; } } } catch (e) {} return null; })() """) except Exception: pass # tenta pegar hashes dos bundles (fallback) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = SEND_RE.search(txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = SUBS_RE.search(txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Intercepta gql_POST para capturar payload-base e headers REAIS captured: Dict[str, Any] = { "payload": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): # Copia TODOS os headers (lower-case keys) hdrs = {k.lower(): v for k, v in request.headers.items()} captured["req_headers"] = {**captured["req_headers"], **hdrs} if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] qn = request.headers.get("poe-queryname", "") body = request.post_data or "" if qn == "sendMessageMutation" or "sendMessageMutation" in body: try: payload = json.loads(body) captured["payload"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # Fulfill com 200 para NÃO enviar a "probe" de verdade await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Dispara um "probe" como o front (será bloqueado pelo fulfill acima) try: locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill("probe") else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await page.keyboard.type("probe") await page.keyboard.press("Enter") except Exception: pass # Aguarda a captura do payload-base for _ in range(100): if captured["payload"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais (headers/payload sempre vencem o que veio de meta) final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_base = captured["payload"] hash_intercept = captured["hash"] return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "sdid_local": sdid_local, "payload_base": payload_base, "req_headers": captured["req_headers"] or {}, }
def extract_text_pieces(event: Any) -> List[str]:
out: List[str] = []
if isinstance(event, dict):
for k in ("text_new", "response", "delta", "text"):
v = event.get(k)
if isinstance(v, str) and v:
out.append(v)
for v in event.values():
if isinstance(v, (dict, list)):
out.extend(extract_text_pieces(v))
elif isinstance(event, list):
for x in event:
out.extend(extract_text_pieces(x))
return out
def is_final_event(event: Dict[str, Any]) -> bool:
data = event.get("data") or {}
cat = (event.get("category") or "").lower()
state = (data.get("state") or data.get("message_state") or "").lower()
if "completed" in cat:
return True
if state in {"complete", "completed", "stopped", "aborted"}:
return True
if data.get("is_final") is True:
return True
return False
async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str:
tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK
if not tchannel:
raise RuntimeError("Sem poe-tchannel (cookie/forçado).")
textws_url = ws_url_from_tchannel(tchannel) cookie_header = cookie_header_from_jar(session.cookies) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full
async def listen_via_longpoll(session: requests.Session) -> str:
acc = ""
while True:
r = session.post(RECV_URL, json=[])
r.raise_for_status()
events = r.json()
if not isinstance(events, list):
continue
for ev in events:
pieces = extract_text_pieces(ev)
if pieces:
txt = "".join(pieces)
print(txt, end="", flush=True)
acc += txt
if is_final_event(ev):
print("\n--- Resposta Completa (receive_POST) ---")
return acc
def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]:
"""
Retorna strings curtas com os principais campos para log/diagnóstico.
"""
vars_obj = (payload_base or {}).get("variables") or {}
exts_obj = (payload_base or {}).get("extensions") or {}
sdid = vars_obj.get("sdid")
bot = vars_obj.get("bot")
hashv = exts_obj.get("hash")
hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…"
krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]}
ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items())
return hsum, ksum
async def main():
# Sessão HTTP
session = requests.Session()
session.headers.update(BASE_HEADERS)
# Cookies base
session.cookies.update({"p-b": P_B_COOKIE})
if P_LAT_COOKIE:
session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/")
if CF_CLEARANCE:
session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/")
if CF_BM:
session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/")
if P_SB:
session.cookies.set("p-sb", P_SB, domain="poe.com", path="/")
texttry: # Bootstrap – captura payload/headers da sessão/conta ATUAL info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, ) payload_base = info.get("payload_base") req_headers = info.get("req_headers") or {} if not isinstance(payload_base, dict) or payload_base.get("queryName") != "sendMessageMutation": raise RuntimeError("Não capturei o payload_base de sendMessageMutation. Tente novamente.") # Preferir header capturado SEMPRE formkey = info.get("formkey") revision = info.get("revision") tag_id = info.get("tag_id") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK send_hash = (payload_base.get("extensions") or {}).get("hash") or info.get("send_hash") # Copiar cookies do header da request interceptada (garante cookie de canal) parse_cookie_header_and_set(session, info.get("cookie_header") or "") # Também forçar cookies de canal coerentes com o header poe-tchannel if tchannel: session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") session.cookies.set("poe-channel-channel", tchannel, domain="poe.com", path="/") # Setar headers críticos A PARTIR DO INTERCEPT session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tag_id: session.headers["poe-tag-id"] = tag_id if tchannel: session.headers["poe-tchannel"] = tchannel # Montar payload FINAL: **NÃO alterar bot nem sdid** – só query e clientNonce gql_payload = json.loads(json.dumps(payload_base)) # deep copy vars_obj = gql_payload.setdefault("variables", {}) vars_obj["query"] = PROMPT vars_obj["clientNonce"] = get_random_nonce() # hash: manter o do payload_base (ou aplicar send_hash se preciso) gql_payload.setdefault("extensions", {})["hash"] = send_hash # Cabeçalhos do POST = sessão + Apollo headers do front gql_headers = dict(session.headers) # copiar Apollo/GraphQL/trace do intercept se existirem copy_allow = { "poe-queryname", "apollographql-client-name", "apollographql-client-version", "x-apollo-operation-name", "x-apollo-operation-id", "sentry-trace", "baggage", "poegraphql", "referer", "origin" } for k, v in req_headers.items(): lk = k.lower() if lk in copy_allow: gql_headers[lk] = v if "poe-queryname" not in gql_headers: gql_headers["poe-queryname"] = "sendMessageMutation" # Garantir Referer coerente com o bot atual gql_headers["Referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["Origin"] = POE_URL_ROOT # LOG – status resumido (sem dados sensíveis completos) p_sum, h_sum = summarize_payload_and_headers(payload_base, {k.lower(): v for k, v in req_headers.items()}) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag:{(tag_id or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"hash:{(send_hash or '')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) print("poe-queryname:", gql_headers.get("poe-queryname")) # ======================= ENVIO DO POST ================================= print("\nEnviando a pergunta via GraphQL…") resp = session.post(POE_API_URL, json=gql_payload, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() # Suporte a messageCreate e messageEdgeCreate root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # ======================= RECEPÇÃO DA RESPOSTA ========================== subs_hash = info.get("sub_hash") if subs_hash and re.fullmatch(r"[0-9a-f]{64}", subs_hash or ""): try: final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) except Exception as ws_err: print(f"[WS falhou: {ws_err}] -> tentando receive_POST…") final_answer = await listen_via_longpoll(session) else: final_answer = await listen_via_longpoll(session) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") try: print("payload_enviado.variables.keys:", list((gql_payload.get("variables") or {}).keys())) print("poe-queryname usado:", gql_headers.get("poe-queryname")) except Exception: pass except Exception as e: print(f"Erro: {e}")
if name == "main":
asyncio.run(main())" pq fica dando: formkey:16318c… tag:8d22ca61… tchannel_hdr:poe-chan107-8888-u… tchannel_cookie:poe-chan107-8888-u… send_hash:c877ecaa… sub_hash:n/a… poe-revision:e2acc7025b…
poe-queryname: sendMessageMutation
Enviando a pergunta via GraphQL…
HTTPError: 400 Client Error: Bad Request for url: https://poe.com/api/gql_POST
Body: {"success":false,"message":"Bad Request"}
payload_enviado.variables.keys: ['chatId', 'bot', 'query', 'source', 'clientNonce', 'sdid', 'attachments', 'existingMessageAttachmentsIds', 'shouldFetchChat', 'referencedMessageId', 'parameters', 'fileHashJwts']
poe-queryname usado: sendMessageMutation
Pelo que você colou, o 400 Bad Request
não é “falta de cookie” nem Cloudflare: é o backend do Poe recusando o corpo do GraphQL porque o poe-tag-id
não bate com o payload que você está enviando.
No Poe, o header poe-tag-id
é recalculado a cada request (e, ao que tudo indica, leva em conta o formkey
, o poe-revision
, o queryName
e o JSON exato enviado em body
). No seu fluxo você:
sendMessageMutation
com a página (ótimo),variables.query
e gera outro clientNonce
,poe-tag-id
capturado antes.Resultado: a assinatura não confere → 400.
Se você quer postar via requests
(fora do navegador), gere o payload final dentro da própria página (para o seu PROMPT
) e capture aquele poe-tag-id
correspondente. Depois, mande exatamente o mesmo JSON e o mesmo header.
Ajustes mínimos no seu código:
Passe o PROMPT
para o get_bootstrap
e digite o próprio prompt (em vez de "probe"
), mantendo o fulfill(200)
para não enviar de verdade.
Não toque em variables.query
, não troque o clientNonce
, não reescreva extensions.hash
. Use o payload
interceptado como veio.
No POST, use o poe-tag-id
daquele intercept (não um estático no session.headers
).
Exemplo (dif/patch conceitual):
python# 1) aceite o prompt no bootstrap async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]: ... async def route_handler(route, request): if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" if hdrs.get("poe-queryname") == "sendMessageMutation" or "sendMessageMutation" in body: payload = json.loads(body) captured["payload"] = payload # ← payload FINAL p/ seu prompt captured["hash"] = (payload.get("extensions") or {}).get("hash") captured["tag_id_hdr"] = hdrs.get("poe-tag-id") # ← tag-id correspondente captured["req_headers"] = hdrs await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return await route.continue_() ... # 2) em vez de "probe", use o seu PROMPT locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) # ← PROMPT real aqui await page.keyboard.press("Enter") # o app monta mutation+headers para este texto ... # retorne tudo como já faz hoje (incluindo tag_id_hdr) return {..., "payload_base": captured["payload"], "req_headers": captured["req_headers"], "tag_id_hdr": captured["tag_id_hdr"]}
E no main()
:
pythoninfo = await get_bootstrap(BOT_NAME, {...}, PROMPT) # use EXATAMENTE o payload e o tag-id que a página gerou gql_payload = info["payload_base"] # não modifique! gql_headers = dict(session.headers) gql_headers["poe-queryname"] = "sendMessageMutation" if info.get("tag_id_hdr"): gql_headers["poe-tag-id"] = info["tag_id_hdr"] # tag id certo para ESTE body gql_headers["poegraphql"] = "0" # ok ser string # envie assim: resp = session.post(POE_API_URL, json=gql_payload, headers=gql_headers)
Dica: remova
session.headers["poe-tag-id"] = tag_id
global. Definapoe-tag-id
por request, sempre o que veio do intercept correspondente ao payload atual.
Mais simples (e menos frágil): não faça o POST manual. Com o Playwright, preencha a caixa de texto com o seu PROMPT
, pressione Enter e não intercepte o gql_POST
. O site faz tudo (formkey, tag, revision, nonce etc.). Depois você consome a resposta pelo WS (chatMessagesSubscription
) ou pelo receive_POST
. Seu listener já está pronto; é só tirar o fulfill
quando for a mensagem real.
poe-tag-id
muda a cada chamada (você mesmo viu ids diferentes nos curls). Usar um id capturado e depois mudar query
/clientNonce
garante 400.poegraphql: 0/1
em header é string de qualquer jeito; “0” funciona.poe-channel-channel
(parece typo). Só poe-tchannel-channel
já basta.Se quiser, eu te mando uma versão do get_bootstrap
já aceitando prompt_for_capture
e ajusto o main()
para seguir a Opção A sem tocar no payload.
Isos, faça isso. E adicione também suas observações: Outras observações rápidas
poe-tag-id muda a cada chamada (você mesmo viu ids diferentes nos curls). Usar um id capturado e depois mudar query/clientNonce garante 400.
poegraphql: 0/1 em header é string de qualquer jeito; “0” funciona.
Evite setar poe-channel-channel (parece typo). Só poe-tchannel-channel já basta.
Se algum cookie CF expirar, o erro típico é 403/401, não 400.
Se quiser, eu te mando uma versão do get_bootstrap já aceitando prompt_for_capture e ajusto o main() para seguir a Opção A sem tocar no payload." código completo: import asyncio
import json
import random
import re
import string
from typing import Optional, Dict, Any, List, Tuple
import urllib.parse
import requests
import websockets
from playwright.async_api import async_playwright
P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b)
P_LAT_COOKIE = "" # opcional
CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w"
CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog"
P_SB = ""
POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm"
BOT_NAME = "GPT-5" # slug visível na URL
PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar
POE_URL_ROOT = "https://poe.com"
POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST"
RECV_URL = f"{POE_URL_ROOT}/api/receive_POST"
BASE_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0",
"Accept": "/",
"Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8",
"Origin": POE_URL_ROOT,
"Referer": f"{POE_URL_ROOT}/{BOT_NAME}",
"Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Content-Type": "application/json",
"poegraphql": "0",
}
def get_random_nonce(n: int = 16) -> str:
return "".join(random.choices(string.ascii_letters + string.digits, k=n))
def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str:
return "; ".join(f"{c.name}={c.value}" for c in jar)
def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]:
for c in jar:
name = c.name.lower()
if "tchannel" in name and "channel" in name:
return c.value
return None
def ws_url_from_tchannel(channel: str) -> str:
m = re.search(r"chan(\d+)-", channel)
if not m:
raise ValueError(f"tchannel inesperado: {channel}")
shard = m.group(1)
return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}"
def parse_cookie_header_and_set(session: requests.Session, cookie_header: str):
"""
Lê um header "Cookie: a=b; c=d" capturado do browser e aplica os cookies no session jar.
Útil para garantir tchannel-cookie e demais sinalizadores.
"""
if not cookie_header:
return
parts = [p.strip() for p in cookie_header.split(";") if "=" in p]
for part in parts:
try:
k, v = part.split("=", 1)
k = k.strip()
v = v.strip()
# domain e path padrão para poe.com
session.cookies.set(k, v, domain="poe.com", path="/")
except Exception:
continue
SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str]) -> Dict[str, Optional[str]]:
"""
1) Abre a página do bot com Playwright com a sua sessão.
2) Captura a request real de sendMessageMutation (payload + headers).
- FULFILL para não enviar de verdade (fake 200), mas guarda uma cópia exata.
3) Coleta formkey, revision, tag-id, sdid e tchannel da MESMA sessão/build.
4) Tenta ler o hash dos bundles JS como fallback.
Retorna um dicionário com tudo que precisamos para fazer um POST válido.
"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
text# Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey a partir da função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas revision_meta = None tag_id_meta = None try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: pass try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: pass # sdid vindo do localStorage (nem sempre existe; o payload-base é a fonte da verdade) sdid_local = None try: sdid_local = await page.evaluate(""" (() => { try { for (const k of Object.keys(localStorage)) { if (/sdid/i.test(k)) { const v = localStorage.getItem(k); if (v && v.length >= 10) return v; } } } catch (e) {} return null; })() """) except Exception: pass # tenta pegar hashes dos bundles (fallback) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = SEND_RE.search(txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = SUBS_RE.search(txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Intercepta gql_POST para capturar payload-base e headers REAIS captured: Dict[str, Any] = { "payload": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): # Copia TODOS os headers (lower-case keys) hdrs = {k.lower(): v for k, v in request.headers.items()} captured["req_headers"] = {**captured["req_headers"], **hdrs} if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] qn = request.headers.get("poe-queryname", "") body = request.post_data or "" if qn == "sendMessageMutation" or "sendMessageMutation" in body: try: payload = json.loads(body) captured["payload"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # Fulfill com 200 para NÃO enviar a "probe" de verdade await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Dispara um "probe" como o front (será bloqueado pelo fulfill acima) try: locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill("probe") else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await page.keyboard.type("probe") await page.keyboard.press("Enter") except Exception: pass # Aguarda a captura do payload-base for _ in range(100): if captured["payload"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais (headers/payload sempre vencem o que veio de meta) final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_base = captured["payload"] hash_intercept = captured["hash"] return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "sdid_local": sdid_local, "payload_base": payload_base, "req_headers": captured["req_headers"] or {}, }
def extract_text_pieces(event: Any) -> List[str]:
out: List[str] = []
if isinstance(event, dict):
for k in ("text_new", "response", "delta", "text"):
v = event.get(k)
if isinstance(v, str) and v:
out.append(v)
for v in event.values():
if isinstance(v, (dict, list)):
out.extend(extract_text_pieces(v))
elif isinstance(event, list):
for x in event:
out.extend(extract_text_pieces(x))
return out
def is_final_event(event: Dict[str, Any]) -> bool:
data = event.get("data") or {}
cat = (event.get("category") or "").lower()
state = (data.get("state") or data.get("message_state") or "").lower()
if "completed" in cat:
return True
if state in {"complete", "completed", "stopped", "aborted"}:
return True
if data.get("is_final") is True:
return True
return False
async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str:
tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK
if not tchannel:
raise RuntimeError("Sem poe-tchannel (cookie/forçado).")
textws_url = ws_url_from_tchannel(tchannel) cookie_header = cookie_header_from_jar(session.cookies) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full
async def listen_via_longpoll(session: requests.Session) -> str:
acc = ""
while True:
r = session.post(RECV_URL, json=[])
r.raise_for_status()
events = r.json()
if not isinstance(events, list):
continue
for ev in events:
pieces = extract_text_pieces(ev)
if pieces:
txt = "".join(pieces)
print(txt, end="", flush=True)
acc += txt
if is_final_event(ev):
print("\n--- Resposta Completa (receive_POST) ---")
return acc
def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]:
"""
Retorna strings curtas com os principais campos para log/diagnóstico.
"""
vars_obj = (payload_base or {}).get("variables") or {}
exts_obj = (payload_base or {}).get("extensions") or {}
sdid = vars_obj.get("sdid")
bot = vars_obj.get("bot")
hashv = exts_obj.get("hash")
hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…"
krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]}
ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items())
return hsum, ksum
async def main():
# Sessão HTTP
session = requests.Session()
session.headers.update(BASE_HEADERS)
# Cookies base
session.cookies.update({"p-b": P_B_COOKIE})
if P_LAT_COOKIE:
session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/")
if CF_CLEARANCE:
session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/")
if CF_BM:
session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/")
if P_SB:
session.cookies.set("p-sb", P_SB, domain="poe.com", path="/")
texttry: # Bootstrap – captura payload/headers da sessão/conta ATUAL info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, ) payload_base = info.get("payload_base") req_headers = info.get("req_headers") or {} if not isinstance(payload_base, dict) or payload_base.get("queryName") != "sendMessageMutation": raise RuntimeError("Não capturei o payload_base de sendMessageMutation. Tente novamente.") # Preferir header capturado SEMPRE formkey = info.get("formkey") revision = info.get("revision") tag_id = info.get("tag_id") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK send_hash = (payload_base.get("extensions") or {}).get("hash") or info.get("send_hash") # Copiar cookies do header da request interceptada (garante cookie de canal) parse_cookie_header_and_set(session, info.get("cookie_header") or "") # Também forçar cookies de canal coerentes com o header poe-tchannel if tchannel: session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") session.cookies.set("poe-channel-channel", tchannel, domain="poe.com", path="/") # Setar headers críticos A PARTIR DO INTERCEPT session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tag_id: session.headers["poe-tag-id"] = tag_id if tchannel: session.headers["poe-tchannel"] = tchannel # Montar payload FINAL: **NÃO alterar bot nem sdid** – só query e clientNonce gql_payload = json.loads(json.dumps(payload_base)) # deep copy vars_obj = gql_payload.setdefault("variables", {}) vars_obj["query"] = PROMPT vars_obj["clientNonce"] = get_random_nonce() # hash: manter o do payload_base (ou aplicar send_hash se preciso) gql_payload.setdefault("extensions", {})["hash"] = send_hash # Cabeçalhos do POST = sessão + Apollo headers do front gql_headers = dict(session.headers) # copiar Apollo/GraphQL/trace do intercept se existirem copy_allow = { "poe-queryname", "apollographql-client-name", "apollographql-client-version", "x-apollo-operation-name", "x-apollo-operation-id", "sentry-trace", "baggage", "poegraphql", "referer", "origin" } for k, v in req_headers.items(): lk = k.lower() if lk in copy_allow: gql_headers[lk] = v if "poe-queryname" not in gql_headers: gql_headers["poe-queryname"] = "sendMessageMutation" # Garantir Referer coerente com o bot atual gql_headers["Referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["Origin"] = POE_URL_ROOT # LOG – status resumido (sem dados sensíveis completos) p_sum, h_sum = summarize_payload_and_headers(payload_base, {k.lower(): v for k, v in req_headers.items()}) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag:{(tag_id or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"hash:{(send_hash or '')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) print("poe-queryname:", gql_headers.get("poe-queryname")) # ======================= ENVIO DO POST ================================= print("\nEnviando a pergunta via GraphQL…") resp = session.post(POE_API_URL, json=gql_payload, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() # Suporte a messageCreate e messageEdgeCreate root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # ======================= RECEPÇÃO DA RESPOSTA ========================== subs_hash = info.get("sub_hash") if subs_hash and re.fullmatch(r"[0-9a-f]{64}", subs_hash or ""): try: final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) except Exception as ws_err: print(f"[WS falhou: {ws_err}] -> tentando receive_POST…") final_answer = await listen_via_longpoll(session) else: final_answer = await listen_via_longpoll(session) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") try: print("payload_enviado.variables.keys:", list((gql_payload.get("variables") or {}).keys())) print("poe-queryname usado:", gql_headers.get("poe-queryname")) except Exception: pass except Exception as e: print(f"Erro: {e}")
if name == "main":
asyncio.run(main())
perfeito — adaptei teu script para a Opção A: o get_bootstrap
agora aceita prompt_for_capture
, digita esse prompt na UI, intercepta a sendMessageMutation real (payload + headers exatos), e no main()
eu não altero nada do payload (nem query
, nem clientNonce
, nem extensions.hash
). Também removi o set do cookie poe-channel-channel
e o poe-tag-id
deixou de ser header “global”: agora ele é aplicado só no POST que reaproveita aquele payload capturado.
incluí as tuas “Observações rápidas” como comentários no topo e sigo elas no código.
python# -*- coding: utf-8 -*- # Poe GraphQL sender — Opção A (reutiliza payload/headers exatos da UI) # # Observações rápidas (importante): # - poe-tag-id muda a cada chamada. Se você capturar um e DEPOIS mudar query/clientNonce, # vai tomar 400 (assinatura não confere). # - poegraphql: "0" ou "1" como string funciona. # - Evite setar "poe-channel-channel" (parece typo). Só "poe-tchannel-channel" já basta. # - Se algum cookie do Cloudflare expirar, o erro típico é 403/401, não 400. # # Fluxo: # 1) get_bootstrap(bot, cookies, prompt) abre a página logada, digita o PROMPT e # intercepta a sendMessageMutation EXATA (payload + headers, inclusive poe-tag-id). # A request real é “fulfill(200)” (não envia de verdade). # 2) No main(), copiamos cookies/headers críticos de sessão (formkey/revision/tchannel) e # fazemos o POST manual com: # - json = payload CAPTURADO (imutável) # - header poe-tag-id = o capturado naquela mesma intercept # 3) Recebemos a resposta e abrimos o WS (ou longpoll) para obter o texto. # # Nota: Este código usa Playwright para capturar o payload “correto” da build atual, # evitando ter que reproduzir o cálculo do poe-tag-id. import asyncio import json import re from typing import Optional, Dict, Any, List, Tuple import urllib.parse import random import string import requests import websockets from playwright.async_api import async_playwright # ============================================================================== # 0) CONFIG – ajuste seus cookies/sessão aqui # ============================================================================== P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b) P_LAT_COOKIE = "" # opcional CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w" CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog" P_SB = "" # Se quiser forçar um canal específico, preencha; caso contrário será capturado POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm" BOT_NAME = "GPT-5" # slug visível na URL PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar POE_URL_ROOT = "https://poe.com" POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST" RECV_URL = f"{POE_URL_ROOT}/api/receive_POST" BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", "Accept": "*/*", "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8", "Origin": POE_URL_ROOT, "Referer": f"{POE_URL_ROOT}/{BOT_NAME}", "Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"', "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Windows"', "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Content-Type": "application/json", "poegraphql": "0", # string mesmo } # ============================== UTILS ========================================= def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str: return "; ".join(f"{c.name}={c.value}" for c in jar) def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]: for c in jar: name = c.name.lower() if "tchannel" in name and "channel" in name: return c.value return None def ws_url_from_tchannel(channel: str) -> str: m = re.search(r"chan(\d+)-", channel) if not m: raise ValueError(f"tchannel inesperado: {channel}") shard = m.group(1) return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}" def parse_cookie_header_and_set(session: requests.Session, cookie_header: str): """ Lê um header "Cookie: a=b; c=d" capturado do browser e aplica os cookies no session jar. Útil para garantir tchannel-cookie e demais sinalizadores. """ if not cookie_header: return parts = [p.strip() for p in cookie_header.split(";") if "=" in p] for part in parts: try: k, v = part.split("=", 1) session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/") except Exception: continue # -------------------- obter payload-base/headers da build atual ---------------- SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]: """ 1) Abre a página do bot com Playwright com a sua sessão. 2) Digita prompt_for_capture e intercepta a sendMessageMutation exata (payload + headers). - fulfill(200) para NÃO enviar de verdade (guarda uma cópia fiel). 3) Coleta formkey, revision, tag-id, sdid e tchannel desta MESMA sessão/build. 4) Tenta ler o hash do chatMessagesSubscription nos bundles JS (fallback p/ WS). """ async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey via função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas revision_meta = None tag_id_meta = None try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: pass try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: pass # sdid do localStorage (opcional) sdid_local = None try: sdid_local = await page.evaluate(""" (() => { try { for (const k of Object.keys(localStorage)) { if (/sdid/i.test(k)) { const v = localStorage.getItem(k); if (v && v.length >= 10) return v; } } } catch (e) {} return null; })() """) except Exception: pass # tentar extrair os hashes do bundle (WS subscription) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = SEND_RE.search(txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = SUBS_RE.search(txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Intercepta gql_POST e captura a sendMessageMutation gerada para ESTE prompt captured: Dict[str, Any] = { "payload": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in body: captured["req_headers"] = {**captured["req_headers"], **hdrs} if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload"] = payload # payload FINAL p/ ESTE prompt captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade — mas guardar tudo await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real (gera headers/payload corretos para ele) try: # Poe usa textarea ou contenteditable, depende da build. locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda a captura (até ~20s) for _ in range(100): if captured["payload"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais (headers/payload sempre vencem o que veio de meta) final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_base = captured["payload"] hash_intercept = captured["hash"] if not payload_base: raise RuntimeError("Não foi possível capturar a sendMessageMutation para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "sdid_local": sdid_local, "payload_base": payload_base, # PAYLOAD EXATO "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS } # -------------------- helpers de parsing de eventos ---------------------------- def extract_text_pieces(event: Any) -> List[str]: out: List[str] = [] if isinstance(event, dict): for k in ("text_new", "response", "delta", "text"): v = event.get(k) if isinstance(v, str) and v: out.append(v) for v in event.values(): if isinstance(v, (dict, list)): out.extend(extract_text_pieces(v)) elif isinstance(event, list): for x in event: out.extend(extract_text_pieces(x)) return out def is_final_event(event: Dict[str, Any]) -> bool: data = event.get("data") or {} cat = (event.get("category") or "").lower() state = (data.get("state") or data.get("message_state") or "").lower() if "completed" in cat: return True if state in {"complete", "completed", "stopped", "aborted"}: return True if data.get("is_final") is True: return True return False # ============================== RECEPÇÃO ======================================= async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str: tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if not tchannel: raise RuntimeError("Sem poe-tchannel (cookie/forçado).") ws_url = ws_url_from_tchannel(tchannel) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full async def listen_via_longpoll(session: requests.Session) -> str: acc = "" while True: r = session.post(RECV_URL, json=[]) r.raise_for_status() events = r.json() if not isinstance(events, list): continue for ev in events: pieces = extract_text_pieces(ev) if pieces: txt = "".join(pieces) print(txt, end="", flush=True) acc += txt if is_final_event(ev): print("\n--- Resposta Completa (receive_POST) ---") return acc # ============================== DIAGNÓSTICO ==================================== def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]: vars_obj = (payload_base or {}).get("variables") or {} exts_obj = (payload_base or {}).get("extensions") or {} sdid = vars_obj.get("sdid") bot = vars_obj.get("bot") hashv = exts_obj.get("hash") hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…" krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]} ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items()) return hsum, ksum # ================================= MAIN ======================================== async def main(): session = requests.Session() session.headers.update(BASE_HEADERS) # Cookies base session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # Bootstrap — captura payload/headers EXATOS para ESTE PROMPT info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_capturado = info.get("payload_base") req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} if not isinstance(payload_capturado, dict) or payload_capturado.get("queryName") != "sendMessageMutation": raise RuntimeError("Não capturei a sendMessageMutation para o prompt.") # Copiar cookies do header da request interceptada (garante cookie de canal etc.) parse_cookie_header_and_set(session, info.get("cookie_header") or "") # Extrair e aplicar headers críticos de sessão formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK subs_hash = info.get("sub_hash") if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel # manter somente o cookie de tchannel "oficial" session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG compacto p_sum, h_sum = summarize_payload_and_headers(payload_capturado, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # ======================= ENVIO DO POST ================================= print("\nEnviando a pergunta via GraphQL com payload CAPTURADO…") # Cabeçalhos por requisição: copiar alguns do intercept gql_headers = dict(session.headers) # O poe-tag-id deve ser o MESMO que veio no intercept (para ESTE payload) tag_id_hdr = info.get("tag_id") if tag_id_hdr: gql_headers["poe-tag-id"] = tag_id_hdr # Copiar demais cabeçalhos informativos do Apollo/trace, se existirem copy_allow = { "poe-queryname", "apollographql-client-name", "apollographql-client-version", "x-apollo-operation-name", "x-apollo-operation-id", "sentry-trace", "baggage", "poegraphql", "referer", "origin" } for k, v in req_headers_cap.items(): if k in copy_allow: gql_headers[k] = v # Garantir nome da operação e origem coerentes gql_headers.setdefault("poe-queryname", "sendMessageMutation") gql_headers["Referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["Origin"] = POE_URL_ROOT # IMPORTANTE: NÃO ALTERAR O PAYLOAD gql_payload = payload_capturado resp = session.post(POE_API_URL, json=gql_payload, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() # Suporte a messageCreate e messageEdgeCreate root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # ======================= RECEPÇÃO DA RESPOSTA ========================== if subs_hash and re.fullmatch(r"[0-9a-f]{64}", subs_hash or ""): try: final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) except Exception as ws_err: print(f"[WS falhou: {ws_err}] -> tentando receive_POST…") final_answer = await listen_via_longpoll(session) else: final_answer = await listen_via_longpoll(session) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") try: print("poe-queryname usado:", session.headers.get("poe-queryname"), "(se setado no session)") except Exception: pass except Exception as e: print(f"Erro: {e}") if __name__ == "__main__": asyncio.run(main())
dicas finais (resumo do porquê antes dava 400):
variables.query
e clientNonce
, mas mantinha o poe-tag-id
capturado — a assinatura não batia → 400 Bad Request
.get_bootstrap
gera a mutation exata para o seu prompt (com o poe-tag-id
que bate com aquele body) e a gente reenvia idêntica via requests
.formkey:16318c… tag_hdr:dcae18f2… tchannel_hdr:poe-chan102-8888-a… tchannel_cookie:poe-chan102-8888-a… sub_hash:n/a… revision:e2acc7025b…
payload_base: bot:GPT-5… sdid:53015a42-2f4… hash:c877ecaa…
req_hdrs: formkey:16318c91334c… revision:e2acc7025b43… tag:dcae18f272bd… tchannel:poe-chan102-… queryname:sendMessageM… poegraphql:0
Enviando a pergunta via GraphQL com payload CAPTURADO…
HTTPError: 400 Client Error: Bad Request for url: https://poe.com/api/gql_POST
Body: {"success":false,"message":"Bad Request"}
poe-queryname usado: None (se setado no session)
show! Aqui vai a versão completa ajustada para a Opção A — agora o get_bootstrap(...)
aceita prompt_for_capture
, digita esse prompt na UI, captura o corpo bruto da sendMessageMutation
+ todos os headers reais (incluindo poe-tag-id
) e, no main()
, eu reenvio exatamente o mesmo body string e o mesmo conjunto de headers (com pequenas limpezas), além de forçar o mesmo Cookie
header capturado — isso elimina diferenças de ordenação/composição de cookies e evita 400 por assinatura inválida.
Também incluí, como comentários no topo, as tuas “Observações rápidas” e apliquei no código (removi o poe-channel-channel
, mantive poegraphql: "0"
, etc.).
Dica: o 400 que você ainda viu costuma ser por qualquer mínima diferença no que entra no cálculo do
poe-tag-id
(JSON serializado diferente, cookies em outra ordem, etc.). Por isso agora usamos o body literal e o Cookie literal da request interceptada.
python# -*- coding: utf-8 -*- # Poe GraphQL sender — Opção A (reenvio 1:1 do payload/headers capturados) # # Observações rápidas (importante): # - poe-tag-id muda a cada chamada. Se você capturar um e DEPOIS mudar query/clientNonce, # vai tomar 400 (assinatura não confere). # - poegraphql: "0" ou "1" como string funciona; aqui uso "0" conforme o front capturou. # - Evite setar "poe-channel-channel" (parece typo). Só "poe-tchannel-channel" já basta. # - Se algum cookie do Cloudflare expirar, o erro típico é 403/401, não 400. # # Estratégia: # 1) get_bootstrap(bot, cookies, prompt) abre a página logada, digita o PROMPT e # intercepta a sendMessageMutation EXATA (headers + body RAW, inclusive poe-tag-id). # A request real é “fulfill(200)” (não envia), mas guardamos uma cópia fiel. # 2) No main(), enviamos via requests: # - headers = quase os MESMOS headers da request interceptada (limpando hop-by-hop) # - data = body RAW capturado (string idêntica) # - Cookie = exatamente o cookie header capturado (para manter ordem/valores) # 3) Consumimos a resposta por WS ou receive_POST. import asyncio import json import re from typing import Optional, Dict, Any, List, Tuple import urllib.parse import requests import websockets from playwright.async_api import async_playwright # ============================================================================== # 0) CONFIG – ajuste seus cookies/sessão aqui # ============================================================================== P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b) P_LAT_COOKIE = "" # opcional CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w" CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog" P_SB = "" # Se quiser forçar um canal específico, preencha; caso contrário será capturado POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm" BOT_NAME = "GPT-5" # slug visível na URL PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar POE_URL_ROOT = "https://poe.com" POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST" RECV_URL = f"{POE_URL_ROOT}/api/receive_POST" BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", "Accept": "*/*", "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8", "Origin": POE_URL_ROOT, "Referer": f"{POE_URL_ROOT}/{BOT_NAME}", "Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"', "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Windows"', "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Content-Type": "application/json", "poegraphql": "0", # string mesmo } # ============================== UTILS ========================================= def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str: return "; ".join(f"{c.name}={c.value}" for c in jar) def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]: for c in jar: name = c.name.lower() if "tchannel" in name and "channel" in name: return c.value return None def ws_url_from_tchannel(channel: str) -> str: m = re.search(r"chan(\d+)-", channel) if not m: raise ValueError(f"tchannel inesperado: {channel}") shard = m.group(1) return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}" def parse_cookie_header_and_set(session: requests.Session, cookie_header: str): """Aplica um Cookie header literal no cookie jar (para uso posterior no WS/receive_POST).""" if not cookie_header: return parts = [p.strip() for p in cookie_header.split(";") if "=" in p] for part in parts: try: k, v = part.split("=", 1) session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/") except Exception: continue # -------------------- obter payload-base/headers da build atual ---------------- SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]: """ Abre a página logada, digita o prompt e intercepta a sendMessageMutation EXATA. Retorna: formkey, revision, tchannel, poe-tag-id, headers capturados, body RAW e o JSON parseado. """ async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey via função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas revision_meta = None tag_id_meta = None try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: pass try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: pass # sdid do localStorage (opcional) try: sdid_local = await page.evaluate(""" (() => { try { for (const k of Object.keys(localStorage)) { if (/sdid/i.test(k)) { const v = localStorage.getItem(k); if (v && v.length >= 10) return v; } } } catch (e) {} return null; })() """) except Exception: sdid_local = None # tentar extrair hash da subscription (WS) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = SEND_RE.search(txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = SUBS_RE.search(txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Captura da request real captured: Dict[str, Any] = { "payload_dict": None, # JSON parseado (opcional) "payload_raw": None, # BODY RAW (string exata) "hash": None, "req_headers": {}, # headers em lowercase "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in (body or ""): captured["req_headers"] = {**captured["req_headers"], **hdrs} captured["payload_raw"] = body if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload_dict"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade — mas guardar tudo await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real try: # Poe varia entre textarea e contenteditable locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda a captura (até ~20s) for _ in range(100): if captured["payload_raw"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_dict = captured["payload_dict"] payload_raw = captured["payload_raw"] hash_intercept = captured["hash"] if not payload_raw: raise RuntimeError("Não foi possível capturar a sendMessageMutation (body RAW) para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "payload_base": payload_dict, # JSON parseado (apenas para logs) "payload_raw": payload_raw, # BODY STRING EXATA (usar no POST) "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS (lowercase) } # -------------------- helpers de parsing de eventos ---------------------------- def extract_text_pieces(event: Any) -> List[str]: out: List[str] = [] if isinstance(event, dict): for k in ("text_new", "response", "delta", "text"): v = event.get(k) if isinstance(v, str) and v: out.append(v) for v in event.values(): if isinstance(v, (dict, list)): out.extend(extract_text_pieces(v)) elif isinstance(event, list): for x in event: out.extend(extract_text_pieces(x)) return out def is_final_event(event: Dict[str, Any]) -> bool: data = event.get("data") or {} cat = (event.get("category") or "").lower() state = (data.get("state") or data.get("message_state") or "").lower() if "completed" in cat: return True if state in {"complete", "completed", "stopped", "aborted"}: return True if data.get("is_final") is True: return True return False # ============================== RECEPÇÃO ======================================= async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str: tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if not tchannel: raise RuntimeError("Sem poe-tchannel (cookie/forçado).") ws_url = ws_url_from_tchannel(tchannel) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full async def listen_via_longpoll(session: requests.Session) -> str: acc = "" while True: r = session.post(RECV_URL, json=[]) r.raise_for_status() events = r.json() if not isinstance(events, list): continue for ev in events: pieces = extract_text_pieces(ev) if pieces: txt = "".join(pieces) print(txt, end="", flush=True) acc += txt if is_final_event(ev): print("\n--- Resposta Completa (receive_POST) ---") return acc # ============================== DIAGNÓSTICO ==================================== def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]: vars_obj = (payload_base or {}).get("variables") or {} exts_obj = (payload_base or {}).get("extensions") or {} sdid = vars_obj.get("sdid") bot = vars_obj.get("bot") hashv = exts_obj.get("hash") hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…" krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]} ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items()) return hsum, ksum # ================================= MAIN ======================================== async def main(): session = requests.Session() session.headers.update(BASE_HEADERS) # Cookies base no jar (para WS/receive posteriormente) session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # 1) Captura 1:1 (headers + body RAW) para ESTE PROMPT info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_raw = info.get("payload_raw") # STRING exata payload_capturado= info.get("payload_base") # JSON (apenas para log) req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} if not payload_raw: raise RuntimeError("Não capturei a sendMessageMutation (body RAW).") # Copiar cookies exatos do intercept para o jar (WS/receive) e, no POST, usar Cookie literal captured_cookie_header = info.get("cookie_header") or "" parse_cookie_header_and_set(session, captured_cookie_header) # Extrair headers críticos de sessão formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK subs_hash = info.get("sub_hash") # Guardar alguns no session (para WS/receive), mas no POST vamos usar headers capturados if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG compacto p_sum, h_sum = summarize_payload_and_headers(payload_capturado or {}, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # 2) Envio 1:1 — headers quase idênticos + body RAW + Cookie literal print("\nEnviando a pergunta via GraphQL com payload/headers CAPTURADOS…") # Comece dos headers capturados e limpe os que a requests deve gerir blocked = { "content-length", "host", "connection", "accept-encoding", # hop-by-hop # qualquer header que o requests/com TLS queira definir sozinho } gql_headers = {k: v for k, v in req_headers_cap.items() if k not in blocked} # Garanta Content-Type correto (caso o frontend mude) gql_headers["content-type"] = "application/json" # Garanta Referer/Origin coerentes gql_headers["referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["origin"] = POE_URL_ROOT # IMPORTANTÍSSIMO: usar o Cookie literal capturado (ordem/valores idênticos) if captured_cookie_header: gql_headers["cookie"] = captured_cookie_header # Faça o POST usando o BODY BRUTO, não json= (evita mudanças de serialização) resp = session.post(POE_API_URL, data=payload_raw, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() # Suporte a messageCreate e messageEdgeCreate root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # 3) Recepção da resposta if subs_hash and re.fullmatch(r"[0-9a-f]{64}", subs_hash or ""): try: final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) except Exception as ws_err: print(f"[WS falhou: {ws_err}] -> tentando receive_POST…") final_answer = await listen_via_longpoll(session) else: final_answer = await listen_via_longpoll(session) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") # Dica de fallback: se ainda vier 400, remova o fulfill e deixe o front enviar de verdade (Opção B). except Exception as e: print(f"Erro: {e}") if __name__ == "__main__": asyncio.run(main())
Se mesmo assim estourar 400, é quase certo que o poe-tag-id
também checa algo efêmero (p.ex. timestamp) e você está repostando tarde demais. Aí o plano B é simples: tire o route.fulfill
na captura e deixe o front postar de verdade (sem replay); depois só consome pelo WS/receive_POST
— seu listener já está pronto pra isso.
erro: formkey:16318c… tag_hdr:85a246e5… tchannel_hdr:poe-chan120-8888-l… tchannel_cookie:poe-chan120-8888-l… sub_hash:n/a… revision:e2acc7025b…
payload_base: bot:GPT-5… sdid:ed8d1d08-673… hash:c877ecaa…
req_hdrs: formkey:16318c91334c… revision:e2acc7025b43… tag:85a246e5a25c… tchannel:poe-chan120-… queryname:sendMessageM… poegraphql:0
Enviando a pergunta via GraphQL com payload/headers CAPTURADOS…
Chat criado! ID: 1248116027
HTTPError: 400 Client Error: Bad Request for url: https://poe.com/api/receive_POST
Body: {"success":false,"message":"Bad Request"}" código atual:
import asyncio
import json
import re
from typing import Optional, Dict, Any, List, Tuple
import urllib.parse
import requests
import websockets
from playwright.async_api import async_playwright
P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b)
P_LAT_COOKIE = "" # opcional
CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w"
CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog"
P_SB = ""
POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm"
BOT_NAME = "GPT-5" # slug visível na URL
PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar
POE_URL_ROOT = "https://poe.com"
POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST"
RECV_URL = f"{POE_URL_ROOT}/api/receive_POST"
BASE_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0",
"Accept": "/",
"Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8",
"Origin": POE_URL_ROOT,
"Referer": f"{POE_URL_ROOT}/{BOT_NAME}",
"Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Content-Type": "application/json",
"poegraphql": "0", # string mesmo
}
def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str:
return "; ".join(f"{c.name}={c.value}" for c in jar)
def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]:
for c in jar:
name = c.name.lower()
if "tchannel" in name and "channel" in name:
return c.value
return None
def ws_url_from_tchannel(channel: str) -> str:
m = re.search(r"chan(\d+)-", channel)
if not m:
raise ValueError(f"tchannel inesperado: {channel}")
shard = m.group(1)
return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}"
def parse_cookie_header_and_set(session: requests.Session, cookie_header: str):
"""Aplica um Cookie header literal no cookie jar (para uso posterior no WS/receive_POST)."""
if not cookie_header:
return
parts = [p.strip() for p in cookie_header.split(";") if "=" in p]
for part in parts:
try:
k, v = part.split("=", 1)
session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/")
except Exception:
continue
SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]:
"""
Abre a página logada, digita o prompt e intercepta a sendMessageMutation EXATA.
Retorna: formkey, revision, tchannel, poe-tag-id, headers capturados, body RAW e o JSON parseado.
"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
text# Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey via função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas revision_meta = None tag_id_meta = None try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: pass try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: pass # sdid do localStorage (opcional) try: sdid_local = await page.evaluate(""" (() => { try { for (const k of Object.keys(localStorage)) { if (/sdid/i.test(k)) { const v = localStorage.getItem(k); if (v && v.length >= 10) return v; } } } catch (e) {} return null; })() """) except Exception: sdid_local = None # tentar extrair hash da subscription (WS) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = SEND_RE.search(txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = SUBS_RE.search(txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Captura da request real captured: Dict[str, Any] = { "payload_dict": None, # JSON parseado (opcional) "payload_raw": None, # BODY RAW (string exata) "hash": None, "req_headers": {}, # headers em lowercase "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in (body or ""): captured["req_headers"] = {**captured["req_headers"], **hdrs} captured["payload_raw"] = body if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload_dict"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade — mas guardar tudo await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real try: # Poe varia entre textarea e contenteditable locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda a captura (até ~20s) for _ in range(100): if captured["payload_raw"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_dict = captured["payload_dict"] payload_raw = captured["payload_raw"] hash_intercept = captured["hash"] if not payload_raw: raise RuntimeError("Não foi possível capturar a sendMessageMutation (body RAW) para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "payload_base": payload_dict, # JSON parseado (apenas para logs) "payload_raw": payload_raw, # BODY STRING EXATA (usar no POST) "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS (lowercase) }
def extract_text_pieces(event: Any) -> List[str]:
out: List[str] = []
if isinstance(event, dict):
for k in ("text_new", "response", "delta", "text"):
v = event.get(k)
if isinstance(v, str) and v:
out.append(v)
for v in event.values():
if isinstance(v, (dict, list)):
out.extend(extract_text_pieces(v))
elif isinstance(event, list):
for x in event:
out.extend(extract_text_pieces(x))
return out
def is_final_event(event: Dict[str, Any]) -> bool:
data = event.get("data") or {}
cat = (event.get("category") or "").lower()
state = (data.get("state") or data.get("message_state") or "").lower()
if "completed" in cat:
return True
if state in {"complete", "completed", "stopped", "aborted"}:
return True
if data.get("is_final") is True:
return True
return False
async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str:
tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK
if not tchannel:
raise RuntimeError("Sem poe-tchannel (cookie/forçado).")
textws_url = ws_url_from_tchannel(tchannel) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full
async def listen_via_longpoll(session: requests.Session) -> str:
acc = ""
while True:
r = session.post(RECV_URL, json=[])
r.raise_for_status()
events = r.json()
if not isinstance(events, list):
continue
for ev in events:
pieces = extract_text_pieces(ev)
if pieces:
txt = "".join(pieces)
print(txt, end="", flush=True)
acc += txt
if is_final_event(ev):
print("\n--- Resposta Completa (receive_POST) ---")
return acc
def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]:
vars_obj = (payload_base or {}).get("variables") or {}
exts_obj = (payload_base or {}).get("extensions") or {}
sdid = vars_obj.get("sdid")
bot = vars_obj.get("bot")
hashv = exts_obj.get("hash")
hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…"
krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]}
ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items())
return hsum, ksum
async def main():
session = requests.Session()
session.headers.update(BASE_HEADERS)
text# Cookies base no jar (para WS/receive posteriormente) session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # 1) Captura 1:1 (headers + body RAW) para ESTE PROMPT info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_raw = info.get("payload_raw") # STRING exata payload_capturado= info.get("payload_base") # JSON (apenas para log) req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} if not payload_raw: raise RuntimeError("Não capturei a sendMessageMutation (body RAW).") # Copiar cookies exatos do intercept para o jar (WS/receive) e, no POST, usar Cookie literal captured_cookie_header = info.get("cookie_header") or "" parse_cookie_header_and_set(session, captured_cookie_header) # Extrair headers críticos de sessão formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK subs_hash = info.get("sub_hash") # Guardar alguns no session (para WS/receive), mas no POST vamos usar headers capturados if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG compacto p_sum, h_sum = summarize_payload_and_headers(payload_capturado or {}, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # 2) Envio 1:1 — headers quase idênticos + body RAW + Cookie literal print("\nEnviando a pergunta via GraphQL com payload/headers CAPTURADOS…") # Comece dos headers capturados e limpe os que a requests deve gerir blocked = { "content-length", "host", "connection", "accept-encoding", # hop-by-hop # qualquer header que o requests/com TLS queira definir sozinho } gql_headers = {k: v for k, v in req_headers_cap.items() if k not in blocked} # Garanta Content-Type correto (caso o frontend mude) gql_headers["content-type"] = "application/json" # Garanta Referer/Origin coerentes gql_headers["referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["origin"] = POE_URL_ROOT # IMPORTANTÍSSIMO: usar o Cookie literal capturado (ordem/valores idênticos) if captured_cookie_header: gql_headers["cookie"] = captured_cookie_header # Faça o POST usando o BODY BRUTO, não json= (evita mudanças de serialização) resp = session.post(POE_API_URL, data=payload_raw, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() # Suporte a messageCreate e messageEdgeCreate root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # 3) Recepção da resposta if subs_hash and re.fullmatch(r"[0-9a-f]{64}", subs_hash or ""): try: final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) except Exception as ws_err: print(f"[WS falhou: {ws_err}] -> tentando receive_POST…") final_answer = await listen_via_longpoll(session) else: final_answer = await listen_via_longpoll(session) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") # Dica de fallback: se ainda vier 400, remova o fulfill e deixe o front enviar de verdade (Opção B). except Exception as e: print(f"Erro: {e}")
if name == "main":
asyncio.run(main())
Boa! O sendMessageMutation
funcionou (“Chat criado!”), então o gargalo agora é o receive_POST 400. Isso rola porque o receive_POST
também usa assinatura por poe-tag-id
dependente do body; mandar []
com um tag de outro payload cai em 400. A melhor saída é não usar receive_POST
e puxar tudo via WebSocket com a hash correta da chatMessagesSubscription
.
Abaixo vai o código completo atualizado para:
chatMessagesSubscription
direto dos frames WS que a própria página envia (sem depender de ler bundles JS).receive_POST
.Também mantive e apliquei tuas “Observações rápidas” nos comentários do topo.
python# -*- coding: utf-8 -*- # Poe GraphQL sender — Opção A (reenvio 1:1) + WebSocket hash sniff # # Observações rápidas: # - poe-tag-id muda a cada chamada. Se capturar e depois mudar query/clientNonce, # 400 na certa (assinatura não confere). # - poegraphql: "0" ou "1" como string funciona. # - Evite setar "poe-channel-channel" (typo). Só "poe-tchannel-channel" já basta. # - Se algum cookie do Cloudflare expirar, o erro típico é 403/401, não 400. # # Estratégia: # 1) get_bootstrap(bot, cookies, prompt): abre a página logada, digita o PROMPT e # intercepta a sendMessageMutation EXATA (headers + body RAW) e, EM PARALELO, # “fareja” (sniff) os frames de WebSocket enviados pela página para extrair a # hash da chatMessagesSubscription. # 2) No main(), reenvia o body RAW (headers quase idênticos) -> obtém chatId real. # 3) Conecta ao tch WebSocket e assina usando a hash capturada -> lê a resposta. # import asyncio import json import re from typing import Optional, Dict, Any, List, Tuple import urllib.parse import requests import websockets from playwright.async_api import async_playwright # ============================================================================== # 0) CONFIG – ajuste seus cookies/sessão aqui # ============================================================================== P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b) P_LAT_COOKIE = "" # opcional CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w" CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog" P_SB = "" # Se quiser forçar um canal específico, preencha; caso contrário será capturado POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm" BOT_NAME = "GPT-5" # slug visível na URL PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar POE_URL_ROOT = "https://poe.com" POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST" BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", "Accept": "*/*", "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8", "Origin": POE_URL_ROOT, "Referer": f"{POE_URL_ROOT}/{BOT_NAME}", "Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"', "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Windows"', "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Content-Type": "application/json", "poegraphql": "0", # string mesmo } # ============================== UTILS ========================================= def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str: return "; ".join(f"{c.name}={c.value}" for c in jar) def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]: for c in jar: name = c.name.lower() if "tchannel" in name and "channel" in name: return c.value return None def ws_url_from_tchannel(channel: str) -> str: m = re.search(r"chan(\d+)-", channel) if not m: raise ValueError(f"tchannel inesperado: {channel}") shard = m.group(1) return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}" def parse_cookie_header_and_set(session: requests.Session, cookie_header: str): """Aplica um Cookie header literal no cookie jar (para uso posterior no WS).""" if not cookie_header: return parts = [p.strip() for p in cookie_header.split(";") if "=" in p] for part in parts: try: k, v = part.split("=", 1) session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/") except Exception: continue # -------------------- obter hashes (fallback por bundle — mantido, mas não obrigatório) --- SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"') # ============================== BOOTSTRAP ====================================== async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]: """ Abre a página logada, digita o prompt e intercepta: - sendMessageMutation EXATA (headers + body RAW) - hash da chatMessagesSubscription farejada de frames WS enviados pela página """ async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() # -------- farejar hash da subscription em frames WS -------- captured_ws = {"sub_hash_ws": None} def on_websocket(ws): # Queremos frames ENVIADOS pelo cliente (framesent) — é quando ele manda o "subscriptions" def sniff(frame_text: str): try: data = json.loads(frame_text) except Exception: return if isinstance(data, dict) and data.get("type") == "subscriptions": subs = data.get("subscriptions") or {} if isinstance(subs, dict): for v in subs.values(): if isinstance(v, dict) and v.get("queryName") == "chatMessagesSubscription": h = v.get("hash") if isinstance(h, str) and re.fullmatch(r"[0-9a-f]{64}", h): captured_ws["sub_hash_ws"] = h # nada mais a fazer try: ws.on("framesent", lambda f: sniff(f)) except Exception: pass page.on("websocket", on_websocket) await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey (global ou __NEXT_DATA__) formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: revision_meta = None try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: tag_id_meta = None # tentar extrair hashes por bundle (fallback, caso WS não pegue) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = re.search(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"', txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = re.search(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"', txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Intercepta a sendMessageMutation (headers + body RAW) captured = { "payload_dict": None, "payload_raw": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in (body or ""): captured["req_headers"] = {**captured["req_headers"], **hdrs} captured["payload_raw"] = body if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload_dict"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real try: locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda captura do body e (em paralelo) da sub_hash via WS (até ~20s) for _ in range(100): if captured["payload_raw"] and captured_ws["sub_hash_ws"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_dict = captured["payload_dict"] payload_raw = captured["payload_raw"] hash_intercept = captured["hash"] sub_hash_ws = captured_ws["sub_hash_ws"] sub_hash_final = sub_hash_ws or subs_hash_bundles # preferir WS; cair p/ bundle se preciso if not payload_raw: raise RuntimeError("Não foi possível capturar a sendMessageMutation (body RAW) para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": sub_hash_final, # <- agora vem do WS sniff (ou bundle fallback) "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "payload_base": payload_dict, # JSON parseado (apenas para log) "payload_raw": payload_raw, # BODY STRING EXATA (usar no POST) "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS (lowercase) } # ============================== STREAM PARSING ================================= def extract_text_pieces(event: Any) -> List[str]: out: List[str] = [] if isinstance(event, dict): for k in ("text_new", "response", "delta", "text"): v = event.get(k) if isinstance(v, str) and v: out.append(v) for v in event.values(): if isinstance(v, (dict, list)): out.extend(extract_text_pieces(v)) elif isinstance(event, list): for x in event: out.extend(extract_text_pieces(x)) return out def is_final_event(event: Dict[str, Any]) -> bool: data = event.get("data") or {} cat = (event.get("category") or "").lower() state = (data.get("state") or data.get("message_state") or "").lower() if "completed" in cat: return True if state in {"complete", "completed", "stopped", "aborted"}: return True if data.get("is_final") is True: return True return False # ============================== RECEPÇÃO VIA WS ================================= async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str: tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if not tchannel: raise RuntimeError("Sem poe-tchannel (cookie/forçado).") ws_url = ws_url_from_tchannel(tchannel) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) # Assina explicitamente a subscription p/ ESTE chatId real payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full # ============================== DIAGNÓSTICO ==================================== def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]: vars_obj = (payload_base or {}).get("variables") or {} exts_obj = (payload_base or {}).get("extensions") or {} sdid = vars_obj.get("sdid") bot = vars_obj.get("bot") hashv = exts_obj.get("hash") hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…" krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]} ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items()) return hsum, ksum # ================================= MAIN ======================================== async def main(): session = requests.Session() session.headers.update(BASE_HEADERS) # Cookies base no jar (para WS posteriormente) session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # 1) Captura 1:1 e hash da subscription via WS sniff info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_raw = info.get("payload_raw") # STRING exata payload_capturado = info.get("payload_base") # JSON (apenas para log) req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} subs_hash = info.get("sub_hash") if not payload_raw: raise RuntimeError("Não capturei a sendMessageMutation (body RAW).") # Cookies exatos do intercept no jar (para WS) captured_cookie_header = info.get("cookie_header") or "" parse_cookie_header_and_set(session, captured_cookie_header) # Headers críticos de sessão para WS formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG p_sum, h_sum = summarize_payload_and_headers(payload_capturado or {}, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # 2) Envio 1:1 — headers quase idênticos + body RAW + Cookie literal print("\nEnviando a pergunta via GraphQL com payload/headers CAPTURADOS…") blocked = {"content-length", "host", "connection", "accept-encoding"} gql_headers = {k: v for k, v in req_headers_cap.items() if k not in blocked} gql_headers["content-type"] = "application/json" gql_headers["referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["origin"] = POE_URL_ROOT if captured_cookie_header: gql_headers["cookie"] = captured_cookie_header resp = session.post(POE_API_URL, data=payload_raw, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # 3) Recepção via WS (sem receive_POST) if not subs_hash or not re.fullmatch(r"[0-9a-f]{64}", subs_hash): raise RuntimeError("Não consegui obter a hash da chatMessagesSubscription via WS/bundle.") final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") except Exception as e: print(f"Erro: {e}") if __name__ == "__main__": asyncio.run(main())
receive_POST
)receive_POST
(que exige poe-tag-id
calculado em cima do body de cada chamada), a gente captura a hash da subscription diretamente dos frames WebSocket do próprio front e ouve tudo por WS, que não precisa do poe-tag-id
.sendMessageMutation
(headers + corpo RAW + Cookie
literal), que já comprovou funcionar com o seu chatId real.Erro: Não consegui obter a hash da chatMessagesSubscription via WS/bundle. Mande apenas o código completo em seu output com o problema corrigido. Código atual: # -- coding: utf-8 --
import asyncio
import json
import re
from typing import Optional, Dict, Any, List, Tuple
import urllib.parse
import requests
import websockets
from playwright.async_api import async_playwright
P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b)
P_LAT_COOKIE = "" # opcional
CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w"
CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog"
P_SB = ""
POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm"
BOT_NAME = "GPT-5" # slug visível na URL
PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar
POE_URL_ROOT = "https://poe.com"
POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST"
BASE_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0",
"Accept": "/",
"Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8",
"Origin": POE_URL_ROOT,
"Referer": f"{POE_URL_ROOT}/{BOT_NAME}",
"Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Content-Type": "application/json",
"poegraphql": "0", # string mesmo
}
def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str:
return "; ".join(f"{c.name}={c.value}" for c in jar)
def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]:
for c in jar:
name = c.name.lower()
if "tchannel" in name and "channel" in name:
return c.value
return None
def ws_url_from_tchannel(channel: str) -> str:
m = re.search(r"chan(\d+)-", channel)
if not m:
raise ValueError(f"tchannel inesperado: {channel}")
shard = m.group(1)
return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}"
def parse_cookie_header_and_set(session: requests.Session, cookie_header: str):
"""Aplica um Cookie header literal no cookie jar (para uso posterior no WS)."""
if not cookie_header:
return
parts = [p.strip() for p in cookie_header.split(";") if "=" in p]
for part in parts:
try:
k, v = part.split("=", 1)
session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/")
except Exception:
continue
SEND_RE = re.compile(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"')
async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]:
"""
Abre a página logada, digita o prompt e intercepta:
- sendMessageMutation EXATA (headers + body RAW)
- hash da chatMessagesSubscription farejada de frames WS enviados pela página
"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
text# Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() # -------- farejar hash da subscription em frames WS -------- captured_ws = {"sub_hash_ws": None} def on_websocket(ws): # Queremos frames ENVIADOS pelo cliente (framesent) — é quando ele manda o "subscriptions" def sniff(frame_text: str): try: data = json.loads(frame_text) except Exception: return if isinstance(data, dict) and data.get("type") == "subscriptions": subs = data.get("subscriptions") or {} if isinstance(subs, dict): for v in subs.values(): if isinstance(v, dict) and v.get("queryName") == "chatMessagesSubscription": h = v.get("hash") if isinstance(h, str) and re.fullmatch(r"[0-9a-f]{64}", h): captured_ws["sub_hash_ws"] = h # nada mais a fazer try: ws.on("framesent", lambda f: sniff(f)) except Exception: pass page.on("websocket", on_websocket) await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey (global ou __NEXT_DATA__) formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: revision_meta = None try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: tag_id_meta = None # tentar extrair hashes por bundle (fallback, caso WS não pegue) send_hash_bundles = None subs_hash_bundles = None try: script_urls = await page.evaluate('Array.from(document.scripts).map(s=>s.src).filter(Boolean)') except Exception: script_urls = [] sess_tmp = requests.Session() for url in script_urls: try: if url.startswith("/"): url = urllib.parse.urljoin(POE_URL_ROOT, url) r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text if not send_hash_bundles: m = re.search(r'"sendMessageMutation"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"', txt) if m: send_hash_bundles = m.group(1) if not subs_hash_bundles: m2 = re.search(r'"chatMessagesSubscription"\s*,\s*"hash"\s*:\s*"([0-9a-f]{64})"', txt) if m2: subs_hash_bundles = m2.group(1) if send_hash_bundles and subs_hash_bundles: break except Exception: continue # Intercepta a sendMessageMutation (headers + body RAW) captured = { "payload_dict": None, "payload_raw": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in (body or ""): captured["req_headers"] = {**captured["req_headers"], **hdrs} captured["payload_raw"] = body if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload_dict"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real try: locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda captura do body e (em paralelo) da sub_hash via WS (até ~20s) for _ in range(100): if captured["payload_raw"] and captured_ws["sub_hash_ws"]: break await page.wait_for_timeout(200) try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_dict = captured["payload_dict"] payload_raw = captured["payload_raw"] hash_intercept = captured["hash"] sub_hash_ws = captured_ws["sub_hash_ws"] sub_hash_final = sub_hash_ws or subs_hash_bundles # preferir WS; cair p/ bundle se preciso if not payload_raw: raise RuntimeError("Não foi possível capturar a sendMessageMutation (body RAW) para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": sub_hash_final, # <- agora vem do WS sniff (ou bundle fallback) "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "payload_base": payload_dict, # JSON parseado (apenas para log) "payload_raw": payload_raw, # BODY STRING EXATA (usar no POST) "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS (lowercase) }
def extract_text_pieces(event: Any) -> List[str]:
out: List[str] = []
if isinstance(event, dict):
for k in ("text_new", "response", "delta", "text"):
v = event.get(k)
if isinstance(v, str) and v:
out.append(v)
for v in event.values():
if isinstance(v, (dict, list)):
out.extend(extract_text_pieces(v))
elif isinstance(event, list):
for x in event:
out.extend(extract_text_pieces(x))
return out
def is_final_event(event: Dict[str, Any]) -> bool:
data = event.get("data") or {}
cat = (event.get("category") or "").lower()
state = (data.get("state") or data.get("message_state") or "").lower()
if "completed" in cat:
return True
if state in {"complete", "completed", "stopped", "aborted"}:
return True
if data.get("is_final") is True:
return True
return False
async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str:
tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK
if not tchannel:
raise RuntimeError("Sem poe-tchannel (cookie/forçado).")
textws_url = ws_url_from_tchannel(tchannel) ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) # Assina explicitamente a subscription p/ ESTE chatId real payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full
def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]:
vars_obj = (payload_base or {}).get("variables") or {}
exts_obj = (payload_base or {}).get("extensions") or {}
sdid = vars_obj.get("sdid")
bot = vars_obj.get("bot")
hashv = exts_obj.get("hash")
hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…"
krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]}
ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items())
return hsum, ksum
async def main():
session = requests.Session()
session.headers.update(BASE_HEADERS)
text# Cookies base no jar (para WS posteriormente) session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # 1) Captura 1:1 e hash da subscription via WS sniff info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_raw = info.get("payload_raw") # STRING exata payload_capturado = info.get("payload_base") # JSON (apenas para log) req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} subs_hash = info.get("sub_hash") if not payload_raw: raise RuntimeError("Não capturei a sendMessageMutation (body RAW).") # Cookies exatos do intercept no jar (para WS) captured_cookie_header = info.get("cookie_header") or "" parse_cookie_header_and_set(session, captured_cookie_header) # Headers críticos de sessão para WS formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG p_sum, h_sum = summarize_payload_and_headers(payload_capturado or {}, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # 2) Envio 1:1 — headers quase idênticos + body RAW + Cookie literal print("\nEnviando a pergunta via GraphQL com payload/headers CAPTURADOS…") blocked = {"content-length", "host", "connection", "accept-encoding"} gql_headers = {k: v for k, v in req_headers_cap.items() if k not in blocked} gql_headers["content-type"] = "application/json" gql_headers["referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["origin"] = POE_URL_ROOT if captured_cookie_header: gql_headers["cookie"] = captured_cookie_header resp = session.post(POE_API_URL, data=payload_raw, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # 3) Recepção via WS (sem receive_POST) if not subs_hash or not re.fullmatch(r"[0-9a-f]{64}", subs_hash): raise RuntimeError("Não consegui obter a hash da chatMessagesSubscription via WS/bundle.") final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") except Exception as e: print(f"Erro: {e}")
if name == "main":
asyncio.run(main())
python# -*- coding: utf-8 -*- # Poe GraphQL sender — Opção A (reenvio 1:1) + captura robusta da hash de chatMessagesSubscription # # Observações rápidas: # - poe-tag-id muda a cada chamada. Se capturar e depois mudar query/clientNonce, # 400 na certa (assinatura não confere). # - poegraphql: "0" ou "1" como string funciona. # - Evite setar "poe-channel-channel" (typo). Só "poe-tchannel-channel" já basta. # - Se algum cookie do Cloudflare expirar, o erro típico é 403/401, não 400. # # Estratégia: # 1) get_bootstrap(bot, cookies, prompt): abre a página logada, digita o PROMPT e # intercepta a sendMessageMutation EXATA (headers + body RAW). NÃO enviamos de verdade. # 2) Na mesma função, varremos TODOS os scripts (src e inline) e extraímos a hash da # chatMessagesSubscription com regex tolerante (sem depender de frames WS). # 3) No main(), reenvia o body RAW (headers quase idênticos) -> obtém chatId real. # 4) Conecta ao tch WebSocket e assina usando a hash capturada -> lê a resposta (sem receive_POST). # import asyncio import json import re from typing import Optional, Dict, Any, List, Tuple import urllib.parse import requests import websockets from playwright.async_api import async_playwright # ============================================================================== # 0) CONFIG – ajuste seus cookies/sessão aqui # ============================================================================== P_B_COOKIE = "fnWI711UP20Z-5hhB4qmXA%3D%3D" # obrigatório (cookie de sessão p-b) P_LAT_COOKIE = "" # opcional CF_CLEARANCE = "thwhcHgrOeXgMi7doUzJvgtSt5I4gzXO0m.H49jgyA0-1755601426-1.2.1.1-9uKyi3ob4O0gitbAw3w3FFJTZNDhz2lCDq.BSoTiGToSvGQ9u187hnoJSM3F7a9NAKHZS8jmqWNMxWV1zFM5umeCP7eGv8y474o4fpqWEbu.Mcmyajt6ZM10wtzkvKqmdvkTfYOW4HOZkVFxQoCjDKW9dokcRv.zhGvRA4N6li0tvYZARuAgHLx8SRmiB1jL90FI3i8x5rgxObnHHQEOX80IV7.4w1JJe6ipSveEI8w" CF_BM = "EBefqZ01pvWR0B_MsvkraEXhpn3bBvHkAKEt9sBMJV4-1755601425-1.0.1.1-TLMm3MA2hJZrs2G94tXN7ySKgwjGLDDBj.L5EMp8LeFAPjSurFpNwRyCZgJ1Rj0hjHuTm5yxz8vKWHCB5srAqcKfO49qFPqk4fttffxLoog" P_SB = "" # Se quiser forçar um canal específico, preencha; caso contrário será capturado POE_TCHANNEL_FALLBACK = "poe-chan105-8888-eubnxknfqzvcppphsprm" BOT_NAME = "GPT-5" # slug visível na URL PROMPT = "Olá, tudo bem, qm é voce?" # mensagem a enviar POE_URL_ROOT = "https://poe.com" POE_API_URL = f"{POE_URL_ROOT}/api/gql_POST" BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", "Accept": "*/*", "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.8", "Origin": POE_URL_ROOT, "Referer": f"{POE_URL_ROOT}/{BOT_NAME}", "Sec-Ch-Ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"', "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": '"Windows"', "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Content-Type": "application/json", "poegraphql": "0", # string mesmo } # ============================== UTILS ========================================= def cookie_header_from_jar(jar: requests.cookies.RequestsCookieJar) -> str: return "; ".join(f"{c.name}={c.value}" for c in jar) def get_tchannel_from_cookies(jar: requests.cookies.RequestsCookieJar) -> Optional[str]: for c in jar: name = c.name.lower() if "tchannel" in name and "channel" in name: return c.value return None def ws_url_from_tchannel(channel: str) -> str: m = re.search(r"chan(\d+)-", channel) if not m: raise ValueError(f"tchannel inesperado: {channel}") shard = m.group(1) return f"wss://tch{shard}.poe.com/up/{channel.replace('poe-','')}/updates?channel={channel}" def parse_cookie_header_and_set(session: requests.Session, cookie_header: str): """Aplica um Cookie header literal no cookie jar (para uso posterior no WS).""" if not cookie_header: return parts = [p.strip() for p in cookie_header.split(";") if "=" in p] for part in parts: try: k, v = part.split("=", 1) session.cookies.set(k.strip(), v.strip(), domain="poe.com", path="/") except Exception: continue # -------------------- regex robustas p/ hashes em bundles ---------------------- SEND_RE = re.compile(r'"sendMessageMutation"\s*:\s*\{[^}]*?"hash"\s*:\s*"([0-9a-f]{64})"', re.IGNORECASE | re.DOTALL) SUBS_RE = re.compile(r'"chatMessagesSubscription"\s*:\s*\{[^}]*?"hash"\s*:\s*"([0-9a-f]{64})"', re.IGNORECASE | re.DOTALL) # fallback mais frouxo (ordem diferente): SUBS_FALLBACK = re.compile(r'chatMessagesSubscription"\s*[,:\{][^"]{0,400}?"hash"\s*:\s*"([0-9a-f]{64})"', re.IGNORECASE | re.DOTALL) # ============================== BOOTSTRAP ====================================== async def get_bootstrap(bot_name: str, cookie_dict: Dict[str, str], prompt_for_capture: str) -> Dict[str, Optional[str]]: """ Abre a página logada, digita o prompt e intercepta a sendMessageMutation EXATA (headers + body RAW). Também varre scripts (src + inline) para extrair a hash da chatMessagesSubscription. """ async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # Injeta cookies de sessão add = [] for name in ("p-b", "p-lat", "cf_clearance", "__cf_bm", "p-sb"): val = cookie_dict.get(name) if val: add.append({"name": name, "value": val, "domain": "poe.com", "path": "/", "secure": True}) if add: await context.add_cookies(add) page = await context.new_page() await page.goto(f"{POE_URL_ROOT}/{bot_name}", wait_until="domcontentloaded") # formkey via função global ou __NEXT_DATA__ formkey = None try: formkey = await page.evaluate("window.ereNdsRqhp2Rd3LEW && window.ereNdsRqhp2Rd3LEW()") except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): try: json_text = await page.evaluate('document.getElementById("__NEXT_DATA__")?.textContent') if json_text: data = json.loads(json_text) def find_key(node, key): if isinstance(node, dict): for k, v in node.items(): if isinstance(k, str) and k.lower() == key and isinstance(v, str): return v r = find_key(v, key) if r: return r elif isinstance(node, list): for x in node: r = find_key(x, key) if r: return r return None fk2 = find_key(data, "formkey") if isinstance(fk2, str) and len(fk2) >= 32: formkey = fk2 except Exception: pass if not (isinstance(formkey, str) and len(formkey) >= 32): await context.close(); await browser.close() raise RuntimeError("Falha ao extrair poe-formkey.") # metas try: revision_meta = await page.evaluate('document.querySelector("meta[name=\'poe-revision\']")?.content') except Exception: revision_meta = None try: tag_id_meta = await page.evaluate('document.querySelector("meta[name=\'poe-tag-id\']")?.content') except Exception: tag_id_meta = None # ---------- função auxiliar: varrer scripts (src + inline) e achar hashes ---------- def scan_text_for_hashes(text: str) -> Tuple[Optional[str], Optional[str]]: send_hash = None subs_hash = None m1 = SEND_RE.search(text) if m1: send_hash = m1.group(1) m2 = SUBS_RE.search(text) if m2: subs_hash = m2.group(1) else: m2b = SUBS_FALLBACK.search(text) if m2b: subs_hash = m2b.group(1) return send_hash, subs_hash async def find_hashes_via_scripts() -> Tuple[Optional[str], Optional[str]]: send_hash_bundles = None subs_hash_bundles = None # 1) inline scripts try: inline_js: List[str] = await page.evaluate(""" Array.from(document.querySelectorAll('script:not([src])')) .map(s => s.textContent || '') """) except Exception: inline_js = [] for js in inline_js: s, sub = scan_text_for_hashes(js) send_hash_bundles = send_hash_bundles or s subs_hash_bundles = subs_hash_bundles or sub if send_hash_bundles and subs_hash_bundles: return send_hash_bundles, subs_hash_bundles # 2) script URLs (inclui preload) try: script_urls: List[str] = await page.evaluate(""" [ ...Array.from(document.scripts).map(s => s.src).filter(Boolean), ...Array.from(document.querySelectorAll('link[rel="preload"][as="script"]')).map(l => l.href).filter(Boolean) ] """) except Exception: script_urls = [] # normalizar e deduplicar norm_urls = [] seen = set() for u in script_urls: if not u: continue if u.startswith("/"): u = urllib.parse.urljoin(POE_URL_ROOT, u) if u in seen: continue seen.add(u) norm_urls.append(u) sess_tmp = requests.Session() for url in norm_urls: try: r = sess_tmp.get(url, timeout=15) if r.ok and "javascript" in r.headers.get("Content-Type", ""): txt = r.text s, sub = scan_text_for_hashes(txt) send_hash_bundles = send_hash_bundles or s subs_hash_bundles = subs_hash_bundles or sub if send_hash_bundles and subs_hash_bundles: break except Exception: continue return send_hash_bundles, subs_hash_bundles # primeira tentativa de achar hashes (antes de digitar o prompt) send_hash_bundles, subs_hash_bundles = await find_hashes_via_scripts() # Intercepta a sendMessageMutation (headers + body RAW) — e NÃO envia de verdade captured = { "payload_dict": None, "payload_raw": None, "hash": None, "req_headers": {}, "tag_id_hdr": None, "tchannel_hdr": None, "cookie_header": None, "revision_hdr": None, "formkey_hdr": None, } async def route_handler(route, request): try: if request.url.endswith("/api/gql_POST"): hdrs = {k.lower(): v for k, v in request.headers.items()} body = request.post_data or "" qn = hdrs.get("poe-queryname", "") if qn == "sendMessageMutation" or "sendMessageMutation" in (body or ""): captured["req_headers"] = {**captured["req_headers"], **hdrs} captured["payload_raw"] = body if "cookie" in hdrs: captured["cookie_header"] = hdrs["cookie"] if "poe-revision" in hdrs: captured["revision_hdr"] = hdrs["poe-revision"] if "poe-formkey" in hdrs: captured["formkey_hdr"] = hdrs["poe-formkey"] if "poe-tchannel" in hdrs: captured["tchannel_hdr"] = hdrs["poe-tchannel"] if "poe-tag-id" in hdrs: captured["tag_id_hdr"] = hdrs["poe-tag-id"] try: payload = json.loads(body) captured["payload_dict"] = payload captured["hash"] = (payload.get("extensions") or {}).get("hash") except Exception: pass # NÃO enviar de verdade — simulamos 200 e guardamos cópia fiel await route.fulfill(status=200, content_type="application/json", body='{"data":{"messageCreate":null}}') return except Exception: pass await route.continue_() await page.route("**/api/gql_POST", route_handler) # Digita o prompt real e "envia" (será fulfiled acima) try: locator = page.locator("textarea") if await locator.count() > 0: ta = locator.first await ta.click() await ta.fill(prompt_for_capture) else: ed = page.locator('div[contenteditable="true"]').first await ed.click() await ed.fill(prompt_for_capture) await page.keyboard.press("Enter") except Exception: pass # Aguarda captura do body (até ~20s) for _ in range(100): if captured["payload_raw"]: break await page.wait_for_timeout(200) # segunda tentativa de achar a subs hash (scripts podem carregar tardiamente) if not subs_hash_bundles: s2, sub2 = await find_hashes_via_scripts() send_hash_bundles = send_hash_bundles or s2 subs_hash_bundles = subs_hash_bundles or sub2 try: await page.unroute("**/api/gql_POST", route_handler) except Exception: pass await context.close(); await browser.close() # Preferências finais final_formkey = captured["formkey_hdr"] or formkey final_revision = captured["revision_hdr"] or revision_meta final_tag_id = captured["tag_id_hdr"] or tag_id_meta final_tchannel = captured["tchannel_hdr"] payload_dict = captured["payload_dict"] payload_raw = captured["payload_raw"] hash_intercept = captured["hash"] if not payload_raw: raise RuntimeError("Não foi possível capturar a sendMessageMutation (body RAW) para o prompt fornecido.") return { "formkey": final_formkey, "revision": final_revision, "send_hash": (hash_intercept or send_hash_bundles), "sub_hash": subs_hash_bundles, # <- obtido de bundles/inline com regex robusta "tag_id": final_tag_id, "tchannel_hdr": final_tchannel, "cookie_header": captured["cookie_header"], "payload_base": payload_dict, # JSON parseado (apenas para log) "payload_raw": payload_raw, # BODY STRING EXATA (usar no POST) "req_headers": captured["req_headers"] or {}, # HEADERS EXATOS (lowercase) } # ============================== STREAM PARSING ================================= def extract_text_pieces(event: Any) -> List[str]: out: List[str] = [] if isinstance(event, dict): for k in ("text_new", "response", "delta", "text"): v = event.get(k) if isinstance(v, str) and v: out.append(v) for v in event.values(): if isinstance(v, (dict, list)): out.extend(extract_text_pieces(v)) elif isinstance(event, list): for x in event: out.extend(extract_text_pieces(x)) return out def is_final_event(event: Dict[str, Any]) -> bool: data = event.get("data") or {} cat = (event.get("category") or "").lower() state = (data.get("state") or data.get("message_state") or "").lower() if "completed" in cat: return True if state in {"complete", "completed", "stopped", "aborted"}: return True if data.get("is_final") is True: return True return False # ============================== RECEPÇÃO VIA WS ================================= async def listen_via_websocket(chat_id: str, session: requests.Session, subscription_hash: str) -> str: tchannel = session.headers.get("poe-tchannel") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if not tchannel: raise RuntimeError("Sem poe-tchannel (cookie/forçado).") # Monta URL do WS m = re.search(r"chan(\d+)-", tchannel) if not m: raise RuntimeError(f"tchannel inesperado: {tchannel}") shard = m.group(1) ws_url = f"wss://tch{shard}.poe.com/up/{tchannel.replace('poe-','')}/updates?channel={tchannel}" ws_headers = dict(BASE_HEADERS) ws_headers["poe-tchannel"] = tchannel ws_headers["Cookie"] = cookie_header_from_jar(session.cookies) # Assina explicitamente a subscription p/ ESTE chatId real payload = { "type": "subscriptions", "subscriptions": { f"chatMessages_0_{chat_id}": { "queryName": "chatMessagesSubscription", "variables": {"chatId": chat_id}, "hash": subscription_hash, } } } full = "" async with websockets.connect(ws_url, extra_headers=ws_headers, max_size=None) as ws: await ws.send(json.dumps(payload)) async for message in ws: data = json.loads(message) if data.get("mtype") == "subscriptionUpdate": for p in data.get("payload", {}).get("payloads", []): new = p.get("data", {}).get("messageAdded") if not new: continue piece = new.get("text_new") or "" if piece: print(piece, end="", flush=True) full += piece if (new.get("state") or "").lower() in {"complete", "completed"}: print("\n--- Resposta Completa (WS) ---") return full return full # ============================== DIAGNÓSTICO ==================================== def summarize_payload_and_headers(payload_base: Dict[str, Any], req_headers: Dict[str, str]) -> Tuple[str, str]: vars_obj = (payload_base or {}).get("variables") or {} exts_obj = (payload_base or {}).get("extensions") or {} sdid = vars_obj.get("sdid") bot = vars_obj.get("bot") hashv = exts_obj.get("hash") hsum = f"bot:{str(bot)[:12]}… sdid:{str(sdid)[:12]}… hash:{str(hashv)[:8]}…" krit = {k: req_headers.get(k) for k in ["poe-formkey", "poe-revision", "poe-tag-id", "poe-tchannel", "poe-queryname", "poegraphql"]} ksum = " ".join(f"{k.split('-')[1] if k.startswith('poe-') else k}:{(v[:12]+'…') if isinstance(v, str) and len(v)>14 else v}" for k, v in krit.items()) return hsum, ksum # ================================= MAIN ======================================== async def main(): session = requests.Session() session.headers.update(BASE_HEADERS) # Cookies base no jar (para WS posteriormente) session.cookies.update({"p-b": P_B_COOKIE}) if P_LAT_COOKIE: session.cookies.set("p-lat", P_LAT_COOKIE, domain="poe.com", path="/") if CF_CLEARANCE: session.cookies.set("cf_clearance", CF_CLEARANCE, domain="poe.com", path="/") if CF_BM: session.cookies.set("__cf_bm", CF_BM, domain="poe.com", path="/") if P_SB: session.cookies.set("p-sb", P_SB, domain="poe.com", path="/") try: # 1) Captura 1:1 e sub_hash via bundles/inline info = await get_bootstrap( BOT_NAME, {"p-b": P_B_COOKIE, "p-lat": P_LAT_COOKIE, "cf_clearance": CF_CLEARANCE, "__cf_bm": CF_BM, "p-sb": P_SB}, PROMPT, ) payload_raw = info.get("payload_raw") # STRING exata payload_capturado = info.get("payload_base") # JSON (apenas para log) req_headers_cap = {k.lower(): v for k, v in (info.get("req_headers") or {}).items()} subs_hash = info.get("sub_hash") if not payload_raw: raise RuntimeError("Não capturei a sendMessageMutation (body RAW).") if not subs_hash or not re.fullmatch(r"[0-9a-f]{64}", subs_hash): raise RuntimeError("Não foi possível extrair a hash de chatMessagesSubscription dos bundles/inline.") # Cookies exatos do intercept no jar (para WS) captured_cookie_header = info.get("cookie_header") or "" parse_cookie_header_and_set(session, captured_cookie_header) # Headers críticos de sessão para WS formkey = info.get("formkey") revision = info.get("revision") tchannel = info.get("tchannel_hdr") or get_tchannel_from_cookies(session.cookies) or POE_TCHANNEL_FALLBACK if formkey: session.headers["poe-formkey"] = formkey if revision: session.headers["poe-revision"] = revision if tchannel: session.headers["poe-tchannel"] = tchannel session.cookies.set("poe-tchannel-channel", tchannel, domain="poe.com", path="/") # LOG p_sum, h_sum = summarize_payload_and_headers(payload_capturado or {}, req_headers_cap) tch_cookie = get_tchannel_from_cookies(session.cookies) or "n/a" print( f"formkey:{(formkey or '')[:6]}… tag_hdr:{(info.get('tag_id') or '')[:8]}… " f"tchannel_hdr:{(tchannel or '')[:18]}… tchannel_cookie:{tch_cookie[:18]}… " f"sub_hash:{(subs_hash or 'n/a')[:8]}… revision:{(revision or 'n/a')[:10]}…" ) print("payload_base:", p_sum) print("req_hdrs:", h_sum) # 2) Envio 1:1 — headers quase idênticos + body RAW + Cookie literal print("\nEnviando a pergunta via GraphQL com payload/headers CAPTURADOS…") blocked = {"content-length", "host", "connection", "accept-encoding"} gql_headers = {k: v for k, v in req_headers_cap.items() if k not in blocked} gql_headers["content-type"] = "application/json" gql_headers["referer"] = f"{POE_URL_ROOT}/{BOT_NAME}" gql_headers["origin"] = POE_URL_ROOT if captured_cookie_header: gql_headers["cookie"] = captured_cookie_header resp = session.post(POE_API_URL, data=payload_raw, headers=gql_headers) if "poe-tchannel" in resp.headers: session.headers["poe-tchannel"] = resp.headers["poe-tchannel"] resp.raise_for_status() data = resp.json() root = data.get("data", {}) if isinstance(data, dict) else {} chat = None if "messageCreate" in root: chat = (root.get("messageCreate") or {}).get("chat") or {} elif "messageEdgeCreate" in root: chat = (root.get("messageEdgeCreate") or {}).get("chat") or {} if not chat: raise RuntimeError("Resposta sem objeto 'chat' esperado (messageCreate/messageEdgeCreate).") chat_id = chat.get("chatId") or chat.get("id") if not chat_id: raise RuntimeError("Não consegui obter o chatId.") print(f"Chat criado! ID: {chat_id}") # 3) Recepção via WS (sem receive_POST) final_answer = await listen_via_websocket(str(chat_id), session, subs_hash) print("\nRESPOSTA FINAL:") print(final_answer) except requests.HTTPError as e: body = getattr(e.response, "text", "")[:2000] print(f"HTTPError: {e}\nBody: {body}") except Exception as e: print(f"Erro: {e}") if __name__ == "__main__": asyncio.run(main())