GitHubのNewsFeedをTwitterに流すAWS Lambda用スクリプト

flow_to_twitter.py

# coding: utf-8

import sys
import datetime
import re
import requests
import twitter
from datetime import datetime

GITHUB_USER = "hoge"
GITHUB_TOKEN = "hoge"

CONSUMER_KEY = 'fuga'
CONSUMER_SECRET = 'fuga'
ACCESS_TOKEN_KEY = 'fuga'
ACCESS_TOKEN_SECRET = 'fuga'


def to_datetime(s):
    return datetime.strptime(s, '%Y-%m-%dT%H:%M:%SZ')

def download_event():
    return requests.get("https://api.github.com/users/%s/received_events?access_token=%s" % (GITHUB_USER, GITHUB_TOKEN))

def to_item(json):
    created_at = json["created_at"]
    user = json["actor"]["login"]
    repo = json["repo"]["name"]
    event_type = re.compile("Event$").sub('', json["type"])
    content = "%s, %s" % (user, repo)
    url = ""
    short_type = ""

    payload = json["payload"]
    if event_type == "CommitComment":
      short_type = "CommitComment"
      url = payload["comment"]["html_url"]
    elif event_type == "Create":
      short_type = "Create"
      ref_type = payload["ref_type"]
      content += "\n" + ref_type
      url = "https://github.com/" + repo
    elif event_type == "Delete":
      short_type = "Delete"
      ref_type = payload["ref_type"]
      content += "\n" + ref_type
      url = "https://github.com/" + repo
    elif event_type == "Fork":
      short_type = "Fork"
      full_name = payload["forkee"]["full_name"]
      content += "\n" + full_name
      url = payload["forkee"]["html_url"]
    elif event_type == "Gollum":
      short_type = "Gollum"
      url = "https://github.com/%s/wiki" % repo
    elif event_type == "IssueComment":
      short_type = "IssueComment"
      issue_title = payload["issue"]["title"]
      content += "\n\"%s\"" % issue_title
      url = payload["comment"]["html_url"]
    elif event_type == "Issues":
      short_type = "Issues"
      action = payload["action"]
      issue_title = payload["issue"]["title"]
      content += "\n%s \"%s\"" % (action, issue_title)
      url = payload["issue"]["html_url"]
    elif event_type == "Member":
      short_type = "Member"
      action = payload["action"]
      acted_user = payload["member"]["login"]
      content += "\n%s \"%s\"" % (action, acted_user)
      url = "https://github.com/" + repo
    elif event_type == "PullRequest":
      short_type = "PR"
      action = payload["action"]
      title = payload["pull_request"]["title"]
      content += "\n%s \"%s\"" % (action, title)
      url = payload["pull_request"]["html_url"]
    elif event_type == "PullRequestReviewComment":
      short_type = "PRReviewComment"
      action = payload["action"]
      pull_request_title = payload["pull_request"]["title"]
      content += "\n%s \"%s\"" % (action, pull_request_title)
      url = payload["comment"]["html_url"]
    elif event_type == "Push":
      short_type = "Push"
      before = payload["before"][0:10]
      head = payload["head"][0:10]
      url = "https://github.com/%s/compare/%s...%s" % (repo, before, head)
    elif event_type == "Release":
      short_type = "Release"
      action = payload["action"]
      tag_name = payload["release"]["tag_name"]
      content += "\n%s \"%s\"" % (action, tag_name)
      url = payload["release"]["html_url"]
    elif event_type == "TeamAdd":
      short_type = "TeamAdd"
      team_name = payload["team"]["name"]
      content += "\n" + team_name
      url = "https://github.com/" + repo
    elif event_type == "Watch":
      short_type = "Watch"
      action = payload["action"]
      content += "\n" + action
      url = "https://github.com/" + repo

    return { "created_at": created_at, "content": "%s [%s]\n%s" % (created_at, short_type, content), "url": url }


def to_items(jsons):
    return list(reversed(map(to_item, jsons)))


def twitter_client():
    return twitter.Api(consumer_key=CONSUMER_KEY,
                       consumer_secret=CONSUMER_SECRET,
                       access_token_key=ACCESS_TOKEN_KEY,
                       access_token_secret=ACCESS_TOKEN_SECRET)


def is_datetime_string(s):
    try:
        to_datetime(s)
        return True
    except:
        return False


def read_previous_created_at(twitter_client):
    timeline = twitter_client.GetHomeTimeline()
    dt = None
    for t in timeline:
        s = t.text.split(' ')[0]
        if is_datetime_string(s):
            dt = s
            break

    if dt is None:
        return "2000-01-01T00:00:00Z"
    else:
        return dt


def tweet(twitter_client, content):
    try:
        twitter_client.PostUpdate(content)
    except twitter.error.TwitterError as error:
        print(error)


def lambda_handler(event, context):
    tco_length = 23 # t.co length
    lf_length = 2  # \n length
    text_limit_size = 140 - tco_length - lf_length

    res = download_event()
    if res.status_code == 200:
        twitter = twitter_client()
        previous_created_at = read_previous_created_at(twitter)
        items = to_items(res.json())
        new_items = [x for x in items if to_datetime(x["created_at"]) > to_datetime(previous_created_at)]
        for item in new_items:
            text = ""
            if len(item["content"]) > text_limit_size:
                n = len(item["content"]) - text_limit_size
                text = item["content"][0 : len(item["content"]) - n] + "\n" + item["url"]
            else:
                text = item["content"] + "\n" + item["url"]

            tweet(twitter, text)


if __name__ == "__main__":
    lambda_handler(None, None)