60 lines
1.4 KiB
Python
60 lines
1.4 KiB
Python
|
from tgtg import TgtgClient
|
||
|
|
from pprint import pprint
|
||
|
|
import json
|
||
|
|
import time
|
||
|
|
import requests
|
||
|
|
import random
|
||
|
|
#import schedule
|
||
|
|
from collections import defaultdict
|
||
|
|
|
||
|
|
previous = defaultdict(int)
|
||
|
|
|
||
|
|
creds = {}
|
||
|
|
|
||
|
|
|
||
|
|
with open('.credentials.json', 'r') as f:
|
||
|
|
creds = json.load(f)
|
||
|
|
|
||
|
|
with open('.config.json', 'r') as f:
|
||
|
|
config = json.load(f)
|
||
|
|
NTFY_CHANNEL = config["ntfy_channel"]
|
||
|
|
watchlist = config["watchlist"]
|
||
|
|
|
||
|
|
NTFY_CHANNEL_DEBUG = NTFY_CHANNEL + "-debug"
|
||
|
|
|
||
|
|
|
||
|
|
client = TgtgClient(access_token=creds["access_token"], refresh_token=creds["refresh_token"], user_id=creds["user_id"], cookie=creds["cookie"])
|
||
|
|
|
||
|
|
|
||
|
|
def check(item):
|
||
|
|
#pprint(item)
|
||
|
|
available = int(item["items_available"])
|
||
|
|
id = item["store"]["store_id"]
|
||
|
|
print(f"{id:<10} - { item['display_name']}: { item['items_available']} available")
|
||
|
|
if available != previous[id]:
|
||
|
|
previous[id] = available
|
||
|
|
notify(f"{ available } disponibles para {item['display_name'] }")
|
||
|
|
return available
|
||
|
|
|
||
|
|
|
||
|
|
def notify(data, url=NTFY_CHANNEL):
|
||
|
|
requests.post(url,
|
||
|
|
data=data.encode(encoding='utf-8'))
|
||
|
|
|
||
|
|
|
||
|
|
def check_watchlist():
|
||
|
|
total = 0
|
||
|
|
for id in watchlist:
|
||
|
|
item = client.get_item(id)
|
||
|
|
total += check(item)
|
||
|
|
time.sleep(random.randint(2, 10))
|
||
|
|
notify(f"List checked. Total available: {total}", url=NTFY_CHANNEL_DEBUG)
|
||
|
|
|
||
|
|
|
||
|
|
#schedule.every(1).minutes.do(check_watchlist)
|
||
|
|
|
||
|
|
while True:
|
||
|
|
#schedule.run_pending()
|
||
|
|
check_watchlist()
|
||
|
|
time.sleep(30)
|