Restock Check Bot
[DEFAULT] | |
INTERVAL_MINS = 5 | |
[TELEGRAM] | |
TOKEN = *** | |
RECEIVER_ID = ### |
import configparser | |
from bs4 import BeautifulSoup | |
from restock import StockCheck | |
from telegram_bot import TelegramBot | |
from datetime import datetime | |
import time | |
# Loading config | |
config = configparser.ConfigParser() | |
config.read('config.ini') | |
# Initialize scraping classes | |
# TODO Consider function to lambda | |
def legoshopCheck(res): | |
soup = BeautifulSoup(res.text, 'html.parser') | |
stock_div = soup.find("p", attrs={"data-test": "product-overview-availability"}) | |
if stock_div == None: | |
return True | |
stock_result = stock_div.text | |
import re | |
p = re.compile('품절') | |
m = p.search(stock_result) | |
return True if m == None else False | |
# main script | |
if __name__ == "__main__": | |
bot = TelegramBot(config['TELEGRAM']['TOKEN']) | |
bot.sendMessage(config['TELEGRAM']['RECEIVER_ID'], "Monitoring started.") | |
lego_friends_perk = StockCheck("Friends Central Perk" | |
, "https://www.lego.com/ko-kr/product/central-perk-21319" | |
, legoshopCheck, "utf-8") | |
sleep_mins = config['DEFAULT']['INTERVAL_MINS'] | |
def check(checkTargetArray): | |
return list(map(lambda item: item.statusChanged(), checkTargetArray)) | |
while True: | |
returns = check([lego_friends_perk]) | |
print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), returns) | |
alerts = list(filter(lambda item: item[0] , returns)) | |
for item in alerts: | |
bot.sendMessage(config['TELEGRAM']['RECEIVER_ID'], "{} status has changed to {}".format(item[3].name, item[0])) | |
time.sleep(int(sleep_mins) * 60) |
import requests | |
class StockCheck: | |
def __init__(self, name, url, checkMethod, encoding): | |
self.name = name | |
self.url = url | |
self.checkMethod = checkMethod | |
self.encoding = encoding | |
self.last_status = False | |
def getResponse(self): | |
# TODO Pass parameters def getResponse(self, params): | |
# TODO Add some error handling | |
URL = self.url | |
return requests.get(URL) | |
def check(self): | |
res = self.getResponse() | |
if res.encoding != self.encoding: | |
res.encoding = self.encoding | |
return self.checkMethod(res) | |
def statusChanged(self): | |
status = self.check() | |
if (self.last_status != status): | |
self.last_status = status | |
return (True, not(self.last_status), self.last_status, self) | |
return (False, self.last_status, status, self) | |
def __str__(self): | |
return "{} is {}".format(self.name, self.last_status) | |
if __name__ == "__main__": | |
def legoshopCheck(res): | |
from bs4 import BeautifulSoup | |
soup = BeautifulSoup(res.text, 'html.parser') | |
stock_div = soup.find("p", attrs={"data-test": "product-overview-availability"}) | |
if stock_div == None: | |
return True | |
stock_result = stock_div.text | |
import re | |
p = re.compile('품절') | |
m = p.search(stock_result) | |
return True if m == None else False | |
lego_friends_perk = StockCheck("Friends Central Perk" | |
, "https://www.lego.com/ko-kr/product/central-perk-21319" | |
, legoshopCheck, "utf-8") | |
stock = lego_friends_perk.check() | |
print(lego_friends_perk.name, "Available? ", stock) | |
(status_changed, last_status, current_status) = lego_friends_perk.statusChanged() | |
print(lego_friends_perk.name, "Status Changed? ", status_changed, ", Last Status? ", last_status, ", Current Status? ", current_status) |
import telegram | |
class TelegramBot: | |
def __init__(self, token): | |
self.token = token | |
self.bot = telegram.Bot(token = self.token) | |
def sendMessage(self, chat_id, message): | |
self.bot.sendMessage(chat_id = chat_id, text=message) | |
if __name__ == "__main__": | |
token = "*******" # your token | |
bot = TelegramBot(token) | |
receiver_id = *** # your id | |
bot.sendMessage(receiver_id, "Hello") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment