Skip to content

Instantly share code, notes, and snippets.

@samsee
Last active October 4, 2021 16:58
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save samsee/8b25cf4f0e9c6fac1ded6b0574b44532 to your computer and use it in GitHub Desktop.
Save samsee/8b25cf4f0e9c6fac1ded6b0574b44532 to your computer and use it in GitHub Desktop.
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