From dcc4428efe1b4c1e6c3c19eb1bdb3d522761768c Mon Sep 17 00:00:00 2001 From: iiPython Date: Mon, 25 Nov 2024 18:25:26 -0600 Subject: [PATCH] implement server side bot tracking --- nightwatch/bot/client.py | 6 ++++-- nightwatch/rics/__init__.py | 9 ++++++--- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/nightwatch/bot/client.py b/nightwatch/bot/client.py index 26d4eb2..a0aad53 100644 --- a/nightwatch/bot/client.py +++ b/nightwatch/bot/client.py @@ -36,9 +36,10 @@ class User: name: str hex: str admin: bool + bot: bool def __repr__(self) -> str: - return f"" + return f"" @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 ) diff --git a/nightwatch/rics/__init__.py b/nightwatch/rics/__init__.py index e9cd0a1..98f046c 100644 --- a/nightwatch/rics/__init__.py +++ b/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: