Lambda Functions for the NHL Score Twitter Bot

README.md

Need to locally install all dependencies with pip install [package] -t ./ so they can be packaged before uploading to AWS. The packages need to be created in their own folders and the filenames changed to lambda_function.py.

Requirements

• The nhl package is required for get_nhl_game_brief

get_nhl_game_brief.py

import json
import pytz
from nhl import Api
from datetime import date, datetime, timedelta

def get_todays_games(api):
  today = datetime.now(pytz.UTC)
  today = today - timedelta(hours=12)
  return api.get_games_for_date(today)

def get_upcoming_games_and_date(api):
  date = datetime.now(pytz.UTC)
  date = date - timedelta(hours=12)
  games = api.get_games_for_date(date)
  while (len(games) == 0):
    date = date + timedelta(days=1)
    games = api.get_games_for_date(date)
  return date, games
# Construct a tweet based on what's going on right now around the NHL
def lambda_handler(event, context):
  api = Api()
  games = get_todays_games(api)
  if len(games) == 0:
    # There are no games today so find the first day there are games
    # in the future and show those
    date, games = get_upcoming_games_and_date(api)
    pre_msg = 'Next games on {}'.format(date.date().isoformat())
    lines = [game.short_summary for game in games]
    lines.insert(0, pre_msg)
    return lines
  else:
    return [game.short_summary for game in games]

if __name__ == '__main__':
  print(lambda_handler(None, None))

twitter_users_requesting_scores.py

import decimal
import json
import re
import os
import urllib

import boto3
import twitter

from boto3.dynamodb.conditions import Key, Attr

CONSUMER_KEY = os.environ['consumer_key']
CONSUMER_SECRET = os.environ['consumer_secret']
ACCESS_KEY = os.environ['access_token_key']
ACCESS_SECRET = os.environ['access_token_secret']
SEARCH_REGEX = os.environ['regex']

class AWSLambdaException(Exception): pass

def get_last_mention_id(table):
  response = table.query(KeyConditionExpression=Key('name').eq('last_mention_id'))

  items = response['Items']
  last_mention_id = None
  if len(items) == 1:
    last_mention_id = items[0]['val']

  return last_mention_id

def update_last_mention_id(table, last_id, new_id):
  if last_id is None:
    # Create a new entry
    table.put_item(
      Item={
           'name':'last_mention_id',
           'val':new_id
           }
    )
  else:
    # Update the existing entry
    table.update_item(
      Key={
          'name':'last_mention_id'
      },
      UpdateExpression="set val = :v",
      ExpressionAttributeValues={
        ':v': new_id
      },
      ReturnValues="UPDATED_NEW"
    )

def get_list_requesting_scores(api, table):
  reply_list = []

  # Get the last mention ID for query. If one doesn't exist get all recent mentions
  last_id = get_last_mention_id(table)
  mentions = api.GetMentions(20, last_id)

  if len(mentions) > 0:
    new_id = mentions[0].id_str
    for mention in mentions:
      if re.search(SEARCH_REGEX, mention.text):
        reply_list.append((mention.user.screen_name,mention.id_str))
    
    update_last_mention_id(table, last_id, new_id)

  return reply_list

def lambda_handler(event, context):
  api = twitter.Api(consumer_key=CONSUMER_KEY,
                  consumer_secret=CONSUMER_SECRET,
                  access_token_key=ACCESS_KEY,
                  access_token_secret=ACCESS_SECRET)

  dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
  state_table = dynamodb.Table('app_state')
  
  reply_list = get_list_requesting_scores(api, state_table)
  
  if len(reply_list) > 0:
    # Get the tweet text from get_nhl_game_brief lambda
    client = boto3.client('lambda', region_name='us-west-2')
    resp = client.invoke(FunctionName='get_nhl_game_brief')
    if resp['StatusCode'] != 200:
      raise AWSLambdaException("Something went wrong getting the tweet text")
    
    lines_json = resp['Payload'].read()
    lines = json.loads(lines_json)
    tweet_body = '\n'.join(lines)

    for user,tweet_id in reply_list:
      print('Replying to {} with tweet ID {}'.format(user, tweet_id))

      api.PostUpdate(tweet_body, None, None, None, tweet_id, True)

    return reply_list
  else:
    return None


if __name__ == '__main__':
  print(lambda_handler(None, None))