Upload 5 files

__init__.py

@@ -0,0 +1,5 @@
+name: str = "perplexity"
+
+from .utils import *
+from .labs import Labs
+from .perplexity import Perplexity

labs.py

@@ -0,0 +1,113 @@
+from os import listdir
+from uuid import uuid4
+from time import sleep, time
+from threading import Thread
+from json import loads, dumps
+from random import getrandbits
+from websocket import WebSocketApp
+from requests import Session, get, post
+
+class Labs:
+    def __init__(self) -> None:
+        self.history: list = []
+        self.session: Session = Session()
+        self.user_agent: dict = { "User-Agent": "Ask/2.2.1/334 (iOS; iPhone) isiOSOnMac/false", "X-Client-Name": "Perplexity-iOS" }
+        self.session.headers.update(self.user_agent)
+        self._init_session_without_login()
+
+        self.t: str = self._get_t()
+        self.sid: str = self._get_sid()
+    
+        self.queue: list = []
+        self.finished: bool = True
+
+        assert self._ask_anonymous_user(), "failed to ask anonymous user"
+        self.ws: WebSocketApp = self._init_websocket()
+        self.ws_thread: Thread = Thread(target=self.ws.run_forever).start()
+        self._auth_session()
+
+        while not (self.ws.sock and self.ws.sock.connected):
+            sleep(0.01)
+
+    def _init_session_without_login(self) -> None:
+        self.session.get(url=f"https://www.perplexity.ai/search/{str(uuid4())}")
+        self.session.headers.update(self.user_agent)
+    
+    def _auth_session(self) -> None:
+        self.session.get(url="https://www.perplexity.ai/api/auth/session")
+
+    def _get_t(self) -> str:
+        return format(getrandbits(32), "08x")
+
+    def _get_sid(self) -> str:
+        return loads(self.session.get(
+            url=f"https://labs-api.perplexity.ai/socket.io/?transport=polling&EIO=4"
+        ).text[1:])["sid"]
+
+    def _ask_anonymous_user(self) -> bool:
+        response = self.session.post(
+            url=f"https://labs-api.perplexity.ai/socket.io/?EIO=4&transport=polling&t={self.t}&sid={self.sid}",
+            data="40{\"jwt\":\"anonymous-ask-user\"}"
+        ).text
+
+        return response == "OK"
+
+    def _get_cookies_str(self) -> str:
+        cookies = ""
+        for key, value in self.session.cookies.get_dict().items():
+            cookies += f"{key}={value}; "
+        return cookies[:-2]
+
+    def _init_websocket(self) -> WebSocketApp:
+        def on_open(ws: WebSocketApp) -> None:
+            ws.send("2probe")
+            ws.send("5")
+
+        def on_message(ws: WebSocketApp, message: str) -> None:
+            if message == "2":
+                ws.send("3")
+            elif message.startswith("42"):
+                message = loads(message[2:])[1]
+                if "status" not in message:
+                    self.queue.append(message)
+                elif message["status"] == "completed":
+                    self.finished = True
+                    self.history.append({"role": "assistant", "content": message["output"], "priority": 0})
+                elif message["status"] == "failed":
+                    self.finished = True
+
+        headers: dict = self.user_agent
+        headers["Cookie"] = self._get_cookies_str()
+
+        return WebSocketApp(
+            url=f"wss://labs-api.perplexity.ai/socket.io/?EIO=4&transport=websocket&sid={self.sid}",
+            header=headers,
+            on_open=on_open,
+            on_message=on_message,
+            on_error=lambda ws, err: print(f"websocket error: {err}")
+        )
+    
+    def _c(self, prompt: str, model: str) -> dict:
+        assert self.finished, "already searching"
+        assert model in ["codellama-34b-instruct", "llama-2-7b-chat", "llama-2-13b-chat", "llama-2-70b-chat", "mistral-7b-instruct", "pplx-70b-chat"]
+        self.finished = False
+        self.history.append({"role": "user", "content": prompt, "priority": 0})
+        self.ws.send("42[\"perplexity_playground\",{\"version\":\"2.1\",\"source\":\"default\",\"model\":\"" + model + "\",\"messages\":" + dumps(self.history) + "}]")
+    
+    def chat(self, prompt: str, model: str = "llama-2-7b-chat") -> dict:
+        self._c(prompt, model)
+
+        while (not self.finished) or (len(self.queue) != 0):
+            if len(self.queue) > 0:
+                yield self.queue.pop(0)
+
+    def chat_sync(self, prompt: str, model: str = "llama-2-7b-chat") -> dict:
+        self._c(prompt, model)
+
+        while not self.finished:
+            pass
+
+        return self.queue.pop(-1)
+
+    def close(self) -> None:
+        self.ws.close()

onkbtest.py

@@ -0,0 +1,110 @@
+import streamlit as st
+import pandas as pd
+import numpy as np
+import re
+import json
+import requests
+from io import StringIO
+from perplexity import Perplexity
+import codecs
+
+
+
+st.set_page_config(layout="wide")
+def dataread(kk, pasteduse,perplex_use):
+
+    if pasteduse != '':
+        data = pd.read_csv(StringIO(pasteduse),sep='\t',header=0)
+        data['Gene']=data['Gene_name']
+        data['Protein Change']=data['Protein Change'].apply(lambda x: str(x).replace('p.',''))
+
+    st.write('The data you input is the following:')
+    st.write(data)
+
+    all_onc=[]
+    all_perplex=[]
+    all_query=[]
+    querygeneprotein=[]
+    for index,row in data.iterrows():
+       
+        if str(row['Gene']) != 'nan' and str(row['Protein Change']) != 'nan':
+            
+            if str(row['Protein Change']) != 'nan':
+                d1=requests.get("https://www.oncokb.org/api/v1/annotate/mutations/byProteinChange?hugoSymbol="+row['Gene']+"&alteration="+row['Protein Change']+"&tumorType="+kk, headers={'Accept': 'application/json',"Authorization": 'Bearer 64f4aa64-2509-4500-994b-1f2a38422d44'})
+                if perplex_use:
+                    query="what drugs are used to treat "+ row['Gene']+" "+row['Protein Change']+" in "+kk+"?"
+                    perplexity = Perplexity()   
+                    answer=perplexity.search(query)
+                    all_perplex.append(answer)
+                    all_query.append(query)
+                all_onc.append(d1.content)
+                
+                querygeneprotein.append(row['Gene']+"&alteration="+row['Protein Change'])
+                #if perplex_use:
+                #    perplexity.close()
+
+    return all_onc,all_perplex,all_query,querygeneprotein
+
+
+with st.form(key='parameters'):
+
+
+    texttomatch=st.text_input('text to match',value='')
+    pasteduse=st.text_area('paste text to search',value='')
+    perplex_use=st.checkbox('Use Perplexity',value=False)
+    abbrev_perplex=st.checkbox('Abbreviate Perplexity',value=True)
+    submit_button = st.form_submit_button(label='Submit')
+if submit_button:
+
+    if pasteduse != '':
+        kk,perplexity, query,querygeneprotein=dataread(texttomatch,pasteduse,perplex_use)
+        dictionary_of_json_fda={}
+        dictionary_of_json_text={}
+        dictionary_of_json_text_perplexity={}
+        dictionary_of_json_query={}
+        dictionary_of_drugs={}
+        for i in np.arange(0,len(kk)):
+            output = codecs.decode(kk[i])
+            dictionary_of_json_fda[querygeneprotein[i]]=json.loads(output).get('highestFdaLevel')
+            #dictionary_of_drugs[querygeneprotein[i]]=json.loads(output).get('treatments').get('drugs').get('DrugName')
+            dictionary_of_json_text[querygeneprotein[i]]=json.loads(output)
+            dictionary_of_json_text_perplexity[querygeneprotein[i]]=perplexity[i]
+            dictionary_of_json_query[querygeneprotein[i]]=query[i]
+        ord_dict=sorted(dictionary_of_json_fda.items(), key=lambda item: str(item[1]))
+        
+        tt=st.tabs(pd.DataFrame(ord_dict).apply(lambda n: str(n[0])+' '+str(n[1]),axis=1).to_list())
+        counter=0
+       
+        for i in ord_dict:
+            with tt[counter]:
+                st.write('The query was '+i[0])
+                st.write('The results were:')   
+                if dictionary_of_json_text[i[0]]['highestFdaLevel'] != '':
+                    st.write('Drugs in this result')
+                    drugnames=[]
+                    for d in dictionary_of_json_text[i[0]]['treatments']:
+                        for dd in d['drugs']:
+                            drugnames.append(dd['drugName'])
+                    for m in (set(drugnames)):
+                        st.write(m)
+                    st.write(str(len(set(drugnames)))+' drugs in this result')
+                if perplex_use:
+                    st.write('The perplexity query was '+dictionary_of_json_query[i[0]])
+                    
+                    
+                    for jj in dictionary_of_json_text_perplexity[i[0]]:   
+                        forout=jj
+                    st.write('answer from perplexity')              
+                    st.write(forout["answer"])
+                    for m in (set(drugnames)):
+                        if(re.findall(m,forout["answer"])) !=[]:
+                            st.write('YES BOTH '+m+' is in the answer of perplexity')
+                        else:
+                            st.write('NO JUST ONCOKB '+m+' is not in the answer of perplexity')    
+                st.write('all oncokb results')            
+                st.json(dictionary_of_json_text[i[0]])  
+                if perplex_use and abbrev_perplex ==False:  
+                    st.write('All results from perplexity:')
+                    st.write(forout)
+                    #breakpoint()    
+                counter=counter+1

perplexity.py

@@ -0,0 +1,312 @@
+from typing import Iterable, Dict
+
+from os import listdir
+from uuid import uuid4
+from time import sleep, time
+from threading import Thread
+from json import loads, dumps
+from random import getrandbits
+from websocket import WebSocketApp
+from requests import Session, get, post
+
+class Perplexity:
+    def __init__(self, email: str = None) -> None:
+        self.session: Session = Session()
+        self.user_agent: dict = { "User-Agent": "Ask/2.4.1/224 (iOS; iPhone; Version 17.1) isiOSOnMac/false", "X-Client-Name": "Perplexity-iOS" }
+        self.session.headers.update(self.user_agent)
+
+        if email and ".perplexity_session" in listdir():
+            self._recover_session(email)
+        else:
+            self._init_session_without_login()
+
+            if email:
+                self._login(email)
+
+        self.email: str = email
+        self.t: str = self._get_t()
+        self.sid: str = self._get_sid()
+    
+        self.n: int = 1
+        self.base: int = 420
+        self.queue: list = []
+        self.finished: bool = True
+        self.last_uuid: str = None
+        self.backend_uuid: str = None # unused because we can't yet follow-up questions
+        self.frontend_session_id: str = str(uuid4())
+
+        assert self._ask_anonymous_user(), "failed to ask anonymous user"
+        self.ws: WebSocketApp = self._init_websocket()
+        self.ws_thread: Thread = Thread(target=self.ws.run_forever).start()
+        self._auth_session()
+
+        while not (self.ws.sock and self.ws.sock.connected):
+            sleep(0.01)
+
+    def _recover_session(self, email: str) -> None:
+        with open(".perplexity_session", "r") as f:
+            perplexity_session: dict = loads(f.read())
+
+        if email in perplexity_session:
+            self.session.cookies.update(perplexity_session[email])
+        else:
+            self._login(email, perplexity_session)
+    
+    def _login(self, email: str, ps: dict = None) -> None:
+        self.session.post(url="https://www.perplexity.ai/api/auth/signin-email", data={"email": email})
+
+        email_link: str = str(input("paste the link you received by email: "))
+        self.session.get(email_link)
+
+        if ps:
+            ps[email] = self.session.cookies.get_dict()
+        else:
+            ps = {email: self.session.cookies.get_dict()}
+
+        with open(".perplexity_session", "w") as f:
+            f.write(dumps(ps))
+
+    def _init_session_without_login(self) -> None:
+        self.session.get(url=f"https://www.perplexity.ai/search/{str(uuid4())}")
+        self.session.headers.update(self.user_agent)
+    
+    def _auth_session(self) -> None:
+        self.session.get(url="https://www.perplexity.ai/api/auth/session")
+
+    def _get_t(self) -> str:
+        return format(getrandbits(32), "08x")
+
+    def _get_sid(self) -> str:
+        return loads(self.session.get(
+            url=f"https://www.perplexity.ai/socket.io/?EIO=4&transport=polling&t={self.t}"
+        ).text[1:])["sid"]
+
+    def _ask_anonymous_user(self) -> bool:
+        response = self.session.post(
+            url=f"https://www.perplexity.ai/socket.io/?EIO=4&transport=polling&t={self.t}&sid={self.sid}",
+            data="40{\"jwt\":\"anonymous-ask-user\"}"
+        ).text
+
+        return response == "OK"
+
+    def _start_interaction(self) -> None:
+        self.finished = False
+
+        if self.n == 9:
+            self.n = 0
+            self.base *= 10
+        else:
+            self.n += 1
+
+        self.queue = []
+
+    def _get_cookies_str(self) -> str:
+        cookies = ""
+        for key, value in self.session.cookies.get_dict().items():
+            cookies += f"{key}={value}; "
+        return cookies[:-2]
+    
+    def _write_file_url(self, filename: str, file_url: str) -> None:
+        if ".perplexity_files_url" in listdir():
+            with open(".perplexity_files_url", "r") as f:
+                perplexity_files_url: dict = loads(f.read())
+        else:
+            perplexity_files_url: dict = {}
+
+        perplexity_files_url[filename] = file_url
+
+        with open(".perplexity_files_url", "w") as f:
+            f.write(dumps(perplexity_files_url))
+
+    def _init_websocket(self) -> WebSocketApp:
+        def on_open(ws: WebSocketApp) -> None:
+            ws.send("2probe")
+            ws.send("5")
+
+        def on_message(ws: WebSocketApp, message: str) -> None:
+            if message == "2":
+                ws.send("3")
+            elif not self.finished:
+                if message.startswith("42"):
+                    message : list = loads(message[2:])
+                    content: dict = message[1]
+                    if "mode" in content and content["mode"] == "copilot":
+                        content["copilot_answer"] = loads(content["text"])
+                    elif "mode" in content:
+                        content.update(loads(content["text"]))
+                    content.pop("text")
+                    if (not ("final" in content and content["final"])) or ("status" in content and content["status"] == "completed"):
+                        self.queue.append(content)
+                    if message[0] == "query_answered":
+                        self.last_uuid = content["uuid"]
+                        self.finished = True
+                elif message.startswith("43"):
+                    message: dict = loads(message[3:])[0]
+                    if ("uuid" in message and message["uuid"] != self.last_uuid) or "uuid" not in message:
+                        self.queue.append(message)
+                        self.finished = True
+
+        return WebSocketApp(
+            url=f"wss://www.perplexity.ai/socket.io/?EIO=4&transport=websocket&sid={self.sid}",
+            header=self.user_agent,
+            cookie=self._get_cookies_str(),
+            on_open=on_open,
+            on_message=on_message,
+            on_error=lambda ws, err: print(f"websocket error: {err}")
+        )
+    
+    def _s(self, query: str, mode: str = "concise", search_focus: str = "internet", attachments: list[str] = [], language: str = "en-GB", in_page: str = None, in_domain: str = None) -> None:
+        assert self.finished, "already searching"
+        assert mode in ["concise", "copilot"], "invalid mode"
+        assert len(attachments) <= 4, "too many attachments: max 4"
+        assert search_focus in ["internet", "scholar", "writing", "wolfram", "youtube", "reddit"], "invalid search focus"
+
+        if in_page:
+            search_focus = "in_page"
+        if in_domain:
+            search_focus = "in_domain"
+
+        self._start_interaction()
+        ws_message: str = f"{self.base + self.n}" + dumps([
+            "perplexity_ask",
+            query,
+            {
+                "version": "2.1",
+                "source": "default", # "ios"
+                "frontend_session_id": self.frontend_session_id,
+                "language": language,
+                "timezone": "CET",
+                "attachments": attachments,
+                "search_focus": search_focus,
+                "frontend_uuid": str(uuid4()),
+                "mode": mode,
+                # "use_inhouse_model": True
+                "in_page": in_page,
+                "in_domain": in_domain
+            }
+        ])
+
+        self.ws.send(ws_message)
+
+    def search(self, query: str, mode: str = "concise", search_focus: str = "internet", attachments: list[str] = [], language: str = "en-GB", timeout: float = 30) -> Iterable[Dict]:
+        self._s(query, mode, search_focus, attachments, language)
+
+        start_time: float = time()
+        while (not self.finished) or len(self.queue) != 0:
+            if timeout and time() - start_time > timeout:
+                self.finished = True
+                return {"error": "timeout"}
+            if len(self.queue) != 0:
+                yield self.queue.pop(0)
+
+    def search_sync(self, query: str, mode: str = "concise", search_focus: str = "internet", attachments: list[str] = [], language: str = "en-GB", timeout: float = 30) -> dict:
+        self._s(query, mode, search_focus, attachments, language)
+
+        start_time: float = time()
+        while not self.finished:
+            if timeout and time() - start_time > timeout:
+                self.finished = True
+                return {"error": "timeout"}
+        
+        return self.queue.pop(-1)
+
+    def upload(self, filename: str) -> str:
+        assert self.finished, "already searching"
+        assert filename.split(".")[-1] in ["txt", "pdf"], "invalid file format"
+
+        if filename.startswith("http"):
+            file = get(filename).content
+        else:
+            with open(filename, "rb") as f:
+                file = f.read()
+
+        self._start_interaction()
+        ws_message: str = f"{self.base + self.n}" + dumps([
+            "get_upload_url",
+            {
+                "version": "2.1",
+                "source": "default",
+                "content_type": "text/plain" if filename.split(".")[-1] == "txt" else "application/pdf",
+            }
+        ])
+
+        self.ws.send(ws_message)
+
+        while not self.finished or len(self.queue) != 0:
+            if len(self.queue) != 0:
+                upload_data = self.queue.pop(0)
+
+        assert not upload_data["rate_limited"], "rate limited"
+        
+        post(
+            url=upload_data["url"],
+            files={
+                "acl": (None, upload_data["fields"]["acl"]),
+                "Content-Type": (None, upload_data["fields"]["Content-Type"]),
+                "key": (None, upload_data["fields"]["key"]),
+                "AWSAccessKeyId": (None, upload_data["fields"]["AWSAccessKeyId"]),
+                "x-amz-security-token": (None, upload_data["fields"]["x-amz-security-token"]),
+                "policy": (None, upload_data["fields"]["policy"]),
+                "signature": (None, upload_data["fields"]["signature"]),
+                "file": (filename, file)
+            }
+        )
+
+        file_url: str = upload_data["url"] + upload_data["fields"]["key"].split("$")[0] + filename
+
+        self._write_file_url(filename, file_url)
+
+        return file_url
+    
+    def threads(self, query: str = None, limit: int = None) -> list[dict]:
+        assert self.email, "not logged in"
+        assert self.finished, "already searching"
+
+        if not limit: limit = 20
+        data: dict = {"version": "2.1", "source": "default", "limit": limit, "offset": 0}
+        if query: data["search_term"] = query
+
+        self._start_interaction()
+        ws_message: str = f"{self.base + self.n}" + dumps([
+            "list_ask_threads",
+            data
+        ])
+
+        self.ws.send(ws_message)
+
+        while not self.finished or len(self.queue) != 0:
+            if len(self.queue) != 0:
+                return self.queue.pop(0)
+            
+    def list_autosuggest(self, query: str = "", search_focus: str = "internet") -> list[dict]:
+        assert self.finished, "already searching"
+
+        self._start_interaction()
+        ws_message: str = f"{self.base + self.n}" + dumps([
+            "list_autosuggest",
+            query,
+            {
+                "has_attachment": False,
+                "search_focus": search_focus,
+                "source": "default",
+                "version": "2.1"
+            }
+        ])
+
+        self.ws.send(ws_message)
+
+        while not self.finished or len(self.queue) != 0:
+            if len(self.queue) != 0:
+                return self.queue.pop(0)
+    
+    def close(self) -> None:
+        self.ws.close()
+
+        if self.email:
+            with open(".perplexity_session", "r") as f:
+                perplexity_session: dict = loads(f.read())
+
+            perplexity_session[self.email] = self.session.cookies.get_dict()
+
+            with open(".perplexity_session", "w") as f:
+                f.write(dumps(perplexity_session))

utils.py

@@ -0,0 +1,11 @@
+from typing import Iterable, Dict
+
+def return_just_next_token(answer: Iterable[Dict]) -> str:
+    length = 0
+    for partial_answer in answer:
+        if "answer" in partial_answer:
+            yield partial_answer["answer"][length:]
+            length = len(partial_answer["answer"])
+        elif "output" in partial_answer:
+            yield partial_answer["output"][length:]
+            length = len(partial_answer["output"])