tgtg-notifier/checker.py

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):
    print(f"Sending {data} to {url}")
    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))

def notify_all():
    total = sum(previous.values())
    notify(f"List checked. Total available: {total}", url=NTFY_CHANNEL_DEBUG)


schedule.every(1).minutes.do(check_watchlist)
schedule.every(60).minutes.do(notify_all)


while True:
    schedule.run_pending()
    #check_watchlist()
    time.sleep(30)