Rewritten at:
https://github.com/khaudio/mc-stock
Wrapped up real nice in a class now.
maybe i’ll write up a tutorial later for people just looking to use it to score a gpu; out of time for now.
more later.
example code at the bottom
git clone https://github.com/khaudio/mc-stock.git
#/usr/bin/env python
"""
mc-stock checks stock on specified items at Microcenter store locations, and
sends email notifications when changes are detected. Applicably, it helps
the user obtain rare items during shortages, like graphics cards.
"""
from getpass import getpass
from multiprocessing import cpu_count, Pool, Queue
from re import search
from requests import get
from time import sleep
from smtplib import SMTP
class MCStock(object):
def __init__(self, storeNum):
self.storeNum = '131' if not storeNum else str(storeNum)
class Store(MCStock):
def __init__(self, storeNum, server=None, port=None, sender=None, password=None, recipient=None, debug=False):
super().__init__(storeNum)
self.items = set()
self.newInStock, self.totalInStock = 0, 0
self.debug = False if debug is None else debug # Setting debug to True enables false positives for testing
self.server = 'smtp.gmail.com' if server is None else server
self.port = 587 if port is None else port
self.sender = input('Enter sender email address: ') if sender is None else sender
if not self.sender:
raise ValueError('Sender address cannot be empty')
self.recipient = sender if recipient is None else recipient # Assumes loopback if recipient is not provided
if self.server and self.sender and not password:
self.password = getpass('Enter email password: ')
else:
self.password = password
self.recipient = recipient
def __str__(self):
return '\n'.join(item.__str__() for item in self.items)
def add(self, *links):
for link in links:
if isinstance(link, (list, tuple, set)):
for l in link:
self.add_link(l)
elif isinstance(link, str):
self.add_link(link)
else:
raise TypeError('Links must be a string or list of strings')
def add_link(self, link):
if isinstance(link, str):
if link not in (item.link for item in self.items):
new = Item(self.storeNum, link)
new.update()
self.items.add(new)
else:
raise TypeError('Link must be a string')
def remove(self, links):
for link in links:
for item in self.items:
if link == item.link:
self.items.remove(item)
def email_message(self, q):
new = []
while not q.empty():
new.append(q.get())
message = '\n'.join(item.__str__() for item in new)
return message
def email_subject(self):
return f'({self.newInStock} new, {self.totalInStock} total) items in stock at Microcenter {self.storeNum}'
def send_email(self, subject, message):
server = SMTP(self.server, self.port)
server.ehlo()
server.starttls()
server.login(self.sender, self.password)
body = '\n'.join([f'To: {self.recipient}',
f'From: {self.sender}',
f'Subject: {subject}'
'', message])
try:
server.sendmail(self.sender, self.recipient, body)
sent = True
except:
sent = False
server.quit()
return sent
def update(self):
q = Queue()
p = Pool(cpu_count())
for item in p.imap(self.update_item, self.items):
q.put(item)
self.newInStock = q.qsize()
self.totalInStock = sum(item.stock for item in self.items)
return q
def run(self, minutes=15):
if isinstance(minutes, int):
seconds = minutes * 60
else:
raise TypeError('Minutes must be an integer or float')
while True:
q = self.update()
subject = self.email_subject()
message = self.email_message(q)
if self.newInStock:
if self.send_email(subject, message):
print('Recipient notified of stock changes')
sleep(seconds)
def update_item(self, item):
item.update()
if item.stock and item.stockChanged or self.debug:
return item
class Item(MCStock):
def __init__(self, storeNum, link):
super().__init__(storeNum)
self.link = link
self.sku = None
self.price = None
self.stock = None
self.stockChanged = False
self.priceChanged = False
def __str__(self):
if self.stock:
stock = 'in stock'
else:
stock = 'out of stock'
return f'SKU {self.sku} is {stock} for {self.price} at Microcenter {self.storeNum}\n{self.link}\n'
def pull(self):
page = get(self.link, cookies={'storeSelected': self.storeNum}).text
return page
def parse_lines(self, page):
for var in ['SKU', 'inStock', 'productPrice']:
reply = search(f"(?<='{var}':').*?(?=',)", page)
if reply:
yield reply.group()
def update(self):
page = str(self.pull())
data = tuple(self.parse_lines(page))
if not data or any(data) is None:
raise ValueError('Data missing from request or store number invalid')
self.sku, inStock, price = int(data[0]), data[1], float(data[2])
if inStock == 'True':
stock = True
else:
stock = False
if stock != self.stock and self.stock is not None:
self.stockChanged = True
else:
self.stockChanged = False
if price != self.price and self.price is not None:
self.priceChanged = True
else:
self.priceChanged = False
self.stock = stock
self.price = price
if __name__ == '__main__':
rx570 = ['http://www.microcenter.com/product/478850/Radeon_RX-570_ROG_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478907/Radeon_RX_570_Overclocked_4GB_GDDR5_Video_Card']
mc = Store(131)
mc.add(rx570)
mc.run()
you can, just like, ignore all this below here, ya know… it’s bad bad code, so just… scroll back up. please.
pasting here because why the hell not and in case anyone is interested in scraping the page for their local store’s vega stock or whatever.
replace or add item lists with the items you want
also replace the store ID number (you can find it in page source fairly easily), email info, and store info. I let it sleep for 5 seconds so it doesn’t hit too much too fast.
Also you can cron that shit to run as often as you like.
*/5 * * * * /usr/bin/python3 /mc_crawler/RX_5x0_Alert.py
#!/usr/bin/env python3
import re
import requests
from time import sleep
import smtplib
try:
import cPickle as pickle
except:
import pickle
import os
cookies = dict(storeSelected='YOURSTORENUMBER')
global inStockItems
itemURLs = []
global msgText
msgText = ""
stockCurrent = {}
if os.path.exists("items_in_stock"):
stockLast = pickle.load(open(r"items_in_stock", "rb"))
class DictDiffer(object):
"""
Calculate the difference between two dictionaries as:
(1) items added
(2) items removed
(3) keys same in both but changed values
(4) keys same in both and unchanged values
"""
def __init__(self, current_dict, past_dict):
self.current_dict, self.past_dict = current_dict, past_dict
self.set_current, self.set_past = set(current_dict.keys()), set(past_dict.keys())
self.intersect = self.set_current.intersection(self.set_past)
def added(self):
return self.set_current - self.intersect
def removed(self):
return self.set_past - self.intersect
def changed(self):
return set(o for o in self.intersect if self.past_dict[o] != self.current_dict[o])
def unchanged(self):
return set(o for o in self.intersect if self.past_dict[o] == self.current_dict[o])
rx570 = [
'http://www.microcenter.com/product/478850/Radeon_RX-570_ROG_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478907/Radeon_RX_570_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478810/Radeon_RX-570_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478364/Radeon_RX_570_GAMING_X_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478668/Radeon_RX_570_Gaming_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478683/Radeon_NITRO_RX_570_Overclocked_8GB_GDDR5_Video_Card'
]
rx580= [
'http://www.microcenter.com/product/478666/Radeon_RX_580_Gaming_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478363/Radeon_RX_580_ARMOR_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/479406/Radeon_RX_580_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/479279/Radeon_RX_580_Overclocked_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478664/Radeon_RX_580_Gaming_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478662/AORUS_Radeon_RX_580_XTR_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478663/AORUS_Radeon_RX_580_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/479526/Red_Dragon_Radeon_RX-580_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478360/Radeon_RX_580_GAMING_X_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478682/Radeon_Pulse_RX_580_Overclocked_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478700/AXRX_Radeon_RX-580_Red_Devil_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478905/Radeon_RX_580_Overclocked_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478802/Radeon_RX-580_Overclocked_8GB_GDDR5_Graphics_Card',
'http://www.microcenter.com/product/478849/Radeon_RX-580_ROG_Overclocked_8GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478362/Radeon_RX_580_GAMING_X_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478665/AORUS_Radeon_RX_580_4GB_GDDR5_Video_Card',
'http://www.microcenter.com/product/478701/AXRX_Radeon_RX-580_Red_Devil_Overclocked_8GB_GDDR5_Video_Card'
]
for item in rx570:
respData = requests.get(item, cookies=cookies).text
skuNum = re.findall(r'SKU:"(.*?)",',str(respData))
inStock = re.findall(r'inStock:"(.*?)",',str(respData))
productPrice = re.findall(r'price:"(.*?)",',str(respData))
storeId = re.findall(r'storeId:"(.*?)",',str(respData))
stockCurrent[skuNum[0]] = inStock[0]
for stock in inStock:
if stock == "True":
msgText = msgText+"RX 570 -- SKU: "+skuNum[0]+" -- "+productPrice[0]+"\n"+item+"\n\n"
elif stock == "False":
print("RX 570 -- SKU: "+skuNum[0]+" -- Out of stock")
else:
print("Error retrieving stock")
sleep(5)
for item in rx580:
respData = requests.get(item, cookies=cookies).text
skuNum = re.findall(r'SKU:"(.*?)",',str(respData))
inStock = re.findall(r'inStock:"(.*?)",',str(respData))
productPrice = re.findall(r'price:"(.*?)",',str(respData))
storeId = re.findall(r'storeId:"(.*?)",',str(respData))
stockCurrent[skuNum[0]] = inStock[0]
for stock in inStock:
if stock == "True":
msgText = msgText+"RX 580 -- SKU: "+skuNum[0]+" -- "+productPrice[0]+"\n"+item+"\n\n"
elif stock == "False":
print("RX 580 -- SKU: "+skuNum[0]+" -- Out of stock")
else:
print("Error retrieving stock")
sleep(5)
with open(r"items_in_stock", "wb") as outfile:
pickle.dump(stockCurrent, outfile)
stockDiff = DictDiffer(stockCurrent, stockLast).changed()
if stockDiff:
print("Stock changed:")
for i in stockDiff:
if stockCurrent[i] == "False":
print(i+" -- "+"Out of stock")
elif stockCurrent[i] == "True":
print(i+" -- "+"IN STOCK")
else:
print("Stock unchanged")
storeInfo="""Paste store address and other info here"""
if len(msgText) and stockDiff:
print(msgText)
TO = ['[email protected]']
SUBJECT = 'GPU IN STOCK at Microcenter'
TEXT = msgText+"\n\n"+storeInfo
# Gmail Sign In
gmail_sender = '[email protected]'
gmail_passwd = 'YourServerEmailPW'
server = smtplib.SMTP('smtp.gmail.com', 587)
server.ehlo()
server.starttls()
server.login(gmail_sender, gmail_passwd)
BODY = '\r\n'.join(['To: %s' % TO,
'From: %s' % gmail_sender,
'Subject: %s' % SUBJECT,
'', TEXT])
try:
server.sendmail(gmail_sender, TO, BODY)
print ('Email sent')
except:
print ('Error sending mail')
server.quit()
else:
print("No email sent")
del inStock,msgText