Major Update: Auto Retrieving formkey + Fixed some bugs

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',

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:

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:

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

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",

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"'],