feat!: discord replaced by matrix

.gitignore

@@ -4,4 +4,4 @@ venv/
 __pycache__/
 *.py[cod]
 
-config.py
+config.json

Dockerfile

@@ -6,6 +6,6 @@ COPY requirements.txt /app
 RUN pip install -r requirements.txt
 
 COPY stockbot-buyvm.py /app
-COPY config.py /app
+COPY config.json /app
 
 CMD ["python", "stockbot-buyvm.py"]

README.md

@@ -2,9 +2,17 @@
 Send alerts when [BuyVM](https://buyvm.net) has KVM slices in stock.
 
 ## Usage
-1. Create a Discord Webhook and add it to `config.py`:
+1. Create a JSON configuration file in `config.json`:
-```python
+```json
-DISCORD_WEBHOOK = '<discord webhook url>'
+{
+  "memory": [512, 1, 2, 4],
+  "matrix": {
+    "homeserver": "https://matrix.juggalol.com",
+    "username": "",
+    "password": "",
+    "room_id": ""
+  }
+}
 ```
 
 2. Build Docker container:

matrix.py

@@ -0,0 +1,89 @@
+import markdown
+from loguru import logger
+from nio import AsyncClient, LoginResponse
+
+
+class MatrixBot:
+    def __init__(self, config: dict):
+        self.config = config
+
+        self.client = AsyncClient(
+            homeserver=self.config['homeserver'],
+            user=self.config['username']
+        )
+        self.logged_in = False
+
+    async def ensure_logged_in(self):
+        if not self.logged_in:
+            try:
+                response = await self.client.login(password=self.config['password'])
+                if isinstance(response, LoginResponse):
+                    self.logged_in = True
+                    logger.info(f"Logged in as {self.config['username']}")
+                else:
+                    logger.error(f"Failed to login as {self.config['username']}: {response}")
+                    logger.error("Closing nio session")
+                    await self.client.close()
+            except Exception as e:
+                logger.error(f"Exception during login: {e}")
+                await self.client.close()
+                raise
+
+    async def send_message(self, message: str):
+        await self.ensure_logged_in()
+
+        if not self.logged_in:
+            logger.error("Unable to send message, login failed")
+            return
+
+        try:
+            await self.client.room_send(
+                room_id=self.config['room_id'],
+                message_type="m.room.message",
+                content={
+                    "msgtype": "m.text",
+                    "body": message
+                }
+            )
+            logger.info("Message sent")
+        except Exception as e:
+            logger.error(f"Exception during sending message: {e}")
+            raise
+
+    async def send_markdown(self, message: str):
+        await self.ensure_logged_in()
+
+        if not self.logged_in:
+            logger.error("Unable to send message, login failed")
+            return
+
+        try:
+            # Convert message to markdown
+            html = markdown.markdown(message)
+
+            # Send markdown formatted message
+            await self.client.room_send(
+                room_id=self.config['room_id'],
+                message_type="m.room.message",
+                content={
+                    "msgtype": "m.text",
+                    "body": message,
+                    "format": "org.matrix.custom.html",
+                    "formatted_body": html
+                }
+            )
+            logger.info("Markdown message sent")
+        except Exception as e:
+            logger.error(f"Exception during sending markdown message: {e}")
+            raise
+
+    async def close(self):
+        if self.logged_in:
+            try:
+                await self.client.logout()
+                self.logged_in = False
+                logger.info(f"Logged out from {self.config['homeserver']}")
+            except Exception as e:
+                logger.error(f"Exception during logout: {e}")
+            finally:
+                await self.client.close()  # Ensure the client is closed

requirements.txt

@@ -1,3 +1,5 @@
 beautifulsoup4
 requests
 loguru
+matrix-nio
+markdown

stockbot-buyvm.py

@@ -1,9 +1,10 @@
 """buyvm stock checker"""
+import json
+import asyncio
 import requests
 from bs4 import BeautifulSoup
 from loguru import logger
-
+from matrix import MatrixBot
-from config import DISCORD_WEBHOOK
 
 BASE_URL = 'https://my.frantech.ca/'
 URLS = [
@@ -14,13 +15,6 @@ URLS = [
 ]
 
 
-def send_notification(payload):
-    try:
-        requests.post(DISCORD_WEBHOOK, json=payload)
-    except requests.RequestException as e:
-        logger.error(f'error sending notification: {str(e)}')
-
-
 def get_url(url):
     try:
         response = requests.get(url)
@@ -58,8 +52,17 @@ def get_packages(html):
     return packages
 
 
-def main():
+def load_config(filename):
+    with open(filename) as f:
+        return json.load(f)
+
+
+async def main():
     logger.info('checking buyvm stocks')
+    config = load_config('config.json')
+    bot = MatrixBot(config['matrix'])
+    memory_filter = config.get('memory', [512, 1, 2, 4])  # Defaults to price <= $15.00
+
     for url in URLS:
         html = get_url(url)
 
@@ -68,23 +71,31 @@ def main():
 
         packages = get_packages(html)
         for package in packages:
-            if package['qty'] > 0:
+            qty = package['qty']
+            memory = int(package['name'].split()[-1][:-2])
+
+            if qty > 0 and (memory in memory_filter):
                 logger.info(f"{package['name']}: {package['qty']} in stock")
-                send_notification({
+                await bot.send_message(f"🚨 {package['name']}: {package['qty']} in stock 🚨\n{package['url']}")
-                    "username": "stockbot-buyvm",
+
-                    "embeds": [
+    await bot.close()
-                        {
+
-                            "author": {
+
-                                "name": "BuyVM",
+def main_with_shutdown():
-                            },
+    loop = asyncio.get_event_loop()
-                            "title": package['name'],
+    main_task = loop.create_task(main())
-                            "url": package['url'],
+
-                            "description": f"{package['qty']} in stock now!"
+    try:
-                        }
+        loop.run_until_complete(main_task)
-                    ],
+    except asyncio.CancelledError:
-                    "content": "STOCK ALERT"
+        logger.info("Main task has been cancelled.")
-                })
+    finally:
+        pending_tasks = [t for t in asyncio.all_tasks(loop) if not t.done()]
+        if pending_tasks:
+            loop.run_until_complete(asyncio.gather(*pending_tasks, return_exceptions=True))
+        loop.run_until_complete(loop.shutdown_asyncgens())
+        loop.close()
 
 
 if __name__ == '__main__':
-    main()
+    main_with_shutdown()