xkcd_alt.py

"""This is the XKCD Alt Text Bot.

This bot checks once a minute for new Tweets from @xkcdComic. If one is found, it accesses the
linked comic, extracts the image alt text, and Tweets it as a reply."""

import time # Program sleeping
import os # API keys & access tokens
import requests # Accessing API
from requests_oauthlib import OAuth1 # OAuth
from bs4 import BeautifulSoup # HTML searching

class Twitter():
    """This class handles all API requests to Twitter."""
    def __init__(self, auth):
        """This class constructor collects the OAuth keys for the class."""
        self.auth = auth
    
    def get(self):
        """This function determines if a new comic has been posted, and returns it.
        
        Mentions of 'comic' refer to @xkcdComic, mentions of 'alt' refer to @XKCDAltTextBot."""
        # Build payloads for comic bot and alt text bot search
        comic_payload = {'q': 'from:xkcdComic', 'result_type': 'recent', 'count': '1'}
        alt_payload = {'q': 'from:XKCDAltTextBot', 'result_type': 'recent', 'count': '1'}

        # Retrieve data from Twitter searches
        for attempt in range(6):
            if attempt == 5: # Too many attempts
                print('Twitter search failed ({}), see below response.'.format(str(comic_raw.status_code)))
                print('Twitter error message:\n\n{}'.format(comic_raw.json()))
                del alt_payload, comic_payload
                return 'crash' # Enter log protection mode
            
            print('Searching for new comics...')
            alt_raw = requests.get('https://api.twitter.com/1.1/search/tweets.json',
                                    params=alt_payload, auth=self.auth)
            comic_raw = requests.get('https://api.twitter.com/1.1/search/tweets.json',
                                    params=comic_payload, auth=self.auth)
            
            if comic_raw.status_code == 200: # Good request
                pass
            elif comic_raw.status_code >= 429 or comic_raw.status_code == 420:
                # Twitter issue or rate limiting
                print('Twitter search failed ({})'.format(comic_raw.status_code))
                print('Reattempting in 5 minutes...')
                time.sleep(300) # sleep for 5 minutes and reattempt
                continue
            else: # Other problem in code
                print('Twitter search failed ({}), see below '.format(str(comic_raw.status_code)) +
                      'response.')
                print('Twitter error message:\n\n{}'.format(comic_raw.json()))
                del alt_payload, comic_payload, alt_raw, comic_raw
                return 'crash' # Enter log protection mode
            
            # Convert to JSON
            alt = alt_raw.json()
            comic = comic_raw.json()

            if alt['statuses'][0]['id'] is None or \
            comic['statuses'][0]['in_reply_to_status_id'] is None:
                print('Twitter search failed: No Tweet found')
                del alt_payload, comic_payload, alt_raw, comic_raw, alt, comic
                return 'crash' # Enter log protection mode
            
            if alt['statuses'][0]['id'] == comic['statuses'][0]['in_reply_to_status_id']:
                # This tweet has already been replied to
                del alt_payload, comic_payload, alt_raw, comic_raw, alt, comic
                return None # Sleep for 60 seconds
            else:
                # This tweet has not been replied to
                del alt_payload, comic_payload, alt_raw, comic_raw, alt
                return comic['statuses'][0] # Return comic Tweet


    def post(self, tweet, reply):
        """This function Tweets the alt (title) text as a reply to @xkcdComic."""
        print('Tweeting...')

        tweet_payload = {'status': tweet, 'in_reply_to_status_id': reply,
                         'auto_populate_reply_metadata': 'true'}
        
        # POST Tweet
        for attempt in range(6):
            if attempt == 5: # Too many attempts
                print('Tweeting failed ({}), see below response.'.format(str(tweet.status_code)))
                print('Twitter error message:\n\n{}'.format(tweet.json()))
                del tweet_payload
                return 'crash' # Enter log protection mode

            tweet = requests.post('https://api.twitter.com/1.1/statuses/update.json', json=tweet_payload,
                        auth=self.auth)
            
            if tweet.status_code == 200: # Good request
                print('Successfully Tweeted:\n\n{}'.format(tweet.json()))
                del tweet, tweet_payload
                return None
            elif tweet.status_code >= 429 or tweet.status_code == 420 or \
            tweet.status_code == 403:
                # Twitter issue or rate limiting
                print('Tweeting failed ({})'.format(tweet.status_code))
                print('Reattempting in 5 minutes...')
                time.sleep(300) # sleep for 5 minutes and reattempt
                continue
            else: # Other problem in code
                print('Tweeting failed ({}), see below response.'.format(str(tweet.status_code)))
                print('Twitter error message:\n\n{}'.format(tweet.json()))
                del tweet, tweet_payload
                return 'crash' # Enter log protection mode

def get_auth():
    """This function retrieves the API keys and access tokens from environmental variables."""
    print("Building OAuth header...")
    key = [os.environ.get('XKCD_API_KEY', None),
           os.environ.get('XKCD_API_SECRET_KEY', None),
           os.environ.get('XKCD_ACCESS_TOKEN', None),
           os.environ.get('XKCD_ACCESS_SECRET_TOKEN', None)]
    for i in key:
        if i is None: # Verify keys were loaded
            print("OAuth initiation failed: Environmental variable not found")
            del key
            return 'crash' # Enter log protection mode

    auth = OAuth1(key[0], key[1], key[2], key[3])
    print('OAuth initiation successful!')
    del key
    return auth


def retrieve_text(site):
    """This retrieves the HTML of the website, isolates the image title text, and formats it for the
    Tweet."""
    for attempt in range(11):
        print('Accessing {} (attempt {} of 11'.format(site, attempt+1))
        html_raw = requests.get(site) # Retrieving raw HTML data
        if html_raw.status_code != 200: # Data not successfully retrieved
            if attempt < 6:
                print('Could not access XKCD ({}). '.format(html_raw.status_code) +
                      'Trying again in 10 seconds...')
                time.sleep(10) # Make 6 attempts with 10 second delays
            elif attempt < 10:
                print('Could not access XKCD ({}). '.format(html_raw.status_code) +
                      'Trying again in 60 seconds...')
                time.sleep(60) # Make 4 attempts with 60 seconds delays
            else:
                print('XKCD retrieval failed: could not access {}'.format(site))
                return 'crash' # Enter log protection mode
                
    html = BeautifulSoup(html_raw, 'html.parser')
    comic = html.find('img', title=True) # Locates the only image with title text (the comic)
    if comic is None:
        print('Title extraction failed: image not found')
        return 'crash' # Enter log protection mode
    
    title = comic['title'] # Extracts the title text
    tweet = 'Alt/title text: "{}"'.format(title) # Construct the main Tweet body

    print('Tweet constructed')
    del html_raw, html, comic, title
    return tweet

def crash():
    """This function protects logs by pinging google.com every 20 minutes."""
    print('Entering log protection mode.')
    while True:
        a = requests.get('https://google.com') # Ping Google
        del a
        time.sleep(1200)
        continue

# Main program
# All mentions of 'crash' mean the program has, and is entering log protection mode
auth = get_auth() # Build authentication header
if auth == 'crash':
    crash()
twitter = Twitter(auth)

while True: # Initialize main account loop
    original_tweet = twitter.get() # Check for new comics

    if original_tweet == 'crash':
        crash()
    elif original_tweet is None:
        print('No new comics found. Sleeping for 60 seconds...')
        time.sleep(60)
        continue
    else:
        body = retrieve_text(original_tweet['entities']['urls'][0]['expanded_url']) # Build Tweet
        if body == 'crash':
            crash()
        
        result = twitter.post(body, original_tweet['id_str']) # Post Tweet

        if result == 'crash':
            crash()
        elif result is None: # Successful Tweet
            print('Sleeping for 60 seconds...')
            time.sleep(60)
            continue