Last active
October 4, 2021 16:58
-
-
Save samsee/8b25cf4f0e9c6fac1ded6b0574b44532 to your computer and use it in GitHub Desktop.
Restock Check Bot
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[DEFAULT] | |
INTERVAL_MINS = 5 | |
[TELEGRAM] | |
TOKEN = *** | |
RECEIVER_ID = ### |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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