import json
from datetime import datetime
from hashlib import md5
from json import dumps
from pathlib import Path
from random import choice , choices , randint
from re import search , findall
from string import ascii_letters , digits
from typing import Optional
from urllib . parse import unquote
import selenium . webdriver . support . expected_conditions as EC
from pypasser import reCaptchaV3
from requests import Session
from selenium import webdriver
from selenium . webdriver . common . by import By
from selenium . webdriver . support . wait import WebDriverWait
from tls_client import Session as TLS
from quora . api import Client as PoeClient
from quora . mail import Emailnator
# from twocaptcha import TwoCaptcha
# solver = TwoCaptcha('72747bf24a9d89b4dcc1b24875efd358')
MODELS = {
" Sage " : " capybara " ,
" GPT-4 " : " beaver " ,
" Claude+ " : " a2_2 " ,
" Claude-instant " : " a2 " ,
" ChatGPT " : " chinchilla " ,
" Dragonfly " : " nutria " ,
" NeevaAI " : " hutia " ,
}
def extract_formkey ( html ) :
script_regex = r " <script>if \ (.+ \ )throw new Error;(.+)</script> "
script_text = search ( script_regex , html ) . group ( 1 )
key_regex = r ' var .= " ([0-9a-f]+) " , '
key_text = search ( key_regex , script_text ) . group ( 1 )
cipher_regex = r " . \ [( \ d+) \ ]=. \ [( \ d+) \ ] "
cipher_pairs = findall ( cipher_regex , script_text )
formkey_list = [ " " ] * len ( cipher_pairs )
for pair in cipher_pairs :
formkey_index , key_index = map ( int , pair )
formkey_list [ formkey_index ] = key_text [ key_index ]
formkey = " " . join ( formkey_list )
return formkey
class PoeResponse :
class Completion :
class Choices :
def __init__ ( self , choice : dict ) - > None :
self . text = choice [ " text " ]
self . content = self . text . encode ( )
self . index = choice [ " index " ]
self . logprobs = choice [ " logprobs " ]
self . finish_reason = choice [ " finish_reason " ]
def __repr__ ( self ) - > str :
return f """ <__main__.APIResponse.Completion.Choices( \n text = { self . text . encode ( ) } , \n index = { self . index } , \n logprobs = { self . logprobs } , \n finish_reason = { self . finish_reason } )object at 0x1337> """
def __init__ ( self , choices : dict ) - > None :
self . choices = [ self . Choices ( choice ) for choice in choices ]
class Usage :
def __init__ ( self , usage_dict : dict ) - > None :
self . prompt_tokens = usage_dict [ " prompt_tokens " ]
self . completion_tokens = usage_dict [ " completion_tokens " ]
self . total_tokens = usage_dict [ " total_tokens " ]
def __repr__ ( self ) :
return f """ <__main__.APIResponse.Usage( \n prompt_tokens = { self . prompt_tokens } , \n completion_tokens = { self . completion_tokens } , \n total_tokens = { self . total_tokens } )object at 0x1337> """
def __init__ ( self , response_dict : dict ) - > None :
self . response_dict = response_dict
self . id = response_dict [ " id " ]
self . object = response_dict [ " object " ]
self . created = response_dict [ " created " ]
self . model = response_dict [ " model " ]
self . completion = self . Completion ( response_dict [ " choices " ] )
self . usage = self . Usage ( response_dict [ " usage " ] )
def json ( self ) - > dict :
return self . response_dict
class ModelResponse :
def __init__ ( self , json_response : dict ) - > None :
self . id = json_response [ " data " ] [ " poeBotCreate " ] [ " bot " ] [ " id " ]
self . name = json_response [ " data " ] [ " poeBotCreate " ] [ " bot " ] [ " displayName " ]
self . limit = json_response [ " data " ] [ " poeBotCreate " ] [ " bot " ] [ " messageLimit " ] [
" dailyLimit "
]
self . deleted = json_response [ " data " ] [ " poeBotCreate " ] [ " bot " ] [ " deletionState " ]
class Model :
def create (
token : str ,
model : str = " gpt-3.5-turbo " , # claude-instant
system_prompt : str = " You are ChatGPT a large language model developed by Openai. Answer as consisely as possible " ,
description : str = " gpt-3.5 language model from openai, skidded by poe.com " ,
handle : str = None ,
) - > ModelResponse :
models = {
" gpt-3.5-turbo " : " chinchilla " ,
" claude-instant-v1.0 " : " a2 " ,
" gpt-4 " : " beaver " ,
}
if not handle :
handle = f " gptx { randint ( 1111111 , 9999999 ) } "
client = Session ( )
client . cookies [ " p-b " ] = token
formkey = extract_formkey ( client . get ( " https://poe.com " ) . text )
settings = client . get ( " https://poe.com/api/settings " ) . json ( )
client . headers = {
" host " : " poe.com " ,
" origin " : " https://poe.com " ,
" referer " : " https://poe.com/ " ,
" poe-formkey " : formkey ,
" poe-tchannel " : settings [ " tchannelData " ] [ " channel " ] ,
" user-agent " : " Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36 " ,
" connection " : " keep-alive " ,
" sec-ch-ua " : ' " Chromium " ;v= " 112 " , " Google Chrome " ;v= " 112 " , " Not:A-Brand " ;v= " 99 " ' ,
" sec-ch-ua-mobile " : " ?0 " ,
" sec-ch-ua-platform " : ' " macOS " ' ,
" content-type " : " application/json " ,
" sec-fetch-site " : " same-origin " ,
" sec-fetch-mode " : " cors " ,
" sec-fetch-dest " : " empty " ,
" accept " : " */* " ,
" accept-encoding " : " gzip, deflate, br " ,
" accept-language " : " en-GB,en-US;q=0.9,en;q=0.8 " ,
}
payload = dumps (
separators = ( " , " , " : " ) ,
obj = {
" queryName " : " CreateBotMain_poeBotCreate_Mutation " ,
" variables " : {
" model " : models [ model ] ,
" handle " : handle ,
" prompt " : system_prompt ,
" isPromptPublic " : True ,
" introduction " : " " ,
" description " : description ,
" profilePictureUrl " : " https://qph.fs.quoracdn.net/main-qimg-24e0b480dcd946e1cc6728802c5128b6 " ,
" apiUrl " : None ,
" apiKey " : " " . join ( choices ( ascii_letters + digits , k = 32 ) ) ,
" isApiBot " : False ,
" hasLinkification " : False ,
" hasMarkdownRendering " : False ,
" hasSuggestedReplies " : False ,
" isPrivateBot " : False ,
} ,
" query " : " mutation CreateBotMain_poeBotCreate_Mutation( \n $model: String! \n $handle: String! \n $prompt: String! \n $isPromptPublic: Boolean! \n $introduction: String! \n $description: String! \n $profilePictureUrl: String \n $apiUrl: String \n $apiKey: String \n $isApiBot: Boolean \n $hasLinkification: Boolean \n $hasMarkdownRendering: Boolean \n $hasSuggestedReplies: Boolean \n $isPrivateBot: Boolean \n ) { \n poeBotCreate(model: $model, handle: $handle, promptPlaintext: $prompt, isPromptPublic: $isPromptPublic, introduction: $introduction, description: $description, profilePicture: $profilePictureUrl, apiUrl: $apiUrl, apiKey: $apiKey, isApiBot: $isApiBot, hasLinkification: $hasLinkification, hasMarkdownRendering: $hasMarkdownRendering, hasSuggestedReplies: $hasSuggestedReplies, isPrivateBot: $isPrivateBot) { \n status \n bot { \n id \n ...BotHeader_bot \n } \n } \n } \n \n fragment BotHeader_bot on Bot { \n displayName \n messageLimit { \n dailyLimit \n } \n ...BotImage_bot \n ...BotLink_bot \n ...IdAnnotation_node \n ...botHelpers_useViewerCanAccessPrivateBot \n ...botHelpers_useDeletion_bot \n } \n \n fragment BotImage_bot on Bot { \n displayName \n ...botHelpers_useDeletion_bot \n ...BotImage_useProfileImage_bot \n } \n \n fragment BotImage_useProfileImage_bot on Bot { \n image { \n __typename \n ... on LocalBotImage { \n localName \n } \n ... on UrlBotImage { \n url \n } \n } \n ...botHelpers_useDeletion_bot \n } \n \n fragment BotLink_bot on Bot { \n displayName \n } \n \n fragment IdAnnotation_node on Node { \n __isNode: __typename \n id \n } \n \n fragment botHelpers_useDeletion_bot on Bot { \n deletionState \n } \n \n fragment botHelpers_useViewerCanAccessPrivateBot on Bot { \n isPrivateBot \n viewerIsCreator \n } \n " ,
} ,
)
base_string = payload + client . headers [ " poe-formkey " ] + " WpuLMiXEKKE98j56k "
client . headers [ " poe-tag-id " ] = md5 ( base_string . encode ( ) ) . hexdigest ( )
response = client . post ( " https://poe.com/api/gql_POST " , data = payload )
if " success " not in response . text :
raise Exception (
"""
Bot creation Failed
! ! Important ! !
Bot creation was not enabled on this account
please use : quora . Account . create with enable_bot_creation set to True
"""
)
return ModelResponse ( response . json ( ) )
class Account :
def create (
proxy : Optional [ str ] = None ,
logging : bool = False ,
enable_bot_creation : bool = False ,
) :
client = TLS ( client_identifier = " chrome110 " )
client . proxies = (
{ " http " : f " http:// { proxy } " , " https " : f " http:// { proxy } " } if proxy else None
)
mail_client = Emailnator ( )
mail_address = mail_client . get_mail ( )
if logging :
print ( " email " , mail_address )
client . headers = {
" authority " : " poe.com " ,
" accept " : " */* " ,
" accept-language " : " en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3 " ,
" content-type " : " application/json " ,
" origin " : " https://poe.com " ,
" poe-tag-id " : " null " ,
" referer " : " https://poe.com/login " ,
" sec-ch-ua " : ' " Chromium " ;v= " 112 " , " Google Chrome " ;v= " 112 " , " Not:A-Brand " ;v= " 99 " ' ,
" sec-ch-ua-mobile " : " ?0 " ,
" sec-ch-ua-platform " : ' " macOS " ' ,
" sec-fetch-dest " : " empty " ,
" sec-fetch-mode " : " cors " ,
" sec-fetch-site " : " same-origin " ,
" user-agent " : " Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36 " ,
" poe-formkey " : extract_formkey ( client . get ( " https://poe.com/login " ) . text ) ,
" poe-tchannel " : client . get ( " https://poe.com/api/settings " ) . json ( ) [
" tchannelData "
] [ " channel " ] ,
}
token = reCaptchaV3 (
" https://www.recaptcha.net/recaptcha/enterprise/anchor?ar=1&k=6LflhEElAAAAAI_ewVwRWI9hsyV4mbZnYAslSvlG&co=aHR0cHM6Ly9wb2UuY29tOjQ0Mw..&hl=en&v=4PnKmGB9wRHh1i04o7YUICeI&size=invisible&cb=bi6ivxoskyal "
)
# token = solver.recaptcha(sitekey='6LflhEElAAAAAI_ewVwRWI9hsyV4mbZnYAslSvlG',
# url = 'https://poe.com/login?redirect_url=%2F',
# version = 'v3',
# enterprise = 1,
# invisible = 1,
# action = 'login',)['code']
payload = dumps (
separators = ( " , " , " : " ) ,
obj = {
" queryName " : " MainSignupLoginSection_sendVerificationCodeMutation_Mutation " ,
" variables " : {
" emailAddress " : mail_address ,
" phoneNumber " : None ,
" recaptchaToken " : token ,
} ,
" query " : " mutation MainSignupLoginSection_sendVerificationCodeMutation_Mutation( \n $emailAddress: String \n $phoneNumber: String \n $recaptchaToken: String \n ) { \n sendVerificationCode(verificationReason: login, emailAddress: $emailAddress, phoneNumber: $phoneNumber, recaptchaToken: $recaptchaToken) { \n status \n errorMessage \n } \n } \n " ,
} ,
)
base_string = payload + client . headers [ " poe-formkey " ] + " WpuLMiXEKKE98j56k "
client . headers [ " poe-tag-id " ] = md5 ( base_string . encode ( ) ) . hexdigest ( )
print ( dumps ( client . headers , indent = 4 ) )
response = client . post ( " https://poe.com/api/gql_POST " , data = payload )
if " automated_request_detected " in response . text :
print ( " please try using a proxy / wait for fix " )
if " Bad Request " in response . text :
if logging :
print ( " bad request, retrying... " , response . json ( ) )
quit ( )
if logging :
print ( " send_code " , response . json ( ) )
mail_content = mail_client . get_message ( )
mail_token = findall ( r ' ; " >( \ d { 6,7})</div> ' , mail_content ) [ 0 ]
if logging :
print ( " code " , mail_token )
payload = dumps (
separators = ( " , " , " : " ) ,
obj = {
" queryName " : " SignupOrLoginWithCodeSection_signupWithVerificationCodeMutation_Mutation " ,
" variables " : {
" verificationCode " : str ( mail_token ) ,
" emailAddress " : mail_address ,
" phoneNumber " : None ,
} ,
" query " : " mutation SignupOrLoginWithCodeSection_signupWithVerificationCodeMutation_Mutation( \n $verificationCode: String! \n $emailAddress: String \n $phoneNumber: String \n ) { \n signupWithVerificationCode(verificationCode: $verificationCode, emailAddress: $emailAddress, phoneNumber: $phoneNumber) { \n status \n errorMessage \n } \n } \n " ,
} ,
)
base_string = payload + client . headers [ " poe-formkey " ] + " WpuLMiXEKKE98j56k "
client . headers [ " poe-tag-id " ] = md5 ( base_string . encode ( ) ) . hexdigest ( )
response = client . post ( " https://poe.com/api/gql_POST " , data = payload )
if logging :
print ( " verify_code " , response . json ( ) )
def get ( self ) :
cookies = (
open ( Path ( __file__ ) . resolve ( ) . parent / " cookies.txt " , " r " )
. read ( )
. splitlines ( )
)
return choice ( cookies )
class StreamingCompletion :
def create (
model : str = " gpt-4 " ,
custom_model : bool = None ,
prompt : str = " hello world " ,
token : str = " " ,
) :
_model = MODELS [ model ] if not custom_model else custom_model
client = PoeClient ( token )
for chunk in client . send_message ( _model , prompt ) :
yield PoeResponse (
{
" id " : chunk [ " messageId " ] ,
" object " : " text_completion " ,
" created " : chunk [ " creationTime " ] ,
" model " : _model ,
" choices " : [
{
" text " : chunk [ " text_new " ] ,
" index " : 0 ,
" logprobs " : None ,
" finish_reason " : " stop " ,
}
] ,
" usage " : {
" prompt_tokens " : len ( prompt ) ,
" completion_tokens " : len ( chunk [ " text_new " ] ) ,
" total_tokens " : len ( prompt ) + len ( chunk [ " text_new " ] ) ,
} ,
}
)
class Completion :
def create (
model : str = " gpt-4 " ,
custom_model : str = None ,
prompt : str = " hello world " ,
token : str = " " ,
) :
models = {
" sage " : " capybara " ,
" gpt-4 " : " beaver " ,
" claude-v1.2 " : " a2_2 " ,
" claude-instant-v1.0 " : " a2 " ,
" gpt-3.5-turbo " : " chinchilla " ,
}
_model = models [ model ] if not custom_model else custom_model
client = PoeClient ( token )
for chunk in client . send_message ( _model , prompt ) :
pass
return PoeResponse (
{
" id " : chunk [ " messageId " ] ,
" object " : " text_completion " ,
" created " : chunk [ " creationTime " ] ,
" model " : _model ,
" choices " : [
{
" text " : chunk [ " text " ] ,
" index " : 0 ,
" logprobs " : None ,
" finish_reason " : " stop " ,
}
] ,
" usage " : {
" prompt_tokens " : len ( prompt ) ,
" completion_tokens " : len ( chunk [ " text " ] ) ,
" total_tokens " : len ( prompt ) + len ( chunk [ " text " ] ) ,
} ,
}
)
class Poe :
def __init__ ( self , model : str = " ChatGPT " ) :
# validating the model
if model and model not in MODELS :
raise RuntimeError (
" Sorry, the model you provided does not exist. Please check and try again. "
)
self . model = MODELS [ model ]
self . cookie = self . __load_cookie ( )
self . client = PoeClient ( self . cookie )
def __load_cookie ( self ) - > str :
if ( cookie_file := Path ( " cookie.json " ) ) . exists ( ) :
with cookie_file . open ( ) as fp :
cookie = json . load ( fp )
if datetime . fromtimestamp ( cookie [ " expiry " ] ) < datetime . now ( ) :
cookie = self . __register_and_get_cookie ( )
else :
print ( " Loading the cookie from file " )
else :
cookie = self . __register_and_get_cookie ( )
return unquote ( cookie [ " value " ] )
@classmethod
def __register_and_get_cookie ( cls ) - > dict :
mail_client = Emailnator ( )
mail_address = mail_client . get_mail ( )
print ( mail_address )
options = webdriver . FirefoxOptions ( )
# options.add_argument("-headless")
driver = webdriver . Firefox ( options = options )
driver . get ( " https://www.poe.com " )
# clicking use email button
driver . find_element ( By . XPATH , ' //button[contains(text(), " Use email " )] ' ) . click ( )
email = WebDriverWait ( driver , 30 ) . until (
EC . presence_of_element_located ( ( By . XPATH , ' //input[@type= " email " ] ' ) )
)
email . send_keys ( mail_address )
driver . find_element ( By . XPATH , ' //button[text()= " Go " ] ' ) . click ( )
code = findall ( r ' ; " >( \ d { 6,7})</div> ' , mail_client . get_message ( ) ) [ 0 ]
print ( code )
verification_code = WebDriverWait ( driver , 30 ) . until (
EC . presence_of_element_located ( ( By . XPATH , ' //input[@placeholder= " Code " ] ' ) )
)
verification_code . send_keys ( code )
verify_button = EC . presence_of_element_located (
( By . XPATH , ' //button[text()= " Verify " ] ' )
)
login_button = EC . presence_of_element_located (
( By . XPATH , ' //button[text()= " Log In " ] ' )
)
WebDriverWait ( driver , 30 ) . until ( EC . any_of ( verify_button , login_button ) ) . click ( )
cookie = driver . get_cookie ( " p-b " )
with open ( " cookie.json " , " w " ) as fw :
json . dump ( cookie , fw )
driver . close ( )
return cookie
def chat ( self , message : str , model : Optional [ str ] = None ) - > str :
if model and model not in MODELS :
raise RuntimeError (
" Sorry, the model you provided does not exist. Please check and try again. "
)
model = MODELS [ model ] if model else self . model
response = None
for chunk in self . client . send_message ( model , message ) :
response = chunk [ " text " ]
return response
def create_bot (
self ,
name : str ,
/ ,
prompt : str = " " ,
base_model : str = " ChatGPT " ,
description : str = " " ,
) - > None :
if base_model not in MODELS :
raise RuntimeError (
" Sorry, the base_model you provided does not exist. Please check and try again. "
)
response = self . client . create_bot (
handle = name ,
prompt = prompt ,
base_model = MODELS [ base_model ] ,
description = description ,
)
print ( f ' Successfully created bot with name: { response [ " bot " ] [ " displayName " ] } ' )
def list_bots ( self ) - > list :
return list ( self . client . bot_names . values ( ) )