How I scheduled same-day Global Entry Interview (Part 3)

Connect to Twitter API

import os


class Auth:
    CONSUMER_KEY = os.environ.get('CONSUMER_KEY')
    CONSUMER_SECRET = os.environ.get('CONSUMER_SECRET')
    ACCESS_TOKEN = os.environ.get('ACCESS_TOKEN')
    TOKEN_SECRET = os.environ.get('TOKEN_SECRET')

class GlobalEntryBot(object):

    def __init__(self):
        self.auth = tweepy.OAuthHandler(Auth.CONSUMER_KEY, Auth.CONSUMER_SECRET)
        self.auth.set_access_token(Auth.ACCESS_TOKEN, Auth.TOKEN_SECRET)
        self.api = tweepy.API(self.auth)

As you can see an Auth class contains all of my secret keys that GlobalEntryBot.auth and GlobalEntryBot.auth.set_access needs to access twitter.  

Create Task

# main.py
import schedule
import sys

if __name__ == '__main__':
    logging.info("Started Twitter Bot")
    bot = GlobalEntryBot()
    schedule.every(1).minute.do(bot.check_global_entry_schedule)
    while True:
        try:
            schedule.run_pending()
        except KeyboardInterrupt:
            logging.info("Quitting!!")
            sys.exit()

I think that python's schedule module is appropriate for this project, so I've registered check_global_entry_schedule to run every 1 minute and then started all jobs with run_pending().

Final script

import datetime
import json
import logging
import re
import requests
import tweepy
import schedule
import sys
import os


class Auth:
    CONSUMER_KEY = os.environ.get('CONSUMER_KEY')
    CONSUMER_SECRET = os.environ.get('CONSUMER_SECRET')
    ACCESS_TOKEN = os.environ.get('ACCESS_TOKEN')
    TOKEN_SECRET = os.environ.get('TOKEN_SECRET')

class Constants:
    ATL_AIRPORT_ID = 5182
    LOCATION_ID = ATL_AIRPORT_ID
    SCHEDULER_API_ENDPOINT = 'https://ttp.cbp.dhs.gov/schedulerapi/slots?orderBy=soonest&limit=1&locationId={}&minimum=1'.format(
        LOCATION_ID)

class GlobalEntryBot(object):

    def __init__(self):
        self.auth = tweepy.OAuthHandler(Auth.CONSUMER_KEY, Auth.CONSUMER_SECRET)
        self.auth.set_access_token(Auth.ACCESS_TOKEN, Auth.TOKEN_SECRET)
        self.api = tweepy.API(self.auth)    
        self.old_appointment_date = None
        self.today = datetime.date.today()

    def check_global_entry_schedule(self):
        """
        check date for global entry interview
        """
        res = requests.get(Constants.SCHEDULER_API_ENDPOINT)
        next_available_date_time = json.loads(res.text)[0]['startTimestamp'].split("T")
        next_available_date = datetime.datetime.strptime(next_available_date_time[0], '%Y-%m-%d').date()
        if self.old_appointment_date is None:
            try:
                match = re.search(r'\d{4}-\d{2}-\d{2}', self.api.user_timeline(self.api.me().id, count=1)[0].text)
                if match is None:
                    self.old_appointment_date = next_available_date
                self.old_appointment_date = datetime.datetime.strptime(match.group(), '%Y-%m-%d').date()
            except:
                self.old_appointment_date = self.today + datetime.timedelta(365)  # set an arbitrary date
        logging.info("Comparing {} with {}".format(next_available_date, self.old_appointment_date))
        if next_available_date != self.old_appointment_date:
            self.old_appointment_date = next_available_date
            logging.info(
                "Next interview date: {} | Old next interview date: {}".format(next_available_date,
                                                                               self.old_appointment_date))
            try:
                self.api.update_status(
                    "Global Entry appointment available in Atlanta on {} at {}".format(
                        next_available_date, next_available_date_time[1]))
            except Exception as e:
                logging.info("Exception: {}".format(e))
                
                
if __name__ == '__main__':
    logging.info("Started Twitter Bot")
    bot = GlobalEntryBot()
    schedule.every(1).minute.do(bot.check_global_entry_schedule)
    while True:
        try:
            schedule.run_pending()
        except KeyboardInterrupt:
            logging.info("Quitting!!")
            sys.exit()

DEPLOY

I deployed to Heroku