60 lines
1.5 KiB
Python
60 lines
1.5 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 == 20:
|
|
return loads(text)
|
|
|
|
return text
|
|
|
|
@property
|
|
def error(self) -> bool:
|
|
return self.status >= 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()
|
|
return identification
|
|
|
|
async def read(self):
|
|
content = await self._socket_reader.readuntil(b"> ")
|
|
return content.decode().rstrip().rstrip(">").rstrip()
|
|
|
|
async def write(self, message):
|
|
await self._socket_writer.write(message.encode())
|
|
|
|
async def disconnect(self):
|
|
if self._socket_writer:
|
|
await self._socket_writer.close()
|
|
self._socket_writer = None
|
|
self._socket_reader = None
|
|
|