From 4e93d80a687fa7131b9860e83d975e117732e637 Mon Sep 17 00:00:00 2001 From: Hykilpikonna Date: Thu, 3 Mar 2022 00:18:24 -0500 Subject: [PATCH] [O] Encapsulate telegram reporter --- src/PS5TrackingBot.py | 23 +++++++-------------- src/StockTrackingBot.py | 26 ++++++++++-------------- src/utils.py | 45 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 61 insertions(+), 33 deletions(-) diff --git a/src/PS5TrackingBot.py b/src/PS5TrackingBot.py index 25897a7..c23a2f8 100644 --- a/src/PS5TrackingBot.py +++ b/src/PS5TrackingBot.py @@ -1,12 +1,11 @@ import os import time -import requests from selenium.webdriver import Chrome from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By -from utils import parse_retry +from utils import parse_retry, TelegramReporter, CSS # Config # Telegram chat ID that receives update messages (could be a channel in @channel_id format) @@ -18,16 +17,8 @@ TG_TOKEN = os.environ['TG_TOKEN'] ALERT_RECEIVER = -1001655384423 # Constants -CSS = By.CSS_SELECTOR AVAIL_TABLE: dict[str, bool] = {} - - -def send_message(msg: str): - r = requests.get(f'https://api.telegram.org/bot{TG_TOKEN}/sendMessage', - params={'chat_id': TG_RECEIVER, 'parse_mode': 'Markdown', 'text': msg}) - - if r.status_code != 200: - print('Request not OK:', r.status_code, r.text) +TG = TelegramReporter(TG_TOKEN, TG_RECEIVER, ALERT_RECEIVER) def parse_page(browser: Chrome): @@ -41,7 +32,7 @@ def parse_page(browser: Chrome): # Not available, check if it was previously available if len(avail) == 0: if title in AVAIL_TABLE: - send_message(f'Sold out: `{title}`') + TG.send(f'Sold out: `{title}`') del AVAIL_TABLE[title] continue @@ -58,11 +49,11 @@ def parse_page(browser: Chrome): # Available and meets threshold criteria, notify user AVAIL_TABLE[title] = True - send_message(f'PS5 Became Available!\n' - f'- [{title}]({link}) ${price:.2f}') + TG.send(f'PS5 Became Available!\n' + f'- [{title}]({link}) ${price:.2f}') # Check alert - + TG.alert() if __name__ == '__main__': @@ -72,7 +63,7 @@ if __name__ == '__main__': browser = Chrome(options=web_options) browser.get('https://www.bestbuy.ca/en-ca/category/ps5-consoles/17583383') - send_message('Bot started') + TG.send('Bot restarted') # parse_page(browser) # browser.close() diff --git a/src/StockTrackingBot.py b/src/StockTrackingBot.py index 7dd4ad3..fe2bc80 100644 --- a/src/StockTrackingBot.py +++ b/src/StockTrackingBot.py @@ -7,7 +7,7 @@ from selenium.webdriver import Chrome from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By -from utils import parse_retry +from utils import parse_retry, TelegramReporter, CSS # Config # Price increase ratio threshold (ignore everything higher than this ratio) @@ -17,9 +17,10 @@ INCR_MAX = 0.2 TG_RECEIVER = '@toronto_bestbuy_gpu' # Telegram bot token TG_TOKEN = os.environ['TG_TOKEN'] +# Alert receiver telegram chat ID +ALERT_RECEIVER = -1001655384423 # Constants -CSS = By.CSS_SELECTOR MODELS = [ ['3060 ti', 400, 132], ['3070 ti', 600, 167], @@ -34,6 +35,7 @@ USD_TO_CAD = 1.27 AVAIL_TABLE: dict[str, bool] = {} IGNORED = [] TITLE_SHORTEN = re.compile('(rtx|nvidia|geforce|edition|gddr[56]x*|video|card)', flags=re.IGNORECASE) +TG = TelegramReporter(TG_TOKEN, TG_RECEIVER, ALERT_RECEIVER) def shorten_title(title: str): @@ -43,14 +45,6 @@ def shorten_title(title: str): return short_title.strip() -def send_message(msg: str): - r = requests.get(f'https://api.telegram.org/bot{TG_TOKEN}/sendMessage', - params={'chat_id': TG_RECEIVER, 'parse_mode': 'Markdown', 'text': msg}) - - if r.status_code != 200: - print('Request not OK:', r.status_code, r.text) - - def parse_page(browser: Chrome): # Parse page for item in browser.find_elements(By.CLASS_NAME, 'x-productListItem'): @@ -67,7 +61,7 @@ def parse_page(browser: Chrome): # Not available, check if it was previously available if len(avail) == 0: if title in AVAIL_TABLE: - send_message(f'Sold out: `{title}`') + TG.send(f'Sold out: `{title}`') del AVAIL_TABLE[title] continue @@ -103,10 +97,10 @@ def parse_page(browser: Chrome): # Available and meets threshold criteria, notify user AVAIL_TABLE[title] = True - send_message(f'{model[0].upper()} Became Available!\n' - f'\n' - f'${price:.0f} | {price_incr * 100:.0f}% Incr | Value: {value:.0f}\n' - f'- [{title}]({link})') + TG.send(f'{model[0].upper()} Became Available!\n' + f'\n' + f'${price:.0f} | {price_incr * 100:.0f}% Incr | Value: {value:.0f}\n' + f'- [{title}]({link})') if __name__ == '__main__': @@ -120,6 +114,8 @@ if __name__ == '__main__': # parse_page(browser) # browser.close() + TG.send('Bot restarted') + # Refresh indefinitely while True: time.sleep(5) diff --git a/src/utils.py b/src/utils.py index 0d45826..5ec3762 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,8 +1,15 @@ -import traceback -from typing import Callable +from __future__ import annotations +import traceback +from typing import Callable, Optional + +import requests from selenium.common.exceptions import StaleElementReferenceException from selenium.webdriver.chrome.webdriver import WebDriver +from selenium.webdriver.common.by import By + + +CSS = By.CSS_SELECTOR def parse_retry(parser: Callable, browser: WebDriver, tries: int = 0): @@ -13,3 +20,37 @@ def parse_retry(parser: Callable, browser: WebDriver, tries: int = 0): parse_retry(parser, browser, tries + 1) except Exception as e: traceback.print_exc() + + +class TelegramReporter: + token: str + receiver: str | int + alert_receiver: Optional[str | int] + + def __init__(self, token: str, receiver: str, alert_receiver: Optional[str | int]): + self.token = token + self.receiver = receiver + self.alert_receiver = alert_receiver + + def send(self, msg: str, rec: Optional[str | int] = None) -> bool: + """ + Send a message + + :param msg: Message string + :param rec: Receiver + :return: Success or not + """ + if rec is None: + rec = self.receiver + + r = requests.get(f'https://api.telegram.org/bot{self.token}/sendMessage', + params={'chat_id': rec, 'parse_mode': 'Markdown', 'text': msg}) + + if r.status_code != 200: + print('Request not OK:', r.status_code, r.text) + + return r.status_code == 200 + + def alert(self) -> bool: + if self.alert_receiver: + return self.send('/alert', self.alert_receiver)