Skip to content

Commit

Permalink
new v0.5.0
Browse files Browse the repository at this point in the history
  • Loading branch information
@kramcat
kramcat authored May 24, 2023
1 parent 9e07c83 commit c395559
Showing 1 changed file with 90 additions and 41 deletions.
131 changes: 90 additions & 41 deletions characterai/characterai.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,19 +8,14 @@

__all__ = ['pyCAI', 'pyAsyncCAI']

page = None

def goto(link: str, *, wait: bool = False, token: str = None):
if token != None:
page.set_extra_http_headers({"Authorization": f"Token {token}"})

page.goto(link)

if page.title() != 'Waiting Room powered by Cloudflare':
if page.get_by_role("button", name="Accept").is_visible():
page.get_by_role("button", name="Accept").click()

return page
return page

else:
if wait:
Expand All @@ -35,6 +30,20 @@ def GetResponse(link: str, *, wait: bool = False, token: str = None) -> Dict[str

return data

def PostResponse(link: str, post_link: str, data: str, headers: str, *, json: bool = True, wait: bool = False) -> Dict[str, str]:
goto(link, wait=wait)

with page.expect_response(post_link) as response_info:
# From HearYourWaifu
page.evaluate("const {fetch: origFetch} = window;window.fetch = async (...args) => {const response = await origFetch(...args);const raw_text = await new Response(response.clone().body).text();return response;};")

page.evaluate("fetch('" + post_link + "', {method: 'POST',body: JSON.stringify(" + str(data) + "),headers: new Headers(" + str(headers) + "),})")

if json:
return response_info.value.json()
else:
return response_info.value.text()


class pyCAI:
def __init__(self, token: str, *, headless: bool = True):
Expand All @@ -58,10 +67,22 @@ class user:
Just a Responses from site for user info
user.info()
user.posts()
user.followers()
user.following()
user.recent()
"""

def info(self, *, wait: bool = False, token: str = None) -> Dict[str, str]:
return GetResponse(link='chat/user', wait=wait, token=token)
def info(self, username: str = None, *, wait: bool = False, token: str = None) -> Dict[str, str]:
if username == None:
return GetResponse('chat/user', wait=wait, token=token)
else:
return PostResponse(
link=f'https://beta.character.ai/public-profile/?username={username}',
post_link='https://beta.character.ai/chat/user/public/',
data={'username': username},
headers={'Authorization': f'Token {token}','Content-Type': 'application/json'},
wait=wait
)

def posts(self, *, wait: bool = False, token: str = None) -> Dict[str, str]:
return GetResponse(link='chat/posts/user/?scope=user&page=1&posts_to_load=5', wait=wait, token=token)
Expand All @@ -80,6 +101,8 @@ class character:
Just a Responses from site for characters
character.trending()
character.recommended()
character.categories()
character.info('CHAR')
character.search('SEARCH')
"""
Expand All @@ -93,17 +116,29 @@ def categories(self, *, wait: bool = False, token: str = None) -> Dict[str, str]
return GetResponse(link='chat/character/categories', wait=wait, token=token)

def info(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]:
data = GetResponse(f'chat/character/info-cached/{char}/', wait=wait, token=token)

if data != "{'error': 'Server Error (500)'}":
return data
else:
raise errors.CharNotFound('Wrong Char')
return GetResponse(f'chat/character/info-cached/{char}/', wait=wait, token=token)

def search(self, search, *, wait: bool = False, token: str = None) -> Dict[str, str]:
return GetResponse(f'chat/characters/search/?query={search}', wait=wait, token=token)

class chat:
def get_histories(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]:
"""
Getting all character chat histories, return json response
chat.get_histories('CHAR')
"""
return PostResponse(
link=f'https://beta.character.ai/chat?char={char}',
post_link='https://beta.character.ai/chat/character/histories/',
data={
"external_id": char,
"number": 50,
},
headers={'Authorization': f'Token {token}','Content-Type': 'application/json'},
wait=wait
)

def get_history(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]:
"""
Getting character chat history, return json response
Expand All @@ -121,40 +156,54 @@ def get_history(self, char: str, *, wait: bool = False, token: str = None) -> Di
else:
raise errors.CharNotFound('Wrong Char')

def send_message(self, char: str, message: str, *, wait: bool = False, token: str = None) -> Dict[str, str]:
def send_message(self, char: str, message: str, *, history_external_id: str = None, tgt: str = None, wait: bool = False, token: str = None) -> Dict[str, str]:
"""
Sending a message, return json
chat.send_message('CHAR', 'MESSAGE')
"""
goto(f'https://beta.character.ai/chat?char={char}', wait=wait, token=token)

# BIG THANKS - HearYourWaifu
page.evaluate("""
const { fetch: origFetch } = window;
window.fetch = async (...args) => {
const response = await origFetch(...args);
const raw_text = await new Response(response.clone().body).text();
return response;};""")

with page.expect_response("https://beta.character.ai/chat/streaming/") as response_info:
page.get_by_placeholder("Type a message").fill(message)
page.get_by_role("button", name="Submit Message").click()

response = response_info.value.text()
return json.loads('{"replies": ' + response.split(r'{"replies": ')[-1])
# Get history_external_id and tgt
if history_external_id == None and tgt == None:
info = PostResponse(
link=f'https://beta.character.ai/chat?char={char}',
post_link='https://beta.character.ai/chat/history/continue/',
data={'character_external_id': char},
headers={'Authorization': f'Token {token}','Content-Type': 'application/json'},
wait=wait
)

history_external_id = info['external_id']
tgt = info['participants'][1]['user']['username']

response = PostResponse(
link=f'https://beta.character.ai/chat?char={char}',
post_link='https://beta.character.ai/chat/streaming/',
data={
"history_external_id": history_external_id,
"character_external_id": char,
"text": message,
"tgt": tgt
},
headers={'Authorization': f'Token {token}','Content-Type': 'application/json'},
wait=wait,
json=False
)

try:
return json.loads('{"replies": ' + str(response.split('{"replies": ')[-1].split('\n')[0]))
except:
return response

def new_chat(self, char: str, *, wait: bool = False, token: str = None) -> None:
"""
Starting new chat, return json
Starting new chat, return new chat history
chat.new_chat('CHAR')
"""
goto(f'https://beta.character.ai/chat?char={char}', wait=wait, token=token)

with page.expect_response("https://beta.character.ai/chat/history/create/") as response_info:
page.wait_for_selector('.col-auto.px-2.dropdown').click()
page.wait_for_selector('"Save and Start New Chat"').click()

response = response_info.value.text()
return json.loads(response)
return PostResponse(
link=f'https://beta.character.ai/chat?char={char}',
post_link='https://beta.character.ai/chat/history/create/',
data={'character_external_id': char},
headers={'Authorization': f'Token {token}', 'Content-Type': 'application/json'},
wait=wait
)

0 comments on commit c395559

Please sign in to comment.