stockbot-buyvm/stockbot-buyvm.py

"""buyvm stock checker"""
import json
import asyncio
import requests
from bs4 import BeautifulSoup
from loguru import logger
from matrix import MatrixBot

BASE_URL = 'https://my.frantech.ca/'
URLS = [
    'https://my.frantech.ca/cart.php?gid=37',  # Las Vegas
    'https://my.frantech.ca/cart.php?gid=38',  # New York
    'https://my.frantech.ca/cart.php?gid=48',  # Miami
    'https://my.frantech.ca/cart.php?gid=39',  # Luxembourg
]


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()
    except requests.RequestException as e:
        logger.error(f'error fetching {url}: {str(e)}')
        return None

    return response.text


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 = []

    package_elements = soup.find_all('div', class_='package')
    for package_element in package_elements:
        package = {}

        package_name = package_element.find('h3', class_='package-name').text.strip()
        package['name'] = package_name

        package_quantity = package_element.find('div', class_='package-qty').text.strip()
        package['qty'] = int(package_quantity.split()[0])

        order_button = package_element.find('a', class_='btn-primary')
        if order_button:
            order_url = order_button['href']
            package['url'] = BASE_URL + order_url
        else:
            package['url'] = ''

        packages.append(package)

    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.
    """
    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)

        if not html:
            continue

        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):
                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()


if __name__ == '__main__':
    main_with_shutdown()