Twitter API 有料化にともないサービス終了しました
はじめに
こんにちは
できれば福山雅治になりたかったadoth(アドス)と申します
突然ですが、Twitterのアカウント名に日数をぶち込みたいなって思ったことないですか?(私はないです)
そんなときのために
xデー(までorから)の日数をTwitterのアカウント名にぶち込むアプリを作ったのでまとめようかなと
はい、
山本さん(【祝】Twitterの名前をお天気と連動されるアプリを作った時の技術的な紹介【公開した🌥️】)の二番煎じ
煎じすぎてもはやお湯!
ということでやっていきます
概要
Twitterのアカウント名に
(年-月-日)日
を入れて Twitterと連携する を押してログインすると
設定された日付(までorから)の日数に変更されます
変更・更新は日本時間0:00に行われるのでしばしお待ちを
注意事項
- 年、月、日は半角数字で、間には
-
(半角ハイフン)記入してください - 年、月、日の全体を
(
)
(半角カッコ)で囲んでください - 別の日付を設定する際は再度上のフォーマットをアカウント名に入れ直してください
- 利用しない場合は
半角数字 + 日
をアカウント名から消すか、連携を解除してください(連携の解除が確実です) -
半角数字 + 日
がアカウント名に入っている方は前方が更新されます(ごめんなさい)- 例)1日1善@卒業まであと(2019-03-30)日 → 1日1善@卒業まであと30日 → 29日1善@卒業まであと30日
環境
Python 3.6
Twitter API
まぁこれはいろんなサイトにのっているのでググってください
Twitter API 登録 (アカウント申請方法) から承認されるまでの手順まとめ ※2018年9月時点の情報
ポイントとしては後で Callback URL
にAPIのURLを入れるところです
Consumer API Keys
を手に入れたらAWSに移ります
AWS
初めてのAWSなので、この図が合ってるかもわからないけど
今回のシステムはこんな感じ
AWS Lamdba
①、②
他のアカウントのプロフィール変更にはaccess_token
が必要になります
その処理を①と②で行います
ここら辺は以下の記事がとてもわかりやすいです!
Python OAuth認証でTwitter連携/ログインを実装する
1.①のURLにアクセスされるとhttps://api.twitter.com/oauth/authenticate
にouth_tokenをつけてリダイレクトします(Callback URLをAPIのほうに設定)
リダイレクトについては「AWS API-Gateway でLambdaの結果に合わせてリダイレクトする」が参考になります
2.次に許可されるとコールバックURL(②のURL)にouth_token
とouth_verifier
のクリエ文字列が付与されてアクセスされます
3.このouth_token
とouth_verifier
を用いて、access_token
が手に入ります
4.access_token
内のuser_id
、oauth_token
、oauth_token_secret
をデータベースに保存しときます
5.そしてこのouth_token
とouth_token_secret
を用いてプロフィールなどができるようになります
Pythonからデータベースの操作は boto3 を使用
③
このアプリの処理部分
このLambdaはCloudWatchで日本時間の0:00に実行されます
簡単に説明するとこんな感じ
ソースコード
import datetime
import json
import re
import boto3
from dateutil.parser import parse
from requests_oauthlib import OAuth1Session
from setting import CK, CS
USER_SHOW_URL = 'https://api.twitter.com/1.1/users/show.json'
SETTING_URL = 'https://api.twitter.com/1.1/account/settings.json'
UPDATE_PROFILE_URL = 'https://api.twitter.com/1.1/account/update_profile.json'
DB = boto3.resource('dynamodb')
TABLE = DB.Table('twitter_api')
JST = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
XDAY_PATTERN = r'\(\d{1,4}-\d{1,2}-\d{1,2}\)日'
N_DAY_PATTERN = r'\d{1,7}日'
class Account:
def __init__(self, name, db_xday):
self._name = name
self._db_xday = db_xday
def _include_pattern(self, pattern):
match = re.search(pattern, self._name)
if match is None:
return False
return True
def _extract_xday_from_account_name(self):
match = re.search(XDAY_PATTERN, self._name)
xday = match.group().translate(str.maketrans({'(': '', ')': '', '日': ''}))
try:
parse_xday = parse(xday).replace(tzinfo=JST)
except ValueError:
return None
else:
return parse_xday
def retrieve_xday_and_pattern(self):
if self._include_pattern(XDAY_PATTERN):
xday = self._extract_xday_from_account_name()
if xday is None: # e.g. (2019-99-99)日
return None, None
return xday, XDAY_PATTERN
elif self._include_pattern(N_DAY_PATTERN) and self._db_xday != 'None':
xday = parse(self._db_xday).replace(tzinfo=JST)
return xday, N_DAY_PATTERN
else:
return None, None
def replace_pattern_to_remaining_days(self, pattern, remaining_days):
new_account_name = re.sub(pattern=pattern, repl=str(remaining_days) + '日', string=self._name)
return new_account_name
def fetch_now_account_name(request, user_id):
response = request.get(USER_SHOW_URL, params={'user_id': user_id})
if response.status_code == 200:
response = json.loads(response.text)
return response['name']
return None
def update_db(user_id, xday):
TABLE.update_item(
Key={
'user_id': user_id,
},
UpdateExpression='set xday = :xday',
ExpressionAttributeValues={
':xday': xday.isoformat()
},
ReturnValues="UPDATED_NEW"
)
def calculate_remaining_days(xday, *, today=datetime.datetime.now(tz=JST).date()):
try:
remaining_days = abs((xday.date() - today).days)
return remaining_days
except TypeError:
return None
def change_account_name(request, new_account_name): # shorter than 50 characters
response = request.post(UPDATE_PROFILE_URL, params={'name': new_account_name})
def lambda_handler(event, context):
all_data = TABLE.scan()
for account_data in all_data['Items']:
oauth_token = account_data['oauth_token']
oauth_token_secret = account_data['oauth_token_secret']
user_id = account_data['user_id']
db_xday = account_data['xday']
account_request = OAuth1Session(CK, CS, oauth_token, oauth_token_secret)
now_account_name = fetch_now_account_name(account_request, user_id)
if now_account_name is None:
continue
account = Account(now_account_name, db_xday)
xday, pattern = account.retrieve_xday_and_pattern()
if xday is None:
continue
if pattern == XDAY_PATTERN:
update_db(user_id, xday)
remaining_days = calculate_remaining_days(xday)
new_account_name = account.replace_pattern_to_remaining_days(pattern, remaining_days)
change_account_name(account_request, new_account_name)
Amazon API Gateway
API Gatewayはaccess_tokenを取得するため使用しています。
流れはAPI Lambdaで説明したとおりです。
https://api.twitter.com/oauth/authenticate
Callback URL
おわりに
ここまで読んでいただきありがとうございます
初めてのAWSで躓くことも多かったですが一応形にできてよかったです
しかしこれで合っているのか自信がないので
ここはこうしたほうが良いなどありましたらコメントお願いします