backup.py

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