diff --git a/README.md b/README.md index 8573be2..f6c0968 100644 --- a/README.md +++ b/README.md @@ -22,8 +22,8 @@ - [🦄 Documentation](#-documentation) - [Available Default Bots](#available-default-bots) - [How to get your Token](#how-to-get-your-token) - - [Getting p-b and p-lat cookies](#getting-p-b-and-p-lat-cookies) - - [Getting formkey](#getting-formkey) + - [Getting p-b and p-lat cookies (*required*)](#getting-p-b-and-p-lat-cookies-required) + - [Getting formkey (*optional*)](#getting-formkey-optional) - [Basic Usage](#basic-usage) - [Bots Group Chat (beta)](#bots-group-chat-beta) - [Misc](#misc) @@ -133,7 +133,6 @@ import asyncio tokens = { 'p-b': ..., 'p-lat': ..., - 'formkey': ... } async def main(): @@ -150,7 +149,6 @@ from poe_api_wrapper import PoeExample tokens = { 'p-b': ..., 'p-lat': ..., - 'formkey': ... } PoeExample(tokens=tokens).chat_with_bot() ``` @@ -198,7 +196,7 @@ poe -b P-B_HERE -lat P-LAT_HERE -f FORMKEY_HERE ### How to get your Token -#### Getting p-b and p-lat cookies +#### Getting p-b and p-lat cookies (*required*) Sign in at https://poe.com/ F12 for Devtools (Right-click + Inspect) @@ -208,7 +206,11 @@ F12 for Devtools (Right-click + Inspect) Copy the values of `p-b` and `p-lat` cookies -#### Getting formkey +#### Getting formkey (*optional*) +> [!IMPORTANT] +> **poe-api-wrapper** depends on library **js2py** to get formkey. If you are using `Python 3.12+`, it may not be compatible, so please downgrade your version. +> By default, poe_api_wrapper will automatically retrieve formkey for you. If it doesn't work, please pass this token manually by following these steps: + There are two ways to get formkey: F12 for Devtools (Right-click + Inspect) @@ -227,7 +229,6 @@ F12 for Devtools (Right-click + Inspect) tokens = { 'p-b': 'p-b cookie here', 'p-lat': 'p-lat cookie here', - 'formkey': 'formkey here' } # Default setup @@ -246,7 +247,7 @@ proxy_context = [ client = PoeApi(tokens=tokens, proxy=proxy_context) -# Add cloudflare cookies to pass challenges +# Add formkey and cloudflare cookies to pass challenges tokens = { 'p-b': 'p-b cookie here', 'p-lat': 'p-lat cookie here', diff --git a/poe_api_wrapper/api.py b/poe_api_wrapper/api.py index aad9efc..68c2ab6 100644 --- a/poe_api_wrapper/api.py +++ b/poe_api_wrapper/api.py @@ -3,7 +3,7 @@ from requests_toolbelt import MultipartEncoder import os, secrets, string, random, websocket, json, threading, queue, ssl, hashlib from loguru import logger -from .queries import generate_payload +from typing import Generator from .utils import ( BASE_URL, HEADERS, @@ -14,7 +14,8 @@ generate_nonce, generate_file ) -from typing import Generator +from .queries import generate_payload +from .bundles import PoeBundle from .proxies import PROXY if PROXY: from .proxies import fetch_proxy @@ -30,10 +31,11 @@ class PoeApi: def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): self.client = None - if not {'p-b', 'p-lat', 'formkey'}.issubset(tokens): - raise ValueError("Please provide valid p-b, p-lat and formkey") + if not {'p-b', 'p-lat'}.issubset(tokens): + raise ValueError("Please provide valid p-b and p-lat cookies") self.tokens = tokens + self.formkey = None self.ws_connecting = False self.ws_connected = False self.ws_error = False @@ -44,8 +46,8 @@ def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): self.message_generating = True self.ws_refresh = 3 self.groups = {} - self.formkey = self.tokens['formkey'] self.proxies = {} + self.bundle: PoeBundle = None self.client = Client(headers=self.HEADERS, timeout=60, http2=True) self.client.cookies.update({ @@ -58,11 +60,15 @@ def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): '__cf_bm': tokens['__cf_bm'], 'cf_clearance': tokens['cf_clearance'] }) - - self.client.headers.update({ - 'Poe-Formkey': self.formkey, - }) + if 'formkey' in tokens: + self.formkey = tokens['formkey'] + self.client.headers.update({ + 'Poe-Formkey': self.formkey, + }) + + self.load_bundle() + if proxy != [] or auto_proxy == True: self.select_proxy(proxy, auto_proxy=auto_proxy) elif proxy == [] and auto_proxy == False: @@ -74,6 +80,17 @@ def __del__(self): if self.client: self.client.close() + def load_bundle(self): + try: + webData = self.client.get(self.BASE_URL) + self.bundle = PoeBundle(webData.text) + self.formkey = self.bundle.get_form_key() + self.client.headers.update({ + 'Poe-Formkey': self.formkey, + }) + except Exception as e: + logger.error(f"Failed to load bundle. Reason: {e}") + def select_proxy(self, proxy: list, auto_proxy: bool=False): if proxy == [] and auto_proxy == True: if not PROXY: diff --git a/poe_api_wrapper/async_api.py b/poe_api_wrapper/async_api.py index a51deaf..c4348e1 100644 --- a/poe_api_wrapper/async_api.py +++ b/poe_api_wrapper/async_api.py @@ -24,6 +24,7 @@ generate_file ) from .queries import generate_payload +from .bundles import PoeBundle from .proxies import PROXY if PROXY: from .proxies import fetch_proxy @@ -41,8 +42,8 @@ def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): self.client = None if not ASYNC: raise ImportError("Please install Async version using 'pip install poe-api-wrapper[async]'") - if not {'p-b', 'p-lat', 'formkey'}.issubset(tokens): - raise ValueError("Please provide valid p-b, p-lat, and formkey") + if not {'p-b', 'p-lat'}.issubset(tokens): + raise ValueError("Please provide valid p-b and p-lat cookies") self.proxy = proxy self.auto_proxy = auto_proxy @@ -58,8 +59,8 @@ def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): self.message_generating = True self.ws_refresh = 3 self.groups = {} - self.formkey = self.tokens['formkey'] self.proxies = {} + self.bundle: PoeBundle = None self.client = AsyncClient(headers=self.HEADERS, timeout=60, http2=True) self.client.cookies.update({ @@ -74,11 +75,15 @@ def __init__(self, tokens: dict={}, proxy: list=[], auto_proxy: bool=False): 'cf_clearance': tokens['cf_clearance'] }) - self.client.headers.update({ - 'Poe-Formkey': self.formkey, - }) + if 'formkey' in tokens: + self.formkey = tokens['formkey'] + self.client.headers.update({ + 'Poe-Formkey': self.formkey, + }) async def create(self): + await self.load_bundle() + if self.proxy != [] or self.auto_proxy == True: await self.select_proxy(self.proxy, auto_proxy=self.auto_proxy) elif self.proxy == [] and self.auto_proxy == False: @@ -93,6 +98,17 @@ async def create(self): def __del__(self): if self.client: asyncio.get_event_loop().run_until_complete(self.client.aclose()) + + async def load_bundle(self): + try: + webData = await self.client.get(self.BASE_URL) + self.bundle = PoeBundle(webData.text) + self.formkey = self.bundle.get_form_key() + self.client.headers.update({ + 'Poe-Formkey': self.formkey, + }) + except Exception as e: + logger.error(f"Failed to load bundle. Reason: {e}") async def select_proxy(self, proxy: list, auto_proxy: bool=False): if proxy == [] and auto_proxy == True: diff --git a/poe_api_wrapper/bundles.py b/poe_api_wrapper/bundles.py new file mode 100644 index 0000000..3a28671 --- /dev/null +++ b/poe_api_wrapper/bundles.py @@ -0,0 +1,82 @@ +from httpx import Client +from bs4 import BeautifulSoup +from js2py import eval_js +from loguru import logger +import re + +class PoeBundle: + form_key_pattern = r"window\.([a-zA-Z0-9]+)=function\(\)\{return window" + window_secret_pattern = r'let useFormkeyDecode=[\s\S]*?(window\.[\w]+="[^"]+")' + static_pattern = r'static[^"]*\.js' + + def __init__(self, document: str): + self._window = "const window={document:{hack:1},navigator:{userAgent:'safari <3'}};" + self._src_scripts = [] + self._webpack_script: str = None + + self.init_window(document) + + def init_window(self, document: str): + # initialize the window object with document scripts + logger.info("Initializing web data") + + scripts = BeautifulSoup(document, "html.parser").find_all('script') + for script in scripts: + if (src := script.attrs.get("src")) and (src not in self._src_scripts): + if "_app" in src: + self.init_app(src) + if "buildManifest" in src: + self.extend_src_scripts(src) + elif "webpack" in src: + self._webpack_script = src + self.extend_src_scripts(src) + else: + self._src_scripts.append(src) + elif ("document." in script.text) or ("function" not in script.text): + continue + elif script.attrs.get("type") == "application/json": + continue + self._window += script.text + + logger.info("Web data initialized") + + def init_app(self, src: str): + script = self.load_src_script(src) + if not (window_secret_match := re.search(self.window_secret_pattern, script)): + raise RuntimeError("Failed to find window secret in js scripts") + + self._window += window_secret_match.group(1) + ';' + + def extend_src_scripts(self, manifest_src: str): + # extend src scripts list with static scripts from manifest + static_main_url = self.get_base_url(manifest_src) + manifest = self.load_src_script(manifest_src) + + matches = re.findall(self.static_pattern, manifest) + scr_list = [f"{static_main_url}{match}" for match in matches] + + self._src_scripts.extend(scr_list) + + @staticmethod + def load_src_script(src: str) -> str: + with Client() as client: + resp = client.get(src) + if resp.status_code != 200: + logger.warning(f"Failed to load script {src}, status code: {resp.status_code}") + return resp.text + + @staticmethod + def get_base_url(src: str) -> str: + return src.split("static/")[0] + + def get_form_key(self) -> str: + script = self._window + + match = re.search(self.form_key_pattern, script) + if not (secret := match.group(1)): + raise RuntimeError("Failed to parse form-key function in Poe document") + + script += f'window.{secret}().slice(0, 32);' + formkey = str(eval_js(script)) + logger.info(f"Retrieved formkey successfully: {formkey}") + return formkey diff --git a/poe_api_wrapper/utils.py b/poe_api_wrapper/utils.py index 754c8fd..775c78a 100644 --- a/poe_api_wrapper/utils.py +++ b/poe_api_wrapper/utils.py @@ -7,7 +7,6 @@ HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36 Edg/115.0.1901.203", "Accept": "*/*", - "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en;q=0.8,en-GB;q=0.7,en-US;q=0.6", "Sec-Ch-Ua": '"Microsoft Edge";v="123", "Not:A-Brand";v="8", "Chromium";v="123"', "Sec-Ch-Ua-Mobile": "?0", diff --git a/setup.py b/setup.py index c82db33..de1b9e8 100644 --- a/setup.py +++ b/setup.py @@ -4,7 +4,7 @@ base_path = Path(__file__).parent long_description = (base_path / "README.md").read_text(encoding='utf-8') -VERSION = '1.4.5' +VERSION = '1.4.6' DESCRIPTION = 'A simple, lightweight and efficient API wrapper for Poe.com' LONG_DESCRIPTION = '👾 A Python API wrapper for Poe.com. With this, you will have free access to ChatGPT, Claude, Llama, Gemini, Google-PaLM and more! 🚀' @@ -17,7 +17,7 @@ long_description=long_description, packages=find_packages(), python_requires=">=3.7", - install_requires=['httpx', 'websocket-client', 'requests_toolbelt', 'loguru', 'rich==13.3.4'], + install_requires=['httpx[http2]', 'websocket-client', 'requests_toolbelt', 'loguru', 'rich==13.3.4', 'bs4', 'Js2Py'], extras_require={ 'async': ['nest-asyncio'], 'proxy': ['ballyregan; python_version>="3.9"'],