Created api wrapper for inflection pi

Author: hansfzlorenzana

.gitignore

@@ -152,3 +152,4 @@ cython_debug/
 #.idea/
 
 # Added files to ignore
+test.py

example.py

@@ -0,0 +1,9 @@
+from pi import Pi
+
+prompt = "How are you?"
+
+chatbot = Pi("TOKEN HERE", proxy=False)
+
+response = chatbot.send_message(prompt)
+
+print(response)

setup.py

@@ -0,0 +1,31 @@
+from setuptools import setup, find_packages
+from pathlib import Path
+
+base_path = Path(__file__).parent
+long_description = (base_path / "README.md").read_text(encoding='utf-8')
+
+VERSION = '0.1.1'
+DESCRIPTION = 'A reverse engineered API for Inflection AI Personal Intelligence, called PI'
+LONG_DESCRIPTION = 'A reverse engineered API for Inflection AI Personal Intelligence, called PI'
+
+setup(
+    name="inflection-pi-api",
+    version=VERSION,
+    author="hansfzlorenzana",
+    description=DESCRIPTION,
+    long_description_content_type="text/markdown",
+    long_description=long_description,
+    packages=find_packages(),
+    python_requires=">=3.7",
+    install_requires=['httpx', 'websocket-client', 'requests_toolbelt', 'loguru'],
+    extras_require={
+        'proxy': ['ballyregan; python_version>="3.9"']
+    },
+    keywords=['python', 'pi-api', 'inflection', 'chatgpt', 'pi', 'ai', 'api', 'chatbot'],
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: GNU General Public License (GPL)",
+        "Operating System :: OS Independent"
+    ],
+    url="https://github.com/hansfzlorenzana/Inflection-Pi-API"
+)

src/__init__.py

@@ -0,0 +1,67 @@
+import json
+from curl_cffi import requests
+import re
+from proxies import fetch_proxy
+from time import sleep
+from httpx import Client
+
+
+class Pi:
+
+  def __init__(self, cookie: str, proxy: bool=False):
+
+    if proxy == True:
+        proxies = fetch_proxy()
+        for p in range(len(proxies)):
+            try:
+                self.client = Client(timeout=180, proxies= {"http://": f"{proxies[p]}"})
+                print(f"Connection established with {proxies[p]}")
+                break
+            except:
+                print(f"Connection failed with {proxies[p]}. Trying {p+1}/{len(proxies)} ...")
+                sleep(1)
+    else:
+        self.client = Client(timeout=180)
+    self.cookie = cookie
+
+  def send_message(self, prompt):
+
+    payload = json.dumps({"text": prompt})
+
+    url = "https://pi.ai/api/chat"
+
+    headers = {
+            'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
+            'Accept-Language': 'en-US,en;q=0.7',
+            'Referer': 'https://pi.ai/api/chat',
+            'Content-Type': 'application/json',
+            'Sec-Fetch-Dest': 'empty',
+            'Sec-Fetch-Mode': 'cors',
+            'Sec-Fetch-Site': 'same-origin',
+            'Connection': 'keep-alive',
+            'Cookie': f'{self.cookie}'
+    }
+
+    response = requests.post( url, headers=headers, data=payload,impersonate="chrome110",timeout=500)
+    decoded_data = response.content.decode("utf-8")
+
+    decoded_data = re.sub('\n+', '\n', decoded_data).strip()
+    data_strings = decoded_data.split('\n')
+    completions = []
+
+    for data_string in data_strings:
+        # Check if the string starts with 'data:'
+        if data_string.startswith('data:'):
+            # Extract the JSON data part
+            json_str = data_string[6:].strip()
+            try:
+                data = json.loads(json_str)
+                if 'text' in data:
+                    completions.append(data['text'])
+            except json.JSONDecodeError:
+                # Handle invalid JSON here (skip or log the error)
+                pass
+
+    answer = ''.join(completions)
+
+    return answer

src/pi.py

@@ -0,0 +1,72 @@
+import json
+import re
+from proxies import fetch_proxy
+from time import sleep
+from httpx import Client
+
+
+class Pi:
+
+  def __init__(self, cookie: str, proxy: bool=False):
+
+    if proxy == True:
+        proxies = fetch_proxy()
+        for p in range(len(proxies)):
+            try:
+                self.client = Client(timeout=180, proxies= {"http://": f"{proxies[p]}"})
+                print(f"Connection established with {proxies[p]}")
+                break
+            except:
+                print(f"Connection failed with {proxies[p]}. Trying {p+1}/{len(proxies)} ...")
+                sleep(1)
+    else:
+        self.client = Client(timeout=180)
+    self.cookie = cookie
+
+  def send_message(self, prompt):
+
+    payload = json.dumps({"text": prompt})
+
+    url = "https://pi.ai/api/chat"
+
+    headers = {
+            'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
+            'Host': 'www.pi.ai',
+            'Accept': '*/*',
+            'Accept-Language': 'en-US,en;q=0.7',
+            'Referer': 'https://pi.ai/api/chat',
+            'Content-Type': 'application/json',
+            'Sec-Fetch-Dest': 'empty',
+            'Sec-Fetch-Mode': 'cors',
+            'Sec-Fetch-Site': 'same-origin',
+            'Connection': 'keep-alive',
+            'Cookie': f'{self.cookie}'
+    }
+
+    try:
+        response = self.client.post(url, headers=headers, content=payload, timeout=500)
+        response.raise_for_status()
+
+        decoded_data = response.content.decode("utf-8")
+        decoded_data = re.sub('\n+', '\n', decoded_data).strip()
+        data_strings = decoded_data.split('\n')
+        completions = []
+
+        for data_string in data_strings:
+            if data_string.startswith('data:'):
+                json_str = data_string[6:].strip()
+                try:
+                    data = json.loads(json_str)
+                    if 'text' in data:
+                        completions.append(data['text'])
+                except json.JSONDecodeError:
+                    pass
+
+        answer = ''.join(completions)
+        return answer
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+        return None
+    finally:
+        self.client.close()

src/pi_httpx.py

@@ -0,0 +1,32 @@
+from ballyregan import ProxyFetcher
+from ballyregan.models import Protocols, Anonymities
+from fp.fp import FreeProxy
+
+def fetch_proxy():
+    fetcher = ProxyFetcher()
+    try:
+        proxies = fetcher.get(
+        limit=10,
+        protocols=[Protocols.HTTP],
+        anonymities=[Anonymities.ANONYMOUS],
+        )
+    except:
+        print("No Anonymous proxies found. Switching to normal proxies ...") 
+        proxies = fetcher.get(
+        limit=10,
+        protocols=[Protocols.HTTP],
+        )
+    return proxies
+
+def fetch_free_proxy():
+    try:
+        proxies = FreeProxy(country_id=['US'], 
+                            anonym=True,
+                            elite=True
+                            ).get()
+    except:
+        print("No Anonymous proxies found. Switching to normal proxies ...") 
+        proxies = FreeProxy(country_id=['US']
+                            ).get()
+    return proxies
+