clean up client

nightwatch/bot/client.py

@@ -3,61 +3,13 @@
 # Modules
 import typing
 import asyncio
-from dataclasses import dataclass, fields, is_dataclass
 
 import orjson
 import requests
 from websockets import connect
 from websockets.asyncio.client import ClientConnection
 
-# Typing
-T = typing.TypeVar("T")
-
-def from_dict(cls: typing.Type[T], data: dict) -> T:
-    if not is_dataclass(cls):
-        raise ValueError(f"{cls} is not a dataclass")
-
-    field_types = {f.name: f.type for f in fields(cls)}
-    instance_data = {}
-
-    for key, value in data.items():
-        if key in field_types:
-            field_type = field_types[key]
-            if is_dataclass(field_type) and isinstance(value, dict):
-                instance_data[key] = from_dict(field_type, value)  # type: ignore
-
-            else:
-                instance_data[key] = value
-
-    return cls(**instance_data)
-
-@dataclass
-class User:
-    name: str
-    hex: str
-    admin: bool
-    bot: bool
-
-    def __repr__(self) -> str:
-        return f"<User name='{self.name}' hex='{self.hex}' admin={self.admin} bot={self.bot}>"
-
-@dataclass
-class Message:
-    user: User
-    message: str
-    time: int
-
-    def __repr__(self) -> str:
-        return f"<Message user='{self.user}' message='{self.message}' time={self.time}>"
-
-@dataclass
-class RicsInfo:
-    name: str
-    users: list[User]
-    chat_logs: list[Message]
-
-    def __repr__(self) -> str:
-        return f"<RicsInfo name='{self.name}' users=[...] chat_logs=[...]>"
+from .types import from_dict, User, Message, RicsInfo
 
 # Exceptions
 class AuthorizationFailed(Exception):
@@ -66,20 +18,11 @@ class AuthorizationFailed(Exception):
 # Handle state
 class ClientState:
     def __init__(self) -> None:
-        self.__state = {}
-
-        # Typing
         self.user_list: list[User]
         self.chat_logs: list[Message]
         self.rics_info: dict[str, str]
         self.socket   : ClientConnection
 
-    def __getitem__(self, key: str) -> typing.Any:
-        return self.__state.get(key)
-
-    def __setitem__(self, key: str, value: typing.Any) -> None:
-        self.__state[key] = value
-
 class Context:
     def __init__(
         self,
@@ -116,20 +59,16 @@ def __init__(self) -> None:
         self.__session = requests.Session()
 
     # Events (for overwriting)
-    async def on_connect(self, ctx) -> None:
-        print(ctx)
-
-    async def on_message(self, ctx: Context) -> None:
-        if ctx.message.message == "fuck you":
-            await ctx.reply("FUCK YOU1!!!!!!!!!!!")
+    async def on_connect(self, _) -> None:
+        pass
 
-        if ctx.message.message == "what":
-            await ctx.send("test message")
+    async def on_message(self, _) -> None:
+        pass
 
-    async def on_join(self, ctx) -> None:
+    async def on_join(self, _) -> None:
         pass
 
-    async def on_leave(self, ctx) -> None:
+    async def on_leave(self, _) -> None:
         pass
 
     # Handle running
@@ -148,11 +87,7 @@ async def __authorize(self, username: str, hex: str, address: str) -> tuple[str,
         try:
             response = self.__session.post(
                 f"http{protocol}://{host}:{port}/api/join",
-                json = {
-                    "username": username,
-                    "hex": hex,
-                    "bot": True
-                },
+                json = {"username": username, "hex": hex, "bot": True},
                 timeout = 5
             )
             response.raise_for_status()

nightwatch/bot/types.py

@@ -1,33 +1,54 @@
 # Copyright (c) 2024 iiPython
 
-from datetime import datetime
+# Modules
+import typing
+from dataclasses import dataclass, fields, is_dataclass
 
+# Typing
+T = typing.TypeVar("T")
+
+def from_dict(cls: typing.Type[T], data: dict) -> T:
+    if not is_dataclass(cls):
+        raise ValueError(f"{cls} is not a dataclass")
+
+    field_types = {f.name: f.type for f in fields(cls)}
+    instance_data = {}
+
+    for key, value in data.items():
+        if key in field_types:
+            field_type = field_types[key]
+            if is_dataclass(field_type) and isinstance(value, dict):
+                instance_data[key] = from_dict(field_type, value)  # type: ignore
+
+            else:
+                instance_data[key] = value
+
+    return cls(**instance_data)
+
+@dataclass
 class User:
-    def __init__(self, name: str, hex: str, admin: bool) -> None:
-        self.name, self.hex, self.admin = name, hex, admin
+    name: str
+    hex: str
+    admin: bool
+    bot: bool
 
     def __repr__(self) -> str:
-        return f"<User name='{self.name}' hex='{self.hex}' admin={self.admin}>"
-
-    @staticmethod
-    def from_payload(payload: dict):
-        return User(**payload)
+        return f"<User name='{self.name}' hex='{self.hex}' admin={self.admin} bot={self.bot}>"
 
+@dataclass
 class Message:
-    def __init__(self, user: dict, message: str, time: int | datetime) -> None:
-        self.user, self.text = User.from_payload(user), message
-        self.time = time if isinstance(time, datetime) else datetime.fromtimestamp(time)
+    user: User
+    message: str
+    time: int
 
     def __repr__(self) -> str:
-        return f"<Message user={self.user} text='{self.text}'>"
-
-    @staticmethod
-    def from_payload(payload: dict):
-        return Message(**payload)
+        return f"<Message user='{self.user}' message='{self.message}' time={self.time}>"
 
-class Server:
-    def __init__(self, address: str, name: str, users: list[User]) -> None:
-        self.address, self.name, self.users = address, name, users
+@dataclass
+class RicsInfo:
+    name: str
+    users: list[User]
+    chat_logs: list[Message]
 
     def __repr__(self) -> str:
-        return f"<Server address='{self.address}' name='{self.name}' users=[{','.join(str(u) for u in self.users)}]"
+        return f"<RicsInfo name='{self.name}' users=[...] chat_logs=[...]>"