import re import sys import json import random import mimetypes from uuid import uuid4 from curl_cffi import requests, CurlMime cookies = { 'pplx.visitor-id': 'da5c198a-2ba9-4616-ac21-b6fa92040084', 'cf_clearance': '1LBWAHcXulLwB77y37vdwnrvegwyjC40Kl1wrw7r9Lw-1755541849-1.2.1.1-S261jjmxFdy7Y_3ZYcgitpH9bWYg_NX_c0hlPzqHJ95v5qnHTVUWEyAgeLy7T5T2AgDw4lTb0GSBNSyKSFEX5jS2lHFucREliY6cnLCO9YzXP8EN5vLUFX77qDKLiLCRAXLnJjuKQ9Qsd6Ur1lfZmBXyPd0XINQQHm2jYPwAaQ3NtwnGhCUe4slbH86v14nI0n8JpsPxknSMrz9n3ekpJ.HRs1te.r3U2Yf7VDfnkpk', 'segmented-control-popover-studio': '1', 'sidebarHiddenHubs': '[]', '_gcl_au': '1.1.1446397632.1751723919', '__Secure-next-auth.session-token': 'eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..jQAlULuPuiTvm360.uaQKzrd2wL-A6NZui8s5m8mGHjXB5rzt5gMD_RczG8V5U5F2yjpdsK7khNmbNS3IGnvm3Gl6CHOk3kO8x-M07IQBVmHFEly56AyXmW67JvEhfhLCVd9i4O99lfkp7sP5p0e4ZbTHPQxWw4J1ECCEy3Jo_6tYU5qilZXPi2yBC7InNrw56QnajHRvp0BAO4vWoXu1xcHpEFw7zdSiYr78AXqIMl1grXBGMf3kC3EmKRxY2IcoqwxwoR6k3I0TKl7ikAzVKBOdkNg4xUCQ-1Oc-jGPqb8TjN7lS3Gm_TGu3G-paIDEi9pxe7jz3gij0-7DXZ7Op0qyA2xZC-rYzxTnhf7HKiD9jzc1HpXNARpmv-aF1PiBjk7NrgiJJ4RTLiwz7bTdZxg__HN59t4ESl2DCWIgb3VRT8RWKC1Ag6tH9d2eUWXAk41QTw.xmmT3Oe89Yt5OGSneeKigA', 'pplx.metadata': '{%22qc%22:32%2C%22qcu%22:134%2C%22qcm%22:3%2C%22qcc%22:131%2C%22qcr%22:21%2C%22qcco%22:0%2C%22qcdr%22:2%2C%22qcs%22:0%2C%22qcd%22:0%2C%22hli%22:true%2C%22hcga%22:false%2C%22hcds%22:false%2C%22hso%22:false%2C%22hfo%22:false%2C%22hsma%22:false%2C%22fqa%22:1751723941278%2C%22lqa%22:1755541885155}', '__podscribe_perplexityai_referrer': '_', '__podscribe_perplexityai_landing_url': 'https://www.perplexity.ai/?login-source=tryPro^&login-new=false', '_fbp': 'fb.1.1751723930574.245699921229618591', 'pplx.search-models-v3': '{%22research%22:%22pplx_alpha%22%2C%22search%22:%22grok4%22}', '__stripe_mid': 'a3785ab0-0fd2-4fe0-ac88-82067750fdb649476b', 'intercom-id-l2wyozh0': '314b3a1e-1c7a-45aa-9aa2-bf756467787b', 'intercom-device-id-l2wyozh0': 'acc4d380-2161-4a82-b7ee-91052897761d', '__ps_fva': '1753545417872', 'ph_phc_TXdpocbGVeZVm5VJmAsHTMrCofBQu3e0kN8HGMNGTVW_posthog': '%7B%22distinct_id%22%3A%2201984b16-8c3d-7985-9031-0eba8f3aea5a%22%2C%22%24sesid%22%3A%5B1753728391973%2C%220198525b-9adf-725c-b328-6ad837127661%22%2C1753728391903%5D%7D', 'pplx.search-models-v4': '{%22search%22:%22gpt5%22%2C%22studio%22:%22pplx_beta%22%2C%22research%22:%22pplx_alpha%22}', 'pplx.search-mode': 'search', 'AWSALB': 'fLbw6sBD+IJEFaiTlF7JVl6wEt23823Cl3alPQt54AFr3HfnKJUpT6C9ldk9lMWY0cOb17AvWNtKIvNaJtQajAonZ9Nqex5X1DD/cOw7zwxK/bR5MMOUpwzBYPh2', 'AWSALBCORS': 'fLbw6sBD+IJEFaiTlF7JVl6wEt23823Cl3alPQt54AFr3HfnKJUpT6C9ldk9lMWY0cOb17AvWNtKIvNaJtQajAonZ9Nqex5X1DD/cOw7zwxK/bR5MMOUpwzBYPh2', 'pplx.session-id': 'cce53e2e-92ee-4099-ac27-330e26b7dcb4', '__cf_bm': 'H3osiQt3qQQgC8h8nM2kNRaxYeazYB6W.CyhcIUX3DA-1755541822-1.0.1.1-v4aM0Aq3128mwnmb4pgr7iFiGWNYLVjMnmy4j7VEiFPMd_7gJAXzBnikeCXf7dcQTWdFXYnIrsgPLNVONa92DR4LRSdVxG7xHiXwvKkZuik', '__cflb': '02DiuDyvFMmK5p9jVbVnMNSKYZhUL9aGkzS68zLFxFyec', '_dd_s': 'aid=f49f2a6b-ebe2-4b4a-8d25-d2c25599faf8^&rum=2^&id=33ada3a7-23b4-47eb-9d27-18b317142245^&created=1755541822142^&expire=1755542784222^&logs=0', 'next-auth.csrf-token': '3a15f4623e3152ab5c9f3d650d98d7ae5c9ab45dd3db67ccaa05d0a4f9b568ce%7C0e255858f54f5ed36a6cc50ff633c2985895c76832674a80fc1865a07e707dc9', 'next-auth.callback-url': 'https%3A%2F%2Fwww.perplexity.ai%2Fapi%2Fauth%2Fsignin-callback%3Fredirect%3Dhttps%253A%252F%252Fwww.perplexity.ai', '__ps_sr': '_', '__ps_slu': 'https://www.perplexity.ai/search/hi-awP1MSR.TbGFArGJ5SPBHQ', '__stripe_sid': '88ba78bc-9624-470b-b461-10345b56c3cbfea377', '_rdt_uuid': '1751723930283.ed01281c-7332-44d4-85de-08299d6fcb90', } class Client: ''' A client for interacting with the Perplexity AI API. ''' def __init__(self, cookies={}): self.session = requests.Session(headers={ 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'accept-language': 'en-US,en;q=0.9', 'cache-control': 'max-age=0', 'dnt': '1', 'priority': 'u=0, i', 'sec-ch-ua': '"Not;A=Brand";v="24", "Chromium";v="128"', 'sec-ch-ua-arch': '"x86"', 'sec-ch-ua-bitness': '"64"', 'sec-ch-ua-full-version': '"128.0.6613.120"', 'sec-ch-ua-full-version-list': '"Not;A=Brand";v="24.0.0.0", "Chromium";v="128.0.6613.120"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-model': '""', 'sec-ch-ua-platform': '"Windows"', 'sec-ch-ua-platform-version': '"19.0.0"', 'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate', 'sec-fetch-site': 'same-origin', 'sec-fetch-user': '?1', 'upgrade-insecure-requests': '1', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36', }, cookies=cookies, impersonate='chrome') self.own = bool(cookies) self.copilot = 0 if not cookies else float('inf') self.file_upload = 0 if not cookies else float('inf') self.signin_regex = re.compile(r'"(https://www\.perplexity\.ai/api/auth/callback/email\?callbackUrl=.*?)"') self.timestamp = format(random.getrandbits(32), '08x') self.session.get('https://www.perplexity.ai/api/auth/session') def create_account(self, cookies): ''' Function to create a new account ''' while True: try: emailnator_cli = Emailnator(cookies) resp = self.session.post('https://www.perplexity.ai/api/auth/signin/email', data={ 'email': emailnator_cli.email, 'csrfToken': self.session.cookies.get_dict()['next-auth.csrf-token'].split('%')[0], 'callbackUrl': 'https://www.perplexity.ai/', 'json': 'true' }) if resp.ok: new_msgs = emailnator_cli.reload(wait_for=lambda x: x['subject'] == 'Sign in to Perplexity', timeout=20) if new_msgs: break else: print('Perplexity account creating error:', resp) except Exception: pass msg = emailnator_cli.get(func=lambda x: x['subject'] == 'Sign in to Perplexity') new_account_link = self.signin_regex.search(emailnator_cli.open(msg['messageID'])).group(1) self.session.get(new_account_link) self.copilot = 5 self.file_upload = 10 return True def search(self, query, mode='auto', model=None, sources=['web'], files={}, stream=False, language='en-US', follow_up=None, incognito=False): ''' Query function ''' assert mode in ['auto', 'pro', 'reasoning', 'deep research'], 'Search modes -> ["auto", "pro", "reasoning", "deep research"]' # assert model in { # 'auto': [None], # 'pro': [None, 'sonar', 'gpt-4.5', 'gpt-4o', 'claude 3.7 sonnet', 'gemini 2.0 flash', 'grok-2'], # 'reasoning': [None, 'r1', 'o3-mini', 'claude 3.7 sonnet'], # 'deep research': [None] # }[mode] if self.own else True, '''Models for modes -> { # 'auto': [None], # 'pro': [None, 'sonar', 'gpt-4.5', 'gpt-4o', 'claude 3.7 sonnet', 'gemini 2.0 flash', 'grok-2'], # 'reasoning': [None, 'r1', 'o3-mini', 'claude 3.7 sonnet','grok4'], # 'deep research': [None] # }''' assert all([source in ('web', 'scholar', 'social') for source in sources]), 'Sources -> ["web", "scholar", "social"]' assert self.copilot > 0 if mode in ['pro', 'reasoning', 'deep research'] else True, 'You have used all of your enhanced (pro) queries' assert self.file_upload - len(files) >= 0 if files else True, f'You have tried to upload {len(files)} files but you have {self.file_upload} file upload(s) remaining.' self.copilot = self.copilot - 1 if mode in ['pro', 'reasoning', 'deep research'] else self.copilot self.file_upload = self.file_upload - len(files) if files else self.file_upload uploaded_files = [] for filename, file in files.items(): file_type = mimetypes.guess_type(filename)[0] file_upload_info = (self.session.post( 'https://www.perplexity.ai/rest/uploads/create_upload_url?version=2.18&source=default', json={ 'content_type': file_type, 'file_size': sys.getsizeof(file), 'filename': filename, 'force_image': False, 'source': 'default', } )).json() mp = CurlMime() for key, value in file_upload_info['fields'].items(): mp.addpart(name=key, data=value) mp.addpart(name='file', content_type=file_type, filename=filename, data=file) upload_resp = self.session.post(file_upload_info['s3_bucket_url'], multipart=mp) if not upload_resp.ok: raise Exception('File upload error', upload_resp) if 'image/upload' in file_upload_info['s3_object_url']: uploaded_url = re.sub( r'/private/s--.*?--/v\d+/user_uploads/', '/private/user_uploads/', upload_resp.json()['secure_url'] ) else: uploaded_url = file_upload_info['s3_object_url'] uploaded_files.append(uploaded_url) json_data = { 'query_str': query, 'params': { 'attachments': uploaded_files + follow_up['attachments'] if follow_up else uploaded_files, 'frontend_context_uuid': str(uuid4()), 'frontend_uuid': str(uuid4()), 'is_incognito': incognito, 'language': language, 'last_backend_uuid': follow_up['backend_uuid'] if follow_up else None, 'mode': 'concise' if mode == 'auto' else 'copilot', 'model_preference': { 'auto': { None: 'turbo' }, 'pro': { None: 'pplx_pro', 'sonar': 'experimental', 'gpt-4.5': 'gpt45', 'gpt-4o': 'gpt4o', 'claude 3.7 sonnet': 'claude2', 'gemini 2.0 flash': 'gemini2flash', 'grok-2': 'grok' }, 'reasoning': { None: 'pplx_reasoning', 'r1': 'r1-1776', 'o3-mini': 'o3mini', 'grok-4': 'grok4', 'gpt-4.1': 'gpt5', 'gemini-2.5-pro': 'gemini2flash', 'o3':'o3', 'claude-sonnet-4-20250514':'claude37sonnetthinking', 'sonar-pro':'pplx_alpha' }, 'deep research': { None: 'pplx_alpha' } }[mode][model], 'source': 'default', 'sources': sources, 'version': '2.18' } } resp = self.session.post('https://www.perplexity.ai/rest/sse/perplexity_ask', json=json_data, stream=True) chunks = [] # Generator for streaming responses def stream_response(resp): for chunk in resp.iter_lines(delimiter=b'\r\n\r\n'): content = chunk.decode('utf-8') if content.startswith('event: message\r\n'): # Parse the JSON data from the message content_json = json.loads(content[len('event: message\r\ndata: '):]) # Yield the entire chunk for the consumer to process yield content_json elif content.startswith('event: end_of_stream\r\n'): # End the generator when the stream is finished return # If streaming is requested, return the generator if stream: return stream_response(resp) # Non-streaming: process the full response to find the final answer full_answer = None for chunk in resp.iter_lines(delimiter=b'\r\n\r\n'): content = chunk.decode('utf-8') if content.startswith('event: message\r\n'): content_json = json.loads(content[len('event: message\r\ndata: '):]) # Check for the final answer within the new 'blocks' structure for block in content_json.get('blocks', []): if block.get('intended_usage') == 'ask_text': markdown = block.get('markdown_block', {}) if markdown.get('progress') == 'DONE' and 'answer' in markdown: full_answer = markdown['answer'] elif content.startswith('event: end_of_stream\r\n'): # Stop processing once the stream ends break # Return the final answer, or None if not found if full_answer is not None: return full_answer else: print("No full response.") return None # ## # perplexity_cli=Client() # # incognito = Enables incognito mode, for people who are using their own account # resp = perplexity_cli.search('Create an image on happiness', mode='auto', model=None, sources=[], files={}, stream=True, language='en-US', follow_up=None, incognito=False) # for i in resp: # try: # print(i["blocks"][0]["markdown_block"]["chunks"][0],end="") # # print(i["blocks"][0]["markdown_block"]["chunks"][0],end="") # except: # pass