Compare commits

...

3 Commits

Author SHA1 Message Date
269ea6db08 docs: add docstrings 2024-09-15 18:55:32 -04:00
3c9795ac4f docs: add docstrings for MatrixBot 2024-09-15 18:51:34 -04:00
2e5196c6b8 feat!: discord replaced by matrix 2024-09-15 18:31:30 -04:00
7 changed files with 273 additions and 31 deletions

2
.gitignore vendored
View File

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

6
.pylintrc Normal file
View File

@ -0,0 +1,6 @@
[MASTER]
max-line-length=120
init-hook='import sys; sys.path.append("src")'
[MESSAGES CONTROL]
disable=R0903

View File

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

View File

@ -2,9 +2,17 @@
Send alerts when [BuyVM](https://buyvm.net) has KVM slices in stock. Send alerts when [BuyVM](https://buyvm.net) has KVM slices in stock.
## Usage ## 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: 2. Build Docker container:

180
matrix.py Normal file
View File

@ -0,0 +1,180 @@
"""
matrix.py
A module for interacting with the Matrix protocol.
Classes:
MatrixBot: A Matrix bot that can send messages and markdown messages to a room.
Dependencies:
markdown: A library for converting markdown to HTML.
loguru: A library for logging.
nio: A library for interacting with the Matrix protocol.
"""
import markdown
from loguru import logger
from nio import AsyncClient, LoginResponse
class MatrixBot:
"""
A Matrix bot that can send messages and markdown messages to a room.
Attributes:
config (dict): A dictionary containing the bot's configuration.
Expected keys are 'homeserver', 'username', 'password', 'room_id'.
client (AsyncClient): The Matrix client instance.
logged_in (bool): Whether the bot is currently logged in.
Methods:
__init__: Initializes the bot with a given configuration.
ensure_logged_in: Ensures that the bot is logged in to the Matrix homeserver.
send_message: Sends a message to the room specified in the bot's configuration.
send_markdown: Sends a markdown formatted message to the room specified in the bot's configuration.
close: Log out from the Matrix homeserver and close the client.
"""
def __init__(self, config: dict):
"""
A Matrix bot that can send messages and markdown messages to a room.
Args:
config (dict): A dictionary containing the bot's configuration.
Expected keys are 'homeserver', 'username', 'password', 'room_id'.
"""
self.config = config
self.client = AsyncClient(
homeserver=self.config['homeserver'],
user=self.config['username']
)
self.logged_in = False
async def ensure_logged_in(self):
"""
Ensures that the bot is logged in to the Matrix homeserver.
If the bot is not logged in, attempts to log in using the provided
password. If the login attempt fails, logs the error and closes the
nio session.
If an exception occurs during the login attempt, logs the error and
re-raises it.
"""
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):
"""
Sends a message to the room specified in the bot's configuration.
The message is sent as a simple text message, with the 'msgtype' set to
'm.text' and the 'body' set to the provided message.
If the bot is not logged in, attempts to log in using the provided
password. If the login attempt fails, logs the error and closes the
nio session.
If an exception occurs during the login attempt or the message sending,
logs the error and re-raises it.
Args:
message (str): The message to send to the room.
"""
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):
"""
Sends a markdown formatted message to the room specified in the bot's
configuration.
The message is sent as a text message with the 'msgtype' set to
'm.text', the 'body' set to the provided message, and the 'format'
set to 'org.matrix.custom.html'. The 'formatted_body' is set to the
markdown formatted message.
If the bot is not logged in, attempts to log in using the provided
password. If the login attempt fails, logs the error and closes the
nio session.
If an exception occurs during the login attempt or the message sending,
logs the error and re-raises it.
Args:
message (str): The message to send to the room.
"""
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):
"""
Log out from the Matrix homeserver and close the client.
If the bot is logged in, attempts to log out using the provided
password. If the login attempt fails, logs the error and closes the
nio session.
If an exception occurs during the login attempt or the message sending,
logs the error and re-raises it.
"""
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

View File

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

View File

@ -1,9 +1,10 @@
"""buyvm stock checker""" """buyvm stock checker"""
import json
import asyncio
import requests import requests
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from loguru import logger from loguru import logger
from matrix import MatrixBot
from config import DISCORD_WEBHOOK
BASE_URL = 'https://my.frantech.ca/' BASE_URL = 'https://my.frantech.ca/'
URLS = [ URLS = [
@ -14,14 +15,16 @@ 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): def get_url(url):
"""
Fetches a URL and returns its text content.
Args:
url (str): The URL to fetch.
Returns:
str: The text content of the page, or None if there was an error.
"""
try: try:
response = requests.get(url) response = requests.get(url)
response.raise_for_status() response.raise_for_status()
@ -33,6 +36,18 @@ def get_url(url):
def get_packages(html): def get_packages(html):
"""
Takes a string of HTML and extracts all the packages from it.
Args:
html (str): The HTML to parse.
Returns:
list: A list of packages, each represented as a dictionary with the following keys:
'name' (str): The name of the package.
'qty' (int): The current quantity of the package available.
'url' (str): The URL to order the package from, or an empty string if the package is not available.
"""
soup = BeautifulSoup(html, 'html.parser') soup = BeautifulSoup(html, 'html.parser')
packages = [] packages = []
@ -58,8 +73,31 @@ def get_packages(html):
return packages return packages
def main(): def load_config(filename):
with open(filename) as f:
return json.load(f)
async def main():
"""
Check BuyVM for available KVM slices and alert to a Matrix room if any are found.
The following configuration options are supported:
- `memory`: A list of integers specifying the memory quantities to check for.
Defaults to [512, 1, 2, 4], which corresponds to a price of $15.00 or less.
The function will log in to the Matrix server specified in the configuration,
then check each URL in `URLS` for available KVM slices. If any are found,
it will send a message to the room specified in the configuration with the
package name and quantity, and a link to order. Finally, it will close the
Matrix session.
"""
logger.info('checking buyvm stocks') 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: for url in URLS:
html = get_url(url) html = get_url(url)
@ -68,23 +106,31 @@ def main():
packages = get_packages(html) packages = get_packages(html)
for package in packages: 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") 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__': if __name__ == '__main__':
main() main_with_shutdown()