72 lines
2 KiB
Python
72 lines
2 KiB
Python
import asyncio
|
|
|
|
from logging import Logger
|
|
from json import loads
|
|
from typing import Tuple
|
|
|
|
logger = Logger("reportmonster-client")
|
|
|
|
|
|
class ReportMonsterResponse:
|
|
def __init__(self, raw):
|
|
self._raw = raw
|
|
|
|
@property
|
|
def status(self) -> Tuple[int, str]:
|
|
return int(self._raw[:2]), self._raw.split(":")[0][3:]
|
|
|
|
@property
|
|
def content(self) -> str | dict | list:
|
|
text = ":".join(self._raw.split(":")[1:]).strip()
|
|
|
|
if self.status[0] == 20:
|
|
return loads(text)
|
|
|
|
return text
|
|
|
|
@property
|
|
def error(self) -> bool:
|
|
return self.status[0] >= 90
|
|
|
|
|
|
class ReportMonsterClient:
|
|
def __init__(self, username, password, host="127.0.0.1", port=6789):
|
|
self.username = username
|
|
self.password = password
|
|
self.host = host
|
|
self.port = port
|
|
|
|
self._socket_reader = None
|
|
self._socket_writer = None
|
|
|
|
async def connect(self):
|
|
self._socket_reader, self._socket_writer = await asyncio.open_connection(self.host, self.port)
|
|
identification = await self.read(wrap=False)
|
|
return identification
|
|
|
|
async def auth(self):
|
|
response = await self.write(f"login {self.username} {self.password}")
|
|
return response
|
|
|
|
async def read(self, wrap=True):
|
|
response = ""
|
|
while not response.endswith("\n> "):
|
|
data = await self._socket_reader.read(1)
|
|
if not (plain := data.decode()):
|
|
break
|
|
response += plain
|
|
stripped = response.strip().strip(">").strip()
|
|
return ReportMonsterResponse(stripped) if wrap else stripped
|
|
|
|
async def write(self, message, read=True, wrap=True):
|
|
self._socket_writer.write(message.encode() + b"\n")
|
|
await self._socket_writer.drain()
|
|
return await self.read(wrap=wrap) if read else None
|
|
|
|
async def disconnect(self):
|
|
if self._socket_writer:
|
|
await self._socket_writer.close()
|
|
self._socket_writer = None
|
|
self._socket_reader = None
|
|
|