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が必要になります
その処理を①と②で行います

ここら辺は以下の記事がとてもわかりやすいです！

一番分かりやすい OAuth の説明

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で躓くことも多かったですが一応形にできてよかったです

しかしこれで合っているのか自信がないので

ここはこうしたほうが良いなどありましたらコメントお願いします