Merge pull request #2 from ramonvc/sourcery/main

g4f/Provider/Providers/Aws.py

@@ -12,7 +12,7 @@ class Auth(requests.auth.AuthBase):
         self.token = 'sk-1234567890abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL'
         
     def __call__(self, r):
-        r.headers["authorization"] = "Bearer " + self.token
+        r.headers["authorization"] = f"Bearer {self.token}"
         return r
 
 def _create_completion(model: str, messages: list, stream: bool, **kwargs):

g4f/Provider/Providers/Bard.py

@@ -10,17 +10,17 @@ supports_stream = False
 def _create_completion(model: str, messages: list, stream: bool, **kwargs):
     psid = {cookie.name: cookie.value for cookie in browser_cookie3.chrome(
         domain_name='.google.com')}['__Secure-1PSID']
-    
-    formatted = '\n'.join([
-        '%s: %s' % (message['role'], message['content']) for message in messages
-    ])
+
+    formatted = '\n'.join(
+        [f"{message['role']}: {message['content']}" for message in messages]
+    )
     prompt = f'{formatted}\nAssistant:'
 
     proxy = None
-    
-    if proxy == None:
+
+    if proxy is None:
         raise Exception('Proxy is required for Bard (set in g4f/Provider/Providers/Bard.py line 18)')
-    
+
     snlm0e = False
     conversation_id = None
     response_id = None
@@ -41,8 +41,13 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
         'cookie': f'__Secure-1PSID={psid}'
     }
 
-    snlm0e = re.search(r'SNlM0e\":\"(.*?)\"',
-                    client.get('https://bard.google.com/').text).group(1) if not snlm0e else snlm0e
+    snlm0e = (
+        re.search(
+            r'SNlM0e\":\"(.*?)\"', client.get('https://bard.google.com/').text
+        )[1]
+        if not snlm0e
+        else snlm0e
+    )
 
     params = {
         'bl': 'boq_assistant-bard-web-server_20230326.21_p0',
@@ -63,12 +68,11 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
     response = client.post(f'https://bard.google.com/_/BardChatUi/data/{intents}/StreamGenerate',
                         data=data, params=params)
 
-    chat_data = json.loads(response.content.splitlines()[3])[0][2]
-    if chat_data:
+    if chat_data := json.loads(response.content.splitlines()[3])[0][2]:
         json_chat_data = json.loads(chat_data)
 
         yield json_chat_data[0][0]
-        
+
     else:
         yield 'error'
 

g4f/Provider/Providers/Forefront.py

@@ -26,8 +26,7 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
 
     for token in response.iter_lines():
         if b'delta' in token:
-            token = json.loads(token.decode().split('data: ')[1])['delta']
-            yield (token)
+            yield json.loads(token.decode().split('data: ')[1])['delta']
 
 
 params = f'g4f.Providers.{os.path.basename(__file__)[:-3]} supports: ' + \

g4f/Provider/Providers/Phind.py

@@ -33,9 +33,7 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
             yield 'Clouflare error, please try again...'
             os._exit(0)
 
-        else:
-            if b'ping - 2023-' in line:
-                continue
+        elif b'ping - 2023-' not in line:
             yield line.decode('utf-8', errors='ignore')  # [:-1]
 
 

g4f/Provider/Providers/Pierangelo.py

@@ -45,11 +45,12 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
         'temperature': 0.7
     }
 
-    response = requests.post('https://chat.pierangelo.info/api/chat', 
-                             headers=headers, json=json_data, stream=True)
-    
-    for token in response:
-        yield (token)
+    yield from requests.post(
+        'https://chat.pierangelo.info/api/chat',
+        headers=headers,
+        json=json_data,
+        stream=True,
+    )
 
 params = f'g4f.Providers.{os.path.basename(__file__)[:-3]} supports: ' + \
     '(%s)' % ', '.join([f'{name}: {get_type_hints(_create_completion)[name].__name__}' for name in _create_completion.__code__.co_varnames[:_create_completion.__code__.co_argcount]])

g4f/Provider/Providers/Vercel.py

@@ -72,7 +72,7 @@ class Client:
         return {key: param['value'] for key, param in vercel_models[model_id]['parameters'].items()}
 
     def generate(self, model_id: str, prompt: str, params: dict = {}):
-        if not ':' in model_id:
+        if ':' not in model_id:
             model_id = models[model_id]
 
         defaults = self.get_default_params(model_id)
@@ -145,16 +145,13 @@ class Client:
 
 def _create_completion(model: str, messages: list, stream: bool, **kwargs):
     conversation = 'This is a conversation between a human and a language model, respond to the last message accordingly, referring to the past history of messages if needed.\n'
-    
+
     for message in messages:
         conversation += '%s: %s\n' % (message['role'], message['content'])
-    
+
     conversation += 'assistant: '
-    
-    completion = Client().generate(model, conversation)
 
-    for token in completion:
-        yield token
+    yield from Client().generate(model, conversation)
 
 params = f'g4f.Providers.{os.path.basename(__file__)[:-3]} supports: ' + \
     '(%s)' % ', '.join([f"{name}: {get_type_hints(_create_completion)[name].__name__}" for name in _create_completion.__code__.co_varnames[:_create_completion.__code__.co_argcount]])

g4f/Provider/Providers/Yqcloud.py

@@ -15,7 +15,7 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
     }
 
     json_data = {
-        'prompt': '%s' % messages[-1]['content'],
+        'prompt': f"{messages[-1]['content']}",
         'userId': f'#/chat/{int(time.time() * 1000)}',
         'network': True,
         'apikey': '',
@@ -25,7 +25,7 @@ def _create_completion(model: str, messages: list, stream: bool, **kwargs):
 
     response = requests.post('https://api.aichatos.cloud/api/generateStream', headers=headers, json=json_data, stream=True)
     for token in response.iter_content(chunk_size=2046):
-        if not b'always respond in english' in token:
+        if b'always respond in english' not in token:
             yield (token.decode('utf-8'))
 
 params = f'g4f.Providers.{os.path.basename(__file__)[:-3]} supports: ' + \

g4f/Provider/Providers/helpers/bing.py

@@ -76,19 +76,9 @@ def format(msg: dict) -> str:
 
 def get_token():
     return
-    
-    try:
-        cookies = {c.name: c.value for c in browser_cookie3.edge(domain_name='bing.com')}
-        return cookies['_U']
-    except:
-        print('Error: could not find bing _U cookie in edge browser.')
-        exit(1)
 
 class AsyncCompletion:
-    async def create(
-        prompt     : str = None,
-        optionSets : list = None,
-        token     : str = None): # No auth required anymore
+    async def create(self, optionSets : list = None, token     : str = None): # No auth required anymore
         
         create = None
         for _ in range(5):
@@ -128,9 +118,7 @@ class AsyncCompletion:
 
             except Exception as e:
                 time.sleep(0.5)
-                continue
-        
-        if create == None: raise Exception('Failed to create conversation.')
+        if create is None: raise Exception('Failed to create conversation.')
 
         wss: websockets.WebSocketClientProtocol or None = None
 
@@ -165,48 +153,46 @@ class AsyncCompletion:
         struct = {
             'arguments': [
                 {
-                    'source': 'cib', 
-                    'optionsSets': optionSets, 
-                    'isStartOfSession': True, 
+                    'source': 'cib',
+                    'optionsSets': optionSets,
+                    'isStartOfSession': True,
                     'message': {
-                        'author': 'user', 
-                        'inputMethod': 'Keyboard', 
-                        'text': prompt, 
-                        'messageType': 'Chat'
-                    }, 
-                    'conversationSignature': conversationSignature, 
-                    'participant': {
-                        'id': clientId
-                    }, 
-                    'conversationId': conversationId
+                        'author': 'user',
+                        'inputMethod': 'Keyboard',
+                        'text': self,
+                        'messageType': 'Chat',
+                    },
+                    'conversationSignature': conversationSignature,
+                    'participant': {'id': clientId},
+                    'conversationId': conversationId,
                 }
-            ], 
-            'invocationId': '0', 
-            'target': 'chat', 
-            'type': 4
+            ],
+            'invocationId': '0',
+            'target': 'chat',
+            'type': 4,
         }
-        
+
         await wss.send(format(struct))
-        
+
         base_string = ''
-        
+
         final = False
         while not final:
             objects = str(await wss.recv()).split('\x1e')
             for obj in objects:
                 if obj is None or obj == '':
                     continue
-                
+
                 response = json.loads(obj)
                 if response.get('type') == 1 and response['arguments'][0].get('messages',):
                     response_text = response['arguments'][0]['messages'][0]['adaptiveCards'][0]['body'][0].get('text')
-                    
+
                     yield (response_text.replace(base_string, ''))
                     base_string = response_text
-        
+
                 elif response.get('type') == 2:
                     final = True
-        
+
         await wss.close()
 
 async def run(optionSets, messages):

g4f/__init__.py

@@ -37,13 +37,13 @@ class ChatCompletion:
         try:
             if isinstance(model, str):
                 model = Utils.convert[model]
-            
+
             engine = model.best_site if not provider else provider
-            if not engine.supports_stream and stream == True:
+            if not engine.supports_stream and stream:
                 print(
                     f"ValueError: {engine.__name__} does not support 'stream' argument", file=sys.stderr)
                 sys.exit(1)
-            
+
             return (engine._create_completion(model.name, messages, stream, **kwargs)
                     if stream else ''.join(engine._create_completion(model.name, messages, stream, **kwargs)))
 

g4f/typing.py

@@ -3,13 +3,13 @@ from typing import Dict, NewType, Union, Optional, List, get_type_hints
 sha256 = NewType('sha_256_hash', str)
 
 class MetaModels(type):
-    def __str__(cls):
+    def __str__(self):
         output: List = [
             f'class Engines:\n',
-            f'  class {cls.gpt_35_turbo.__name__}:',
+            f'  class {self.gpt_35_turbo.__name__}:',
+            '    ...',
+            f'  class {self.gpt_4.__name__}:',
             '    ...',
-            f'  class {cls.gpt_4.__name__}:',
-            '    ...'
         ]
-        
+
         return '\n'.join(output)

g4f/utils.py

@@ -12,37 +12,28 @@ class Utils:
         browser_cookie3.vivaldi,  #  0.32% market share
     ]
 
-    def get_cookies(domain: str, setName: str = None, setBrowser: str = False) -> dict:
+    def get_cookies(self, setName: str = None, setBrowser: str = False) -> dict:
         cookies = {}
-        
-        if setBrowser != False:
-            for browser in Utils.browsers:
-                if browser.__name__ == setBrowser:
-                    try:
-                        for c in browser(domain_name=domain):
-                            if c.name not in cookies:
-                                cookies = cookies | {c.name: c.value} 
-                    
-                    except Exception as e:
-                        pass
-        
-        else:
-            for browser in Utils.browsers:
+
+        for browser in Utils.browsers:
+            if (
+                setBrowser != False
+                and browser.__name__ == setBrowser
+                or setBrowser == False
+            ):
                 try:
-                    for c in browser(domain_name=domain):
+                    for c in browser(domain_name=self):
                         if c.name not in cookies:
                             cookies = cookies | {c.name: c.value} 
-                
+
                 except Exception as e:
                     pass
-        
-        if setName:
-            try:
-                return {setName: cookies[setName]}
-            
-            except ValueError:
-                print(f'Error: could not find {setName} cookie in any browser.')
-                exit(1)
-        
-        else:
+
+        if not setName:
             return cookies
+        try:
+            return {setName: cookies[setName]}
+
+        except ValueError:
+            print(f'Error: could not find {setName} cookie in any browser.')
+            exit(1)

server/auto_proxy.py

@@ -4,20 +4,18 @@ import time
 import threading  
   
   
-def fetch_proxies():  
+def fetch_proxies():
     """Fetch a list of proxy servers from proxyscrape.com.  
   
     Returns:  
         list: A list of proxy servers in the format "IP:Port".  
     """  
-    url = "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all"  
-    response = requests.get(url)  
-    if response.status_code == 200:  
-        proxy_list = response.text.split("\r\n")[:-1]  
-        return proxy_list  
-    else:  
-        print(f"Error fetching proxies: {response.status_code}")  
-        return []  
+    url = "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all"
+    response = requests.get(url)
+    if response.status_code == 200:
+        return response.text.split("\r\n")[:-1]
+    print(f"Error fetching proxies: {response.status_code}")
+    return []  
   
   
 def test_proxy(proxy, prompt, timeout):  

server/backend.py

@@ -79,20 +79,14 @@ def build_messages(jailbreak):
         f'{set_response_language(prompt)}'
     )
 
-    # Fetch search results if internet access is allowed
-    extra = []
-    if internet_access:
-        extra = fetch_search_results(prompt["content"])
-
+    extra = fetch_search_results(prompt["content"]) if internet_access else []
     # Initialize the conversation with the system message
     conversation = [{'role': 'system', 'content': system_message}]
 
     # Add extra results
     conversation += extra
 
-    # Add jailbreak instructions, if any
-    jailbreak_instructions = isJailbreak(jailbreak)
-    if jailbreak_instructions:
+    if jailbreak_instructions := isJailbreak(jailbreak):
         conversation += jailbreak_instructions
 
     # Add the existing conversation and the prompt
@@ -143,8 +137,7 @@ def generate_stream(response, jailbreak):
                 yield "Error: jailbreak failed. Try again."
                 break
     else:
-        for message in response:
-            yield message
+        yield from response
 
 
 def response_jailbroken_success(response: str) -> bool:
@@ -164,13 +157,7 @@ def response_jailbroken_failed(response):
     :param response: Response string  
     :return: Boolean indicating if the response has not been jailbroken  
     """
-    if len(response) < 4:
-        return False
-
-    if not response.startswith("GPT:"):
-        return True
-    else:
-        return False
+    return False if len(response) < 4 else not response.startswith("GPT:")
 
 
 def set_response_language(prompt):

server/website.py

@@ -26,8 +26,8 @@ class Website:
         }
 
     def _chat(self, conversation_id):
-        if not '-' in conversation_id:
-            return redirect(f'/chat')
+        if '-' not in conversation_id:
+            return redirect('/chat')
 
         return render_template('index.html', chat_id=conversation_id)