diff --git a/characterai/characterai.py b/characterai/characterai.py index b0cce82..93d0d4a 100644 --- a/characterai/characterai.py +++ b/characterai/characterai.py @@ -1,51 +1,92 @@ -from typing import Dict import json from playwright.sync_api import sync_playwright from characterai import errors -from characterai.pyasynccai import pyAsyncCAI +from characterai.pyasynccai import PyAsyncCAI -__all__ = ['pyCAI', 'pyAsyncCAI'] +__all__ = ['PyCAI', 'PyAsyncCAI'] def goto(link: str, *, wait: bool = False, token: str = None): if token != None: - page.set_extra_http_headers({"Authorization": f"Token {token}"}) + page.set_extra_http_headers( + {"Authorization": f"Token {token}"} + ) + + page.goto(f'https://beta.character.ai/{link}') - page.goto(link) + content = (page.locator('body').inner_text()) + + if content.startswith('Not Found'): + raise errors.NotFoundError(content.split('\n')[-1]) + elif content == 'No history found for id provided.': + raise errors.NotFoundError(content) + elif content.startswith('{"error":'): + raise errors.ServerError(json.loads(content)['error']) + elif content.startswith('{"detail":'): + raise errors.AuthError(json.loads(content)['detail']) if page.title() != 'Waiting Room powered by Cloudflare': return page - else: if wait: - page.wait_for_selector('div#wrapper', state='detached', timeout=0) + page.wait_for_selector( + 'div#wrapper', state='detached', timeout=0 + ) goto(link=link, wait=wait) else: raise errors.NoResponse('The Site is Overloaded') -def GetResponse(link: str, *, wait: bool = False, token: str = None) -> Dict[str, str]: - goto(f'https://beta.character.ai/{link}/', wait=wait, token=token) +def GetResponse( + link: str, *, wait: bool = False, + token: str = None + ): + goto(link, wait=wait, token=token) data = json.loads(page.locator('body').inner_text()) return data -def PostResponse(link: str, post_link: str, data: str, headers: str, *, json: bool = True, wait: bool = False, token: str = None) -> Dict[str, str]: +def PostResponse( + link: str, post_link: str, data: str, *, + headers: str = None, json: bool = True, + wait: bool = False, token: str = None, + method: str = 'POST' + ): + post_link = f'https://beta.character.ai/{post_link}' + goto(link, wait=wait, token=token) + if headers == None: + headers = { + 'Authorization': f'Token {token}', + 'Content-Type': 'application/json' + } + 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) + "),})") + 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;};""" + + "fetch('" + + post_link + "', {method: '" + + method + "',body: JSON.stringify(" + + str(data) + "),headers: new Headers(" + + str(headers) + "),})" + ) + + response = response_info.value + + if response.status != 200: + raise errors.ServerError(response.status_text) if json: - return response_info.value.json() + return response.json() else: - return response_info.value.text() - + return response.text() -class pyCAI: +class PyCAI: def __init__(self, token: str, *, headless: bool = True): global page @@ -63,152 +104,356 @@ def __init__(self, token: str, *, headless: bool = True): self.chat = self.chat() class user: - """ - Just a Responses from site for user info + """Just a Responses from site for user info user.info() user.posts() user.followers() user.following() user.recent() + """ - def info(self, username: str = None, *, wait: bool = False, token: str = None) -> Dict[str, str]: + def info( + self, username: str = None, *, + wait: bool = False, token: str = None + ): if username == None: - return GetResponse('chat/user', wait=wait, token=token) + 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/', + link=f'public-profile/?username={username}', + post_link='chat/user/public/', data={'username': username}, - headers={'Authorization': f'Token {token}','Content-Type': 'application/json'}, - wait=wait, - token=token + wait=wait, token=token ) - 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) - - def followers(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse(link='chat/user/followers', wait=wait, token=token) + def posts( + self, username: str = None, *, + wait: bool = False, token: str = None + ): + if username == None: + return GetResponse( + 'chat/posts/user/?scope=user&page=1&posts_to_load=5/', + wait=wait, token=token + ) + else: + return GetResponse( + f'chat/posts/user/?username={username}&page=1&posts_to_load=5/', + wait=wait, token=token + ) - def following(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse(link='chat/user/following', wait=wait, token=token) + def followers(self, *, wait: bool = False, token: str = None): + return GetResponse( + 'chat/user/followers/', + wait=wait, token=token + ) - def recent(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse('chat/characters/recent/', wait=wait, token=token) + def following(self, *, wait: bool = False, token: str = None): + return GetResponse( + 'chat/user/following/', + wait=wait, token=token + ) + + def recent(self, *, wait: bool = False, token: str = None): + return GetResponse( + 'chat/characters/recent/', + wait=wait, token=token + ) class character: - """ - Just a Responses from site for characters + """Just a Responses from site for characters character.trending() character.recommended() character.categories() character.info('CHAR') character.search('SEARCH') + """ - def trending(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse(link='chat/characters/trending', wait=wait, token=token) + def trending( + self, *, wait: bool = False, + token: str = None + ): + return GetResponse( + 'chat/characters/trending/', + wait=wait, token=token + ) - def recommended(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse(link='chat/characters/recommended', wait=wait, token=token) + def recommended( + self, *, wait: bool = False, + token: str = None + ): + return GetResponse( + 'chat/characters/recommended/', + wait=wait, token=token + ) - def categories(self, *, wait: bool = False, token: str = None) -> Dict[str, str]: - return GetResponse(link='chat/character/categories', wait=wait, token=token) + def categories( + self, *, wait: bool = False, + token: str = None + ): + return GetResponse( + 'chat/character/categories/', + wait=wait, token=token + ) - def info(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]: - 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) + def info( + self, char: str, *, + wait: bool = False, token: str = None + ): + return GetResponse( + f'chat/character/info-cached/{char}/', + wait=wait, token=token + ) + + def search( + self, query: str, *, + wait: bool = False, token: str = None + ): + return GetResponse( + f'chat/characters/search/?query={query}/', + wait=wait, token=token + ) class chat: - def get_histories(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]: + def rate( + self, char: str, rate: int, *, + history_external_id: str = None, + message_uuid: str = None, + wait: bool = False, token: str = None + ): + """Rate message, return json + + chat.rate('CHAR', NUM) + """ - Getting all character chat histories, return json response + with page.expect_response( + lambda response: response.url.startswith( + 'https://beta.character.ai/chat/history/msgs/user/' + ) + ) as response_info: + goto(f'chat?char={char}', wait=wait, token=token) - chat.get_histories('CHAR') + if rate == 0: label = [234, 238, 241, 244] #Terrible + elif rate == 1: label = [235, 237, 241, 244] #Bad + elif rate == 2: label = [235, 238, 240, 244] #Good + elif rate == 3: label = [235, 238, 241, 243] #Fantastic + else: raise errors.LabelError('Wrong Rate Value') + + history_data = response_info.value + + history = history_data.json() + history_external_id = history_data.url.split('=')[-1] + + response = PostResponse( + link=f'chat?char={char}', + post_link='chat/annotations/label/', + data={ + "message_uuid": history['messages'][-1]['uuid'], + "history_external_id": history_external_id, + "label_ids": label + }, + wait=wait, json=False, token=token, method='PUT' + ) + + return response + + def next_message( + self, char: str, *, wait: bool = False, + token: str = None, filtering: bool = True + ): + """Next message, return json + + chat.next_message('CHAR', 'MESSAGE') + """ - return PostResponse( - link=f'https://beta.character.ai/chat?char={char}', - post_link='https://beta.character.ai/chat/character/histories/', + with page.expect_response( + lambda response: response.url.startswith( + 'https://beta.character.ai/chat/history/msgs/user/' + ) + ) as response_info: + goto(f'chat?char={char}', wait=wait, token=token) + + history = response_info.value.json() + url = response_info.value.url + + #Get last user message for uuid and text + for h in history['messages']: + if h['src__is_human'] == True: + last_message = h + + response = PostResponse( + link=f'chat?char={char}', + post_link='chat/streaming/', data={ - "external_id": char, - "number": 50, + "character_external_id": char, + "history_external_id": url.split('=')[-1], + "text": last_message['text'], + "tgt": history['messages'][-1]['src__user__username'], + "parent_msg_uuid": last_message['uuid'] }, - headers={'Authorization': f'Token {token}','Content-Type': 'application/json'}, + wait=wait, json=False, token=token + ) + + if response.split('\n')[-1].startswith('{"abort"'): + if filtering: + raise errors.FilterError('No eligible candidates') + else: + return json.loads(response.split('\n')[-3]) + else: + return json.loads(response.split('\n')[-2]) + + def get_histories( + self, char: str, *, + wait: bool = False, token: str = None + ): + """Getting all character chat histories + + chat.get_histories('CHAR') + + """ + return PostResponse( + link=f'chat?char={char}', + post_link='chat/character/histories/', + data={"external_id": char, "number": 50}, wait=wait, token=token ) - def get_history(self, char: str, *, wait: bool = False, token: str = None) -> Dict[str, str]: - """ - Getting character chat history, return json response + def get_history( + self, char: str = None, *, + wait: bool = False, token: str = None + ): + """Getting character chat history - chat.get_history('CHAR') - """ - goto(f'https://beta.character.ai/chat?char={char}', wait=wait, token=token) + chat.get_history('HISTORY_EXTERNAL_ID') - if page.query_selector('h1') == None: - with page.expect_response(lambda response: response.url.startswith('https://beta.character.ai/chat/history/msgs/user/')) as response_info: - goto(f'https://beta.character.ai/chat?char={char}', wait=wait, token=token) + """ + try: + return GetResponse( + f'chat/history/msgs/user/?history_external_id={char}', + wait=wait, token=token + ) + except: + char_data = PostResponse( + link=f'chat?char={char}', + post_link='chat/history/continue/', + data={"character_external_id": char}, + wait=wait, + token=token + ) - response = response_info.value.text() - return json.loads(response) - else: - raise errors.CharNotFound('Wrong Char') + history_id = char_data['external_id'] + + return GetResponse( + f'chat/history/msgs/user/?history_external_id={history_id}', + wait=wait, token=token + ) + + def get_chat( + self, char: str = None, *, + wait: bool = False, token: str = None + ): + """Getting the main information about the chat - 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]: + chat.get_chat('CHAR') + """ - Sending a message, return json + return PostResponse( + link=f'chat?char={char}', + post_link='chat/history/continue/', + data={"character_external_id": char}, + wait=wait, + token=token + ) + + def send_message( + self, char: str, message: str, *, + history_external_id: str = None, + tgt: str = None, wait: bool = False, + token: str = None, filtering: bool = True + ): + """Sending a message, return json chat.send_message('CHAR', 'MESSAGE') + """ # Get history_external_id and tgt - if history_external_id == None and tgt == None: + if history_external_id == None or tgt == None: info = PostResponse( - link=f'https://beta.character.ai/chat?char={char}', - post_link='https://beta.character.ai/chat/history/continue/', + link=f'chat?char={char}', + post_link='chat/history/continue/', data={'character_external_id': char}, - headers={'Authorization': f'Token {token}','Content-Type': 'application/json'}, wait=wait, token=token ) - history_external_id = info['external_id'] - tgt = info['participants'][1]['user']['username'] + if history_external_id == None: + history_external_id = info['external_id'] + + if tgt == None: + # In the list of "participants", + # a character can be at zero or in the first place + if not info['participants'][0]['is_human']: + tgt = info['participants'][0]['user']['username'] + else: + 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/', + link=f'chat?char={char}', + post_link='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, token=token ) + + if response.split('\n')[-1].startswith('{"abort"'): + if filtering: + raise errors.FilterError('No eligible candidates') + else: + return json.loads(response.split('\n')[-3]) + else: + return json.loads(response.split('\n')[-2]) - try: - return json.loads('{"replies": ' + str(response.split('{"replies": ')[-1].split('\n')[0])) - except: - return response + def delete_message( + self, history_id: str, uuids_to_delete: list, *, + wait: bool = False, token: str = None + ): + """Delete a message - def new_chat(self, char: str, *, wait: bool = False, token: str = None) -> None: + chat.new_chat('CHAR') + """ - Starting new chat, return new chat history + return PostResponse( + link='chat', + post_link='chat/history/msgs/delete/', + data={ + "history_id": history_id, + "uuids_to_delete": uuids_to_delete + }, + wait=wait, + token=token + ) + + def new_chat( + self, char: str, *, + wait: bool = False, token: str = None + ): + """Starting new chat, return new chat history chat.new_chat('CHAR') + """ return PostResponse( - link=f'https://beta.character.ai/chat?char={char}', - post_link='https://beta.character.ai/chat/history/create/', + link=f'chat?char={char}', + post_link='chat/history/create/', data={'character_external_id': char}, - headers={'Authorization': f'Token {token}', 'Content-Type': 'application/json'}, wait=wait, token=token )