98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
import os
|
|
from pathlib import Path
|
|
import logging
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
from twilio.rest import Client
|
|
|
|
load_dotenv(verbose=True)
|
|
|
|
config = {
|
|
'api_url': 'https://www.boden.co.uk/api/stock/v1/styles?market=UK&language=en-gb&id=y1357',
|
|
'item_id': '40263442', # 40263447 low stock, 40263442 12-18m
|
|
'message_sent_file': './sent.txt',
|
|
'msg_to': os.environ['SMS_TO'],
|
|
'website_url': 'https://www.boden.co.uk/en-gb/rainbow-3-in-1-printed-jacket/sty-y1357',
|
|
'log_level': logging.WARN
|
|
}
|
|
|
|
|
|
def get_logger(log_level=logging.WARN):
|
|
"""Creates, configures and returns a logger that outputs to console with supplied log_level
|
|
"""
|
|
log = logging.getLogger(__name__)
|
|
log.setLevel(log_level)
|
|
log_handler = logging.StreamHandler()
|
|
log_handler.setFormatter(logging.Formatter(
|
|
'[%(filename)s:(%(lineno)d)] %(levelname)s >> %(message)s'))
|
|
log.addHandler(log_handler)
|
|
|
|
return log
|
|
|
|
|
|
def get_json_from_url(url: str) -> dict:
|
|
req = requests.get(url)
|
|
return req.json()
|
|
|
|
|
|
def send_sms(target: str, msg: str):
|
|
account_sid = os.environ['TWILIO_ACCOUNT_SID']
|
|
auth_token = os.environ['TWILIO_AUTH_TOKEN']
|
|
messaging_sid = os.environ['TWILIO_MESSAGE_SID']
|
|
|
|
client = Client(account_sid, auth_token)
|
|
|
|
message = client.messages.create(
|
|
messaging_service_sid=messaging_sid,
|
|
body=msg,
|
|
to=target
|
|
)
|
|
|
|
log = logging.getLogger(__name__)
|
|
log.debug(message.sid)
|
|
|
|
|
|
def does_file_exist(lock_file: str) -> bool:
|
|
"""Checks for presence of lockfile
|
|
|
|
Args:
|
|
lock_file (str): path to lockfile
|
|
|
|
Returns:
|
|
(bool): True if file exists
|
|
"""
|
|
return os.path.isfile(lock_file)
|
|
|
|
|
|
def create_file(lock_file):
|
|
Path(lock_file).touch()
|
|
|
|
|
|
def main(config, log):
|
|
# This script runs once, to avoid spamming SMS every time run
|
|
if does_file_exist(config['message_sent_file']):
|
|
log.info(
|
|
f'Already had positive result, remove {config["message_sent_file"]} to allow script to run again')
|
|
raise StopIteration('Already run once')
|
|
|
|
json = get_json_from_url(config['api_url'])
|
|
|
|
for id in json['Y1357']:
|
|
if id['edp'] == config['item_id']:
|
|
if id['stockStatus'] == 'OutOfStock':
|
|
log.info("Item is out of stock")
|
|
return False
|
|
else:
|
|
log.info("Item is in stock")
|
|
send_sms(
|
|
config['msg_to'], 'The coat is in stock at ' + config['website_url'])
|
|
|
|
# Prevents it being run again
|
|
create_file(config['message_sent_file'])
|
|
return True
|
|
|
|
|
|
if __name__ == '__main__':
|
|
log = get_logger(config['log_level'])
|
|
main(config, log)
|