agatha/stockbot-buyvm
1
Fork 0
mirror of https://git.juggalol.com/agatha/stockbot-buyvm.git synced 2024-11-22 07:36:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: agatha:269ea6db08d0ca72edaa43dd0f882ad442c1088e
Branches Tags
agatha:master
agatha:matrix
..
compare: agatha:544796e0ca900f1ef32e254fbe2dd22fc1f8c104
Branches Tags
agatha:master
agatha:matrix

No commits in common. "269ea6db08d0ca72edaa43dd0f882ad442c1088e" and "544796e0ca900f1ef32e254fbe2dd22fc1f8c104" have entirely different histories.
269ea6db08 ... 544796e0ca

7 changed files with 31 additions and 273 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

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

6
.pylintrc
View File

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

2
Dockerfile
View File

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

14
README.md
View File

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

180
matrix.py
View File

@ -1,180 +0,0 @@
"""
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

2
requirements.txt
View File

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

98
stockbot-buyvm.py
View File

@ -1,10 +1,9 @@
"""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 = [
@ -15,16 +14,14 @@ 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):
"""
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:
response = requests.get(url)
response.raise_for_status()
@ -36,18 +33,6 @@ def get_url(url):
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')
packages = []
@ -73,31 +58,8 @@ def get_packages(html):
return packages
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.
"""
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)
@ -106,31 +68,23 @@ async def main():
packages = get_packages(html)
for package in packages:
qty = package['qty']
memory = int(package['name'].split()[-1][:-2])
if qty > 0 and (memory in memory_filter):
if package['qty'] > 0:
logger.info(f"{package['name']}: {package['qty']} in stock")
await bot.send_message(f"🚨 {package['name']}: {package['qty']} in stock 🚨\n{package['url']}")
await bot.close()
def main_with_shutdown():
loop = asyncio.get_event_loop()
main_task = loop.create_task(main())
try:
loop.run_until_complete(main_task)
except asyncio.CancelledError:
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()
send_notification({
"username": "stockbot-buyvm",
"embeds": [
{
"author": {
"name": "BuyVM",
},
"title": package['name'],
"url": package['url'],
"description": f"{package['qty']} in stock now!"
}
],
"content": "STOCK ALERT"
})
if __name__ == '__main__':
main_with_shutdown()
main()