Compare commits

..

7 Commits
v0.1 ... main

Author SHA1 Message Date
trueold89 b3f9e7c8d9
fixed bug of getting motd 2024-07-13 20:54:52 +03:00
trueold89 c0b26f1fe0
BugFix: player list now updated correctly 2024-07-13 05:40:13 +03:00
trueold89 2ae08b8f7e
Add docstrings 2024-07-08 00:09:08 +03:00
trueold89 62e8aad3a8
Add all_info property 2024-07-07 23:04:49 +03:00
trueold89 c56bc396fe
Add players_list property 2024-07-07 21:59:59 +03:00
trueold89 4d9b2586bb
Add _data param 2024-07-07 21:21:17 +03:00
trueold89 be81204cd9
Add README.md 2024-07-07 15:55:40 +03:00
4 changed files with 126 additions and 21 deletions

View File

@ -13,7 +13,7 @@ pip install --extra-index-url https://git.orudo.ru/api/packages/trueold89/pypi/s
**From PyPi**
```bash
pip install trueold89/mcaio
pip install mcaio
```
## Build:
@ -68,6 +68,8 @@ print(name)
| motd | Server motd |
| players_count | Current number of players on the server |
| maxplayers | Max number of players on the server |
| players_list | List of current players on server |
| all_info | Dict with all information about server |
### As cli:
@ -83,3 +85,5 @@ MC_HOST=localhost MC_PORT=25565 mcaio name
| motd | Server motd |
| pcount | Current number of players on the server |
| pmax | Max number of players on the server |
| players | List of current players on server |
| all | Dict with all information about server |

View File

@ -33,19 +33,31 @@ async def get_max(server: AIOMCServer) -> int:
return await server.players_count
async def get_players(server: AIOMCServer) -> tuple:
return tuple(await server.players_list)
async def get_all(server: AIOMCServer) -> dict:
return await server.all_info
async def action() -> None:
try:
HOST, PORT = get_env("MC_HOST"), int(get_env("MC_PORT"))
server = AIOMCServer(HOST, PORT)
match arg():
case 'name':
case "name":
out = get_name(server)
case 'pmax':
case "pmax":
out = get_max(server)
case 'pcount':
case "pcount":
out = get_count(server)
case 'motd':
case "motd":
out = get_motd(server)
case "players":
out = get_players(server)
case 'all':
out = get_all(server)
case _:
raise RuntimeError
print(await out)

View File

@ -1,5 +1,6 @@
from asyncio import open_connection as aiocon, StreamReader, StreamWriter
from abc import ABC, abstractmethod
from collections.abc import Iterable
from struct import pack as struct_pack
from json import loads as jl
@ -36,36 +37,89 @@ class AIOConnection(object):
class IMCServer(ABC):
def __init__(self, host: str, port: int) -> None:
"""
:param host: Minecraft server ip or hostname
:param port: Minecraft server port
"""
self.host = host
self.port = port
@property
@abstractmethod
async def players_count(self) -> int:
async def players_count(self) -> int | None:
"""
Returns current number of players on server
:return: Number of players on server
"""
raise NotImplementedError
@property
@abstractmethod
async def name(self) -> str:
async def name(self) -> str | None:
"""
Returns server core name
:return: Server name
"""
raise NotImplementedError
@property
@abstractmethod
async def maxplayers(self) -> int:
async def maxplayers(self) -> int | None:
"""
Returns max number of players available on server
:return: Max number of players
"""
raise NotImplementedError
@property
@abstractmethod
async def motd(self) -> str:
async def motd(self) -> str | None:
"""
Returns Message of the day
:return: Server motd
"""
raise NotImplementedError
@property
@abstractmethod
async def players_list(self) -> Iterable:
"""
Returns iterable object with online players nicknames
:return: Players nicknames iterable object
"""
raise NotImplementedError
@property
@abstractmethod
async def all_info(self) -> dict:
"""
Returns dict with all information about server
:return: All information about server
"""
raise NotImplementedError
class AIOMCServer(IMCServer):
_name: str
_max: int
_count: int
_motd: str
_name: str | None
_max: int | None
_count: int | None
_motd: str | None
_data: bytes
_players: Iterable
def __init__(self, host: str, port: int) -> None:
super().__init__(host, port)
self._data = self._pack_data(
b"\x00\x00" + self._pack_data(self.host.encode('utf8')) + self.
_pack_port(self.port) + b"\x01")
self.clean()
@staticmethod
async def _unpack_varint(s):
@ -94,15 +148,20 @@ class AIOMCServer(IMCServer):
d = bytes(d, "utf-8")
return h + d
def clean(self) -> None:
self._name = None
self._motd = None
self._max = None
self._count = None
self._players = ()
@staticmethod
def _pack_port(i):
return struct_pack('>H', i)
async def _get_data(self) -> dict:
async with AIOConnection(self.host, self.port) as socket:
await socket.send(self._pack_data(
b"\x00\x00" + self._pack_data(self.host.encode('utf8')) + self.
_pack_port(self.port) + b"\x01"))
await socket.send(self._data)
await socket.send(self._pack_data("\x00"))
await self._unpack_varint(socket)
await self._unpack_varint(socket)
@ -114,28 +173,52 @@ class AIOMCServer(IMCServer):
async def update(self) -> None:
data = await self._get_data()
self.clean()
self._name = data["version"]["name"]
self._motd = data["description"]
if isinstance(self._motd, dict):
self._motd = self._motd["text"]
players = data["players"]
self._count = int(players["online"])
self._max = int(players["max"])
if "sample" in tuple(players.keys()):
self._players = tuple(map(lambda player: player["name"],
players["sample"]))
@property
async def players_count(self) -> int:
async def players_count(self) -> int | None | None:
await self.update()
return self._count
@property
async def name(self) -> str:
async def name(self) -> str | None | None:
await self.update()
return self._name
@property
async def maxplayers(self) -> int:
async def maxplayers(self) -> int | None | None:
await self.update()
return self._max
@property
async def motd(self) -> str:
async def motd(self) -> str | None | None:
await self.update()
return self._motd
@property
async def players_list(self) -> Iterable:
await self.update()
return self._players
@property
async def all_info(self) -> dict:
await self.update()
return {
"name": self._name,
"motd": self._motd,
"players": {
"max": self._max,
"online": self._count,
"list": self._players
}
}

View File

@ -1,12 +1,18 @@
from setuptools import setup
from pathlib import Path
this_directory = Path(__file__).parent
long_description = (this_directory / "README.md").read_text()
setup(
name="mcaio",
version="0.1",
version="0.2.1",
url="https://git.orudo.ru/trueold89/mcaio",
author="trueold89",
author_email="trueold89@orudo.ru",
description="Asynс lib to get information about Minecraft server",
long_description=long_description,
long_description_content_type='text/markdown',
packages=["mcaio"],
entry_points={
"console_scripts": ["mcaio = mcaio.cli:main"]