気ままに実装する機械学習

機械学習に興味のある大学院生によるブログです. 機械学習以外のトピック多めです.

google calendarに予定を追加するTwitter Botを作ってみた

 

今回は、Twitter上でgoogle calendarに予定を追加し、確認できるbotを作成してみたので記事にしてみました。
(ソースは一応githubにも載せておきます。)

使用言語はpythonです。
予定を追加するときはbot
"r year/month/day/hour of start/minutes of start/hour of end/minutes of end"
とリプライすることで登録できるようになっています。
開始時刻終了時刻は共に省略可能です。
開始時刻が省略されている場合は終日の予定、終了時刻だけが省略されている場合は開始時刻から1時間の予定を追加するようにしています。(オプションのrはregister)

予定を確認するときは
"c"
と送るだけで直近の予定を返してくれます。(オプションのcはcheck)

1. bot用のアカウントをTwitter上で登録

2. Twitter APIで各種keyを取得

Twitter App で新規appを作成 (作成したアカウントでログイン)
以下のページに遷移するので適当に入力 (参考までにそれぞれの項目が何を意味するのか列挙しておきます)
・ Name : アプリ名
Website : 自分が持つwebサイトなどのURL
・ Callback URL : OAuthによる認証成功時にリダイレクトされるURL
       (適当なもので良いらしい)
以下のkeyを取得
・ Consumer Key
・ Consumer Secret
Access Token
Access Token Secret

3. Google Calendar APIの登録

Google APIsから登録をします。
(画面通り進むとプロジェクトへの認証情報への追加という画面に遷移するので、一度キャンセルを押します。
認証情報OAuth同意画面タブでユーザに表示するサービス名を適当に設定します。
認証情報 -> 認証情報を作成 -> OAuthクライアントIDを選択します。
クライアントIDの作成その他を選択して名前を適当に入力するとクライアントIDが発行されます。作成されたクライアントIDをクリックするとJSONをダウンロードと出るのでダウンロードします。ダウンロードされたファイルをclient_secret.jsonという名前で保存しておきます。

あとはGoogle APイクライアントをインストールすれば終わりです。

pip install google-api-python-client

4. Pythonソースコードを書く

その前に必要なライブラリをインストール

pip install twitter
  1. 先ほど取得したkeyを入力 (ファイル名 : settings.py)
CONSUMER_KEY = "*********"
CONSUMER_SECRET = "*********"
ACCESS_TOKEN ="*********"
ACCESS_TOKEN_SECRET = "*********"
  1. botが直近の予定をツイートしている部分は以下の通りです。(ファイル名 : tweet.py)
#coding: utf-8

from requests_oauthlib import OAuth1Session
import json
import settings
import random
import datetime
import quickstart
import tweepy


CK = settings.CONSUMER_KEY
CS = settings.CONSUMER_SECRET
AT = settings.ACCESS_TOKEN
AS = settings.ACCESS_TOKEN_SECRET

# create a twitter object
auth = tweepy.OAuthHandler(CK, CS)
auth.set_access_token(AT, AS)
 
api = tweepy.API(auth)

def tweet(userName):
	twitter = OAuth1Session(settings.CONSUMER_KEY,
							settings.CONSUMER_SECRET,
							settings.ACCESS_TOKEN,
							settings.ACCESS_TOKEN_SECRET)

	schedule = quickstart.get_schedule()
	if schedule is not None:
		tweets = [schedule + '!!']
	else:
		tweets = ['No events today!!!']
	randomtweet = tweets[random.randrange(len(tweets))]

	params = {"status": '@' + userName + ' ' + randomtweet + ' ' + str(datetime.datetime.today())} 
	req = twitter.post("https://api.twitter.com/1.1/statuses/update.json", params = params)

リプライを受け取って処理を行っている部分が以下の部分です。 (ファイル名 : auto_reply.py )

#coding:utf-8
import tweepy
import datetime
import settings

import write_schedule as ws
import tweet as tw

CK = settings.CONSUMER_KEY
CS = settings.CONSUMER_SECRET
AT = settings.ACCESS_TOKEN
AS = settings.ACCESS_TOKEN_SECRET

# create a twitter objext
auth = tweepy.OAuthHandler(CK, CS)
auth.set_access_token(AT, AS)
 
api = tweepy.API(auth)
 
class Listener(tweepy.StreamListener):
    def on_status(self, status):
        status.created_at += datetime.timedelta(hours=9)

        userName = 'your user Name'
        botName = 'bot Name'
        gmail = 'gmail address'
        try:
            # if bot recieve reply, it reply or add schedule
            if str(status.in_reply_to_screen_name)== botName and str(status.user.screen_name) == userName:
                reply = str(status.text).split(' ')
                if reply[1] == 'c':
                    tw.tweet(useName)
                elif reply[1] == 'r':
                    day = reply[2]
                    event = reply[3]
                    tweet = '@' + str(status.user.screen_name) + ' I add schedule about ' + event + ' at ' + day + '!\n' + str(datetime.datetime.today())
                    print (tweet)
                    api.update_status(status=tweet)
                    ws.main(day,event,userName,gmail)
                
        except:
            tweet = '@' + userName + ' You should reply : y/m/d/(start h/start m/end h/end m) event'
            api.update_status(status=tweet)

        return True

    def on_error(self, status_code):
        print('Got an error with status code: ' + str(status_code))
        return True
     
    def on_timeout(self):
        print('Timeout...')
        return True
 
listener = Listener()
stream = tweepy.Stream(auth, listener)
stream.userstream()

最後にカレンダーに予定を追加する部分です。(ファイル名 : write_schedule.py)

#coding:utf-8
from __future__ import print_function
import httplib2
import os

from apiclient import discovery
from oauth2client import client
from oauth2client import tools
from oauth2client.file import Storage

import datetime
import tweepy
import settings

CK = settings.CONSUMER_KEY
CS = settings.CONSUMER_SECRET
AT = settings.ACCESS_TOKEN
AS = settings.ACCESS_TOKEN_SECRET

# create a twitter object
auth = tweepy.OAuthHandler(CK, CS)
auth.set_access_token(AT, AS)
 
api = tweepy.API(auth)

try:
    import argparse
    flags = argparse.ArgumentParser(parents=[tools.argparser]).parse_args()
except ImportError:
    flags = None

# If modifying these scopes, delete your previously saved credentials
# at ~/.credentials/calendar-python-quickstart.json
SCOPES = 'https://www.googleapis.com/auth/calendar'
CLIENT_SECRET_FILE = 'client_secret.json'
APPLICATION_NAME = 'Google Calendar API Python Quickstart'


def get_credentials():
    '''Gets valid user credentials from storage.

    If nothing has been stored, or if the stored credentials are invalid,
    the OAuth2 flow is completed to obtain the new credentials.

    Returns:
        Credentials, the obtained credential.
    '''
    home_dir = os.path.expanduser('~')
    credential_dir = os.path.join(home_dir, '.credentials')
    if not os.path.exists(credential_dir):
        os.makedirs(credential_dir)
    credential_path = os.path.join(credential_dir,
                                   'calendar-python-quickstart.json')

    store = Storage(credential_path)
    credentials = store.get()
    if not credentials or credentials.invalid:
        flow = client.flow_from_clientsecrets(CLIENT_SECRET_FILE, SCOPES)
        flow.user_agent = APPLICATION_NAME
        if flags:
            credentials = tools.run_flow(flow, store, flags)
        else: # Needed only for compatibility with Python 2.6
            credentials = tools.run(flow, store)
        print('Storing credentials to ' + credential_path)
    return credentials

def main(day,event,userName,gmail):
    '''Shows basic usage of the Google Calendar API.

    Creates a Google Calendar API service object and outputs a list of the next
    10 events on the user's calendar.
    '''

    credentials = get_credentials()
    http = credentials.authorize(httplib2.Http())
    service = discovery.build('calendar', 'v3', http=http)

    #now = datetime.datetime.utcnow().isoformat() + 'Z' # 'Z' indicates UTC time
    
    try:
        tmp = day.split('/')
        day = datetime.date(int(tmp[0]),int(tmp[1]),int(tmp[2]))
        length = len(tmp)
        if length <= 3:
            allday = True
        else:
            day2 = day
            start = datetime.time(int(tmp[3]),int(tmp[4]),00)
            allday = False
            if length == 7:
               end = datetime.time(int(tmp[5]),int(tmp[6]),00)
            else:
                end = datetime.time(int(tmp[3])+1,int(tmp[4]),00)
        
        if not allday:
            event = {
                'summary': event,
                'start': {
                'dateTime': str(day) + 'T' + str(start),
                'timeZone': 'Asia/Tokyo',
            },
            'end': {
                'dateTime': str(day2) + 'T' + str(end),
                'timeZone': 'Asia/Tokyo',
            },
            'recurrence': [
                'RRULE:FREQ=DAILY;COUNT=1'
            ],
            }
        else:
            event = {
                'summary': event,
                'start': {
                'date' : str(day),
                'timeZone': 'Asia/Tokyo',
            },
            'end': {
                'date' : str(day),
                'timeZone': 'Asia/Tokyo',
            },
            'recurrence': [
                'RRULE:FREQ=DAILY;COUNT=1'
            ],
            }
        
        print(event)
        calendarId = gmail
        event = service.events().insert(calendarId=calendarId, body=event).execute()
    except:
        import traceback
        traceback.print_exc()
        tweet = '@' + userName + ' You should reply : r y/m/d/(start h/start m/end h/end m) event'
        api.update_status(status=tweet)

if __name__ == '__main__':
    main('2018/02/12','sample')

最後に

python auto_reply.py

botが稼働します。
色んなかたが作成したソースコードを寄せ集めて少し修正した程度ですがこれで一応動きます。

5. 動作確認

f:id:linearml:20180224014425p:plain
f:id:linearml:20180224014016p:plain
f:id:linearml:20180224014226p:plain

今後も何か機能増やしたりして、Twitterで全てのことができるようにしたいなあと思ってたりします笑