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):
    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):
    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():
    logger.info('checking buyvm stocks')

    # load configuration
    config = load_config('config.json')

    # initialize bot
    bot = MatrixBot(config['matrix'])

    for url in URLS:
        html = get_url(url)

        if not html:
            continue

        packages = get_packages(html)
        for package in packages:
            if package['qty'] > 0:
                logger.info(f"{package['name']}: {package['qty']} in stock")
                await bot.send_message(f"{package['name']}: {package['qty']} in stock")

    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()
        logger.info("Event loop closed.")


if __name__ == '__main__':
    main_with_shutdown()