Refactor

src/revChatGPT/V1.py

@@ -78,7 +78,7 @@ def random_int(min: int, max: int) -> int:
 log = logging.getLogger(__name__)
 
 
-def logger(is_timed: bool):
+def logger(is_timed: bool) -> function:
     """Logger decorator
 
     Args:
@@ -88,7 +88,7 @@ def logger(is_timed: bool):
         _type_: decorated function
     """
 
-    def decorator(func):
+    def decorator(func: function) -> function:
         wraps(func)
 
         def wrapper(*args, **kwargs):
@@ -146,9 +146,7 @@ def captcha_solver(images: list[str], challenge_details: dict) -> int:
             "Developer instructions: The captcha images have an index starting from 0 from left to right",
         )
         print("Enter the index of the images that matches the captcha instructions:")
-        index = int(input())
-
-        return index
+        return int(input())
 
 
 CAPTCHA_URL = getenv("CAPTCHA_URL", "https://bypass.churchless.tech/captcha/")
@@ -198,41 +196,39 @@ def get_arkose_token(
         if resp.status_code != 200:
             raise Exception("Failed to verify captcha")
         return resp_json.get("token")
-    else:
-        working_endpoints: list[str] = []
-        # Check uptime for different endpoints via gatus
-        resp2: list[dict] = requests.get(
-            "https://stats.churchless.tech/api/v1/endpoints/statuses?page=1"
-        ).json()
-        for endpoint in resp2:
-            # print(endpoint.get("name"))
-            if endpoint.get("group") != "Arkose Labs":
-                continue
-            # Check the last 5 results
-            results: list[dict] = endpoint.get("results", [])[-5:-1]
-            # print(results)
-            if not results:
-                print(f"Endpoint {endpoint.get('name')} has no results")
-                continue
-            # Check if all the results are up
-            if all(result.get("success") == True for result in results):
-                working_endpoints.append(endpoint.get("name"))
-        if not working_endpoints:
-            print("No working endpoints found. Please solve the captcha manually.")
-            return get_arkose_token(download_images=True, captcha_supported=True)
-        # Choose a random endpoint
-        endpoint = random.choice(working_endpoints)
-        resp: requests.Response = requests.get(endpoint)
-        if resp.status_code != 200:
-            if resp.status_code != 511:
-                raise Exception("Failed to get captcha token")
-            else:
-                print("Captcha required. Please solve the captcha manually.")
-                return get_arkose_token(download_images=True, captcha_supported=True)
-        try:
-            return resp.json().get("token")
-        except Exception:
-            return resp.text
+    working_endpoints: list[str] = []
+    # Check uptime for different endpoints via gatus
+    resp2: list[dict] = requests.get(
+        "https://stats.churchless.tech/api/v1/endpoints/statuses?page=1",
+    ).json()
+    for endpoint in resp2:
+        # print(endpoint.get("name"))
+        if endpoint.get("group") != "Arkose Labs":
+            continue
+        # Check the last 5 results
+        results: list[dict] = endpoint.get("results", [])[-5:-1]
+        # print(results)
+        if not results:
+            print(f"Endpoint {endpoint.get('name')} has no results")
+            continue
+        # Check if all the results are up
+        if all(result.get("success") is True for result in results):
+            working_endpoints.append(endpoint.get("name"))
+    if not working_endpoints:
+        print("No working endpoints found. Please solve the captcha manually.")
+        return get_arkose_token(download_images=True, captcha_supported=True)
+    # Choose a random endpoint
+    endpoint = random.choice(working_endpoints)
+    resp: requests.Response = requests.get(endpoint)
+    if resp.status_code != 200:
+        if resp.status_code != 511:
+            raise Exception("Failed to get captcha token")
+        print("Captcha required. Please solve the captcha manually.")
+        return get_arkose_token(download_images=True, captcha_supported=True)
+    try:
+        return resp.json().get("token")
+    except Exception:
+        return resp.text
 
 
 class Chatbot:
@@ -315,8 +311,8 @@ def __init__(
             else:
                 self.session.proxies.update(proxies)
 
-        self.conversation_id = conversation_id or config.get("conversation_id", None)
-        self.parent_id = parent_id or config.get("parent_id", None)
+        self.conversation_id = conversation_id or config.get("conversation_id")
+        self.parent_id = parent_id or config.get("parent_id")
         self.conversation_mapping = {}
         self.conversation_id_prev_queue = []
         self.parent_id_prev_queue = []
@@ -482,7 +478,7 @@ def __write_cache(self, info: dict) -> None:
         json.dump(info, open(self.cache_path, "w", encoding="utf-8"), indent=4)
 
     @logger(is_timed=False)
-    def __read_cache(self):
+    def __read_cache(self) -> dict[str, dict[str, str]]:
         try:
             cached = json.load(open(self.cache_path, encoding="utf-8"))
         except (FileNotFoundError, json.decoder.JSONDecodeError):
@@ -529,7 +525,7 @@ def __send_request(
                 # print(f"Arkose token obtained: {data['arkose_token']}")
             except Exception as e:
                 print(e)
-                raise e
+                raise
 
         cid, pid = data["conversation_id"], data["parent_message_id"]
         message = ""
@@ -584,10 +580,12 @@ def __send_request(
             author = {}
             if line.get("message"):
                 author = metadata.get("author", {}) or line["message"].get("author", {})
-                if line["message"].get("content"):
-                    if line["message"]["content"].get("parts"):
-                        if len(line["message"]["content"]["parts"]) > 0:
-                            message_exists = True
+                if (
+                    line["message"].get("content")
+                    and line["message"]["content"].get("parts")
+                    and len(line["message"]["content"]["parts"]) > 0
+                ):
+                    message_exists = True
             message: str = (
                 line["message"]["content"]["parts"][0] if message_exists else ""
             )
@@ -629,7 +627,7 @@ def post_messages(
         messages: list[dict],
         conversation_id: str | None = None,
         parent_id: str | None = None,
-        plugin_ids: list = [],
+        plugin_ids: list = None,
         model: str | None = None,
         auto_continue: bool = False,
         timeout: float = 360,
@@ -656,6 +654,8 @@ def post_messages(
                 "citations": list[dict],
             }
         """
+        if plugin_ids is None:
+            plugin_ids = []
         if parent_id and not conversation_id:
             raise t.Error(
                 source="User",
@@ -720,7 +720,7 @@ def ask(
         conversation_id: str | None = None,
         parent_id: str = "",
         model: str = "",
-        plugin_ids: list = [],
+        plugin_ids: list = None,
         auto_continue: bool = False,
         timeout: float = 360,
         **kwargs,
@@ -745,6 +745,8 @@ def ask(
                 "recipient": str,
             }
         """
+        if plugin_ids is None:
+            plugin_ids = []
         messages = [
             {
                 "id": str(uuid.uuid4()),
@@ -1034,7 +1036,12 @@ def rollback_conversation(self, num: int = 1) -> None:
             self.parent_id = self.parent_id_prev_queue.pop()
 
     @logger(is_timed=True)
-    def get_plugins(self, offset: int = 0, limit: int = 250, status: str = "approved"):
+    def get_plugins(
+        self,
+        offset: int = 0,
+        limit: int = 250,
+        status: str = "approved",
+    ) -> dict[str, str]:
         """
         Get plugins
         :param offset: Integer. Offset (Only supports 0)
@@ -1048,7 +1055,7 @@ def get_plugins(self, offset: int = 0, limit: int = 250, status: str = "approved
         return json.loads(response.text)
 
     @logger(is_timed=True)
-    def install_plugin(self, plugin_id: str):
+    def install_plugin(self, plugin_id: str) -> None:
         """
         Install plugin by ID
         :param plugin_id: String. ID of plugin
@@ -1156,10 +1163,12 @@ async def __send_request(
                         "author",
                         {},
                     )
-                    if line["message"].get("content"):
-                        if line["message"]["content"].get("parts"):
-                            if len(line["message"]["content"]["parts"]) > 0:
-                                message_exists = True
+                    if (
+                        line["message"].get("content")
+                        and line["message"]["content"].get("parts")
+                        and len(line["message"]["content"]["parts"]) > 0
+                    ):
+                        message_exists = True
                 message: str = (
                     line["message"]["content"]["parts"][0] if message_exists else ""
                 )
@@ -1200,7 +1209,7 @@ async def post_messages(
         messages: list[dict],
         conversation_id: str | None = None,
         parent_id: str | None = None,
-        plugin_ids: list = [],
+        plugin_ids: list = None,
         model: str | None = None,
         auto_continue: bool = False,
         timeout: float = 360,
@@ -1229,6 +1238,8 @@ async def post_messages(
                 "citations": list[dict],
             }
         """
+        if plugin_ids is None:
+            plugin_ids = []
         if parent_id and not conversation_id:
             raise t.Error(
                 source="User",
@@ -1292,7 +1303,7 @@ async def ask(
         conversation_id: str | None = None,
         parent_id: str = "",
         model: str = "",
-        plugin_ids: list = [],
+        plugin_ids: list = None,
         auto_continue: bool = False,
         timeout: int = 360,
         **kwargs,
@@ -1320,6 +1331,8 @@ async def ask(
             }
         """
 
+        if plugin_ids is None:
+            plugin_ids = []
         messages = [
             {
                 "id": str(uuid.uuid4()),
@@ -1649,7 +1662,7 @@ def handle_commands(command: str) -> bool:
         elif command == "!exit":
             if isinstance(chatbot.session, httpx.AsyncClient):
                 chatbot.session.aclose()
-            exit()
+            sys.exit()
         else:
             return False
         return True
@@ -1703,7 +1716,7 @@ def handle_commands(command: str) -> bool:
                 print()
 
     except (KeyboardInterrupt, EOFError):
-        exit()
+        sys.exit()
     except Exception as exc:
         error = t.CLIError("command line program unknown error")
         raise error from exc

src/revChatGPT/__init__.py

@@ -18,21 +18,20 @@ def verify() -> None:
     if int(__import__("platform").python_version_tuple()[0]) < 3:
         error = t.NotAllowRunning("Not available Python version")
         raise error
-    elif int(__import__("platform").python_version_tuple()[1]) < 9:
+    if int(__import__("platform").python_version_tuple()[1]) < 9:
         error = t.NotAllowRunning(
             f"Not available Python version: {__import__('platform').python_version()}",
         )
         raise error
-    else:
-        if (
-            int(__import__("platform").python_version_tuple()[1]) < 10
-            and int(__import__("platform").python_version_tuple()[0]) == 3
-        ):
-            __import__("warnings").warn(
-                UserWarning(
-                    "The current Python is not a recommended version, 3.10+ is recommended",
-                ),
-            )
+    if (
+        int(__import__("platform").python_version_tuple()[1]) < 10
+        and int(__import__("platform").python_version_tuple()[0]) == 3
+    ):
+        __import__("warnings").warn(
+            UserWarning(
+                "The current Python is not a recommended version, 3.10+ is recommended",
+            ),
+        )
 
 
 verify()

src/revChatGPT/__main__.py

@@ -12,7 +12,7 @@
 __all__ = ()
 
 
-def main():
+def main() -> None:
     """
     main function for CLI
     """

src/revChatGPT/typings.py

@@ -1,13 +1,14 @@
 """
 A module that contains all the types used in this project
 """
+
 import os
 import platform
 from enum import Enum
 from typing import Union
 
 
-python_version = [each for each in platform.python_version_tuple()]
+python_version = list(platform.python_version_tuple())
 SUPPORT_ADD_NOTES = int(python_version[0]) >= 3 and int(python_version[1]) >= 11
 
 

tests/test_recipient.py

@@ -7,7 +7,7 @@
 config = json.load(open("/home/acheong/.config/revChatGPT/config.json"))
 
 
-async def main():
+async def main() -> None:
     chatbot = AsyncChatbot(config)
     async for message in chatbot.ask("Hello, how are you?"):
         print(message.get("message"))