[O] Encapsulate telegram reporter

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] = {}
-
+TG = TelegramReporter(TG_TOKEN, TG_RECEIVER, ALERT_RECEIVER)
-
-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):
@@ -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'
+        TG.send(f'PS5 Became Available!\n'
-                     f'- [{title}]({link}) ${price:.2f}')
+                        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()

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'
+        TG.send(f'{model[0].upper()} Became Available!\n'
-                     f'\n'
+                f'\n'
-                     f'${price:.0f} | {price_incr * 100:.0f}% Incr | Value: {value:.0f}\n'
+                f'${price:.0f} | {price_incr * 100:.0f}% Incr | Value: {value:.0f}\n'
-                     f'- [{title}]({link})')
+                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)

src/utils.py

@@ -1,8 +1,15 @@
-import traceback
+from __future__ import annotations
-from typing import Callable
 
+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)