implement server side bot tracking

Author: iiPythonx

nightwatch/bot/client.py

@@ -36,9 +36,10 @@ 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}>"
+        return f"<User name='{self.name}' hex='{self.hex}' admin={self.admin} bot={self.bot}>"
 
 @dataclass
 class Message:
@@ -149,7 +150,8 @@ async def __authorize(self, username: str, hex: str, address: str) -> tuple[str,
                 f"http{protocol}://{host}:{port}/api/join",
                 json = {
                     "username": username,
-                    "hex": hex
+                    "hex": hex,
+                    "bot": True
                 },
                 timeout = 5
             )

nightwatch/rics/__init__.py

@@ -47,15 +47,17 @@ async def broadcast(payload: dict) -> None:
 class Client:
     def __init__(self, websocket: WebSocket, user_data) -> None:
         self.websocket = websocket
-        self.username, self.hex_code, self.admin = user_data["username"], user_data["hex"], False
+        self.username, self.hex_code = user_data["username"], user_data["hex"]
 
-        self._callback = None
+        # Attributes
+        self.admin, self.bot = False, user_data["bot"]
 
         # Attach to client list
+        self._callback = None
         app.state.clients[self.username] = self
 
     def serialize(self) -> dict[str, str | bool]:
-        return {"name": self.username, "hex": self.hex_code, "admin": self.admin}
+        return {"name": self.username, "hex": self.hex_code, "admin": self.admin, "bot": self.bot}
 
     def cleanup(self) -> None:
         del app.state.clients[self.username]
@@ -100,6 +102,7 @@ async def receive(self) -> typing.Any:
 class ClientJoinModel(BaseModel):
     username: str = Field(..., min_length = 3, max_length = 30)
     hex: str = Field(..., min_length = 6, max_length = 6, pattern = "^[0-9A-Fa-f]{6}$")
+    bot: bool = False
 
 @app.post("/api/join")
 async def route_index(client: ClientJoinModel) -> JSONResponse: