Last active
November 22, 2015 00:53
-
-
Save kariyayo/246cb98c40a6e94559e6 to your computer and use it in GitHub Desktop.
GitHubのNewsFeedをTwitterに流すAWS Lambda用スクリプト
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment