Genesis commit

libminecraft.py

@@ -0,0 +1,111 @@
+import asyncio
+import atexit
+from re import compile
+from typing import Any, Callable
+
+from aiohttp import ClientSession
+
+
+class Minecraft:
+    def __init__(self, host: str, port: int, password: str):
+        self.host = host.rstrip("/")
+        self.port = port
+        self.headers = {"Authorization": password}
+        self.session = ClientSession()
+        atexit.register(self.exit)
+        self._hooks: list[Callable] = []
+
+    def exit(self):
+        asyncio.run(self.session.close())
+
+    async def _get(self, method: str, **json: Any):
+        async with self.session.get(
+            f"{self.host}:{self.port}{method}", json=json, headers=self.headers
+        ) as response:
+            if response.status == 200:
+                return await response.json()
+            raise ConnectionError(
+                f"Error while sending {method} to Server: HTTP code {response.status}."
+            )
+
+    async def _stream(self, method: str, **json: Any):
+        async with self.session.get(
+            f"{self.host}:{self.port}{method}", json=json, headers=self.headers
+        ) as response:
+            if response.status == 200:
+                async for chunk, _ in response.content.iter_chunks():
+                    yield chunk
+            else:
+                raise ConnectionError(
+                    f"Error while sending {method} to Server: HTTP code {response.status}."
+                )
+
+    async def start(self):
+        return await self._get("/start")
+
+    async def status(self):
+        return await self._get("/status")
+
+    async def stop(self, countdown: int = 60, reason: str = "", timeout: int = 10):
+        return await self._get(
+            "/stop", countdown=countdown, reason=reason, timeout=timeout
+        )
+
+    async def restart(self, countdown: int = 60, reason: str = "", timeout: int = 10):
+        return await self._get(
+            "/restart", countdown=countdown, reason=reason, timeout=timeout
+        )
+
+    async def maintainance(
+        self, countdown: int = 60, reason: str = "", timeout: int = 10
+    ):
+        return await self._get(
+            "/maintainance", countdown=countdown, reason=reason, timeout=timeout
+        )
+
+    async def command(self, command: str):
+        return await self._get("/command", command=command)
+
+    async def logs_stream(self):
+        async for chunk in self._stream("/logs/stream"):
+            yield chunk.decode().strip()
+
+    async def logs_tail(self, back: int = 10):
+        async for chunk in self._stream("/logs/tail", back=back):
+            yield chunk.decode().strip()
+
+    def onConsoleLog(self, pattern: str | None = None) -> Callable:
+        def wrapper(func: Callable):
+            compiled = compile(pattern) if pattern else None
+
+            async def callback(line: str):
+                if compiled is None:
+                    func(line)
+                    return
+                match = compiled.fullmatch(line)
+                if match is None:
+                    return
+                groups = match.groups()
+                if groups:
+                    func(*groups)
+                else:
+                    func(line)
+
+            self._hooks.append(callback)
+
+            return func
+
+        return wrapper
+
+    def onPlayerChat(self) -> Callable:
+        def wrapper(func: Callable):
+            return self.onConsoleLog(
+                r"\[[0-9]{2}:[0-9]{2}:[0-9]{2}\] \[Server thread\/INFO\]: <(.*)> (.*)"
+            )(func)
+
+        return wrapper
+
+    async def loop(self):
+        async for line in self.logs_stream():
+            for callback in self._hooks:
+                await callback(line)