毎日定時に推し画像が送られてくるLINE Bot を作った

はじめに

松岡茉優ファン（通称まゆらー）である僕が、
僕だけの僕だけによる僕だけのためのLINE Bot を作った。

ファンになって5年くらいになると、

→ツイッターのタイムラインで松岡茉優さん関連のツイートを探す
→松岡茉優さんの画像や動画を保存する

この作業に昔ほどの楽しみを覚えなくなり、かつ私自身も社会人になったりして、毎日ツイッターサーフィンする時間も無くなってきます。

全部自動化しちゃおう！！

ってことでその第一弾として、ツイッターで最近人気の画像・動画付きのツイートを検索、画像URL取得、LINE Bot を使って毎日定時に送信する！！！！

よかったら友達追加してください。

twitter API 利用方法・LINE Developers の登録方法等の説明はここでは割愛します。

tweepy で ツイート検索、画像・動画取得

Twitter Developers ページ
https://developer.twitter.com/en/portal/dashboard

tweepy 公式ドキュメント
http://docs.tweepy.org/en/latest/

tweepy 使用時の環境変数の設定

env フォルダにある Yaml ファイルに環境変数を記述しています。

search_tweets.py

import os
import tweepy
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta

# 環境変数のセット
consumer_key = os.getenv('TWITTER_CONSUMER_KEY')
consumer_secret = os.getenv('TWITTER_CONSUMER_SECRET')
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
bearer_token = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')


# twitter で#松岡茉優の画像検索してURL取得する function
def search_tweets():
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)

env/stg.yml

STAGE: stg
LINE_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
LINE_CHANNEL_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
USER_ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_KEY: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_BEAR_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TZ: Asia/Tokyo

ツイート検索と画像・動画の取得

search_tweets.py

    # 昨日の日付を取得
    yesterday = datetime.strftime(datetime.today() - relativedelta(days=1), f"%Y-%m-%d")
    # Twitter検索ワード
    q = f'#松岡茉優 OR 松岡茉優 -松岡茉優似 filter:media exclude:retweets min_faves:10 since:{yesterday} min_retweets:2'
    # 検索
    cric_tweet = tweepy.Cursor(
        api.search, q=q, 
        tweet_mode='extended', # 省略されたツイートを全て取得
        include_entities=True).items(20) # 省略されたリンクを全て取得

    # 画像
    contents = []

    for tweet in cric_tweet:
        print(tweet.full_text)
        try:
            # ツイートの中で、画像・動画のURLが格納されている配列
            media = tweet.extended_entities['media']
            print(media)
            for m in media:
                print(m)
                # LINE Bot で送信する際に必要なプレビュー画面の画像URL
                preview = m['media_url_https']
                # 動画の場合
                if m['type'] == 'video':
                    # 動画URLの取得の際、content_type が video/mp4 でないと、LINE Bot で送信したときに動画が再生されない。
                    # また、なんとか1行で書きたかったので、内包表記を無理矢理使って、最後に[0]でURL を取得しました。
                    origin = [variant['url'] for variant in m['video_info']
                              ['variants'] if variant['content_type'] == 'video/mp4'][0]
                # 画像の場合
                else:
                    # プレビュー画像をクリックしたときに表示される中身の画像のURL
                    origin = m['media_url_https']

                # LINE Bot Messaging API で送信する際に必要な形に整える。
                content = {'preview': preview,
                           'origin': origin, 'type': m['type']}
                contents.append(content)

            print('--------------------------------------------')
        except:
            print('noUrl')
            print('--------------------------------------------')
    return contents

この工程でハマったときのことを記事に書いているのでこちらもぜひ↓

tweepy で 動画 url 取得しようとしてハマった話
https://qiita.com/soma_sekimoto/items/65c664f00573284b0b74

LINE Bot からメッセージを送信

LINE Bot 公式リファレンス
https://developers.line.me/ja/docs/messaging-api/getting-started/

LINE Bot SDK Github
https://github.com/line/line-bot-sdk-python

send_media.py

# -*- coding:utf-8 -*-
import logging
from linebot import LineBotApi

from linebot.exceptions import LineBotApiError
from linebot.models import (
    TextSendMessage, ImageSendMessage, VideoSendMessage
)

import requests
import os


import search_tweets

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def send_media(event, context):

    logger.info("Authentication OK.")
    # LineBotAPIオブジェクトを作成する
    token = os.getenv('LINE_ACCESS_TOKEN')

    line_bot_api = LineBotApi(token)

    try:
        # twitter 検索で画像と動画の URL オブジェクトの配列を取得
        all_media_list = search_tweets.search_tweets()
        print('all_media_list')
        print(all_media_list)

        messages = []

        # tweepy で取得した配列をさらに LINE Bot用に変形
        for media in all_media_list:
            item = VideoSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview']) if media['type'] == 'video' else ImageSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview'])

            messages.append(item)

        print('messages')
        print(messages)
        # 本番環境以外では、自分のみにメッセージが送られてくるようにする。
        if os.getenv('STAGE') == 'prod':
            # なぜか一度に４枚しか送信できないので、二回に分けて送信する。（それでも計７枚しか送れないのは未だに謎）
            line_bot_api.broadcast(
                messages[0:3]
            )
            line_bot_api.broadcast(
                messages[4:8]
            )
        else:
            user_id = os.getenv('USER_ID')
            line_bot_api.push_message(
                user_id,
                messages[0:3]
            )
            line_bot_api.push_message(
                user_id,
                messages[4:8]
            )

    except LineBotApiError as e:
        print(e.status_code)
        print(e.error.message)
        print(e.error.details)

    return {"stautsCode": 200, "body": "OK"}


if __name__ == '__main__':
    send_media(None, None)

Serverless Framework でデプロイ

requirements.py

import os
import sys


requirements = os.path.join(
    os.path.split(__file__)[0],
    '.requirements',
)

if requirements not in sys.path:
    sys.path.append(requirements)

requirements.txt

requests 
print_function
line-bot-sdk # Python で LINE Bot SDK が使える module
tweepy # Python で Twitter API が使える module 

serverless.yml

service: mayu-delivery

provider:
  name: aws
  runtime: python3.7
  region: ap-northeast-1
  stage: stg
  deploymentBucket: sls-deps # デプロイバケットの指定
  environment: ${file(./env/${opt:stage, self:provider.stage}.yml)}

plugins:
  - serverless-python-requirements

custom:
  scheduleEnabled:
    prod: true
    stg: false
    local: false

functions:
  send_media:
    handler: media_deliver.send_media
    timeout: 300
    events:
      - http:
          path: linebot/send_media
          method: post
      - schedule:
          rate: cron(30 3 * * ? *)
          enabled: ${self:custom.scheduleEnabled.${opt:stage, self:provider.stage}}

sls deploy

endpoints:
  POST - https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/linebot/send_media
functions:
  send_media: mayu-delivery-prod-send_media

上記の endpoints のURLを LINE Bot 側にコピペするのを忘れずに。

完成コード

search_tweets.py

import os
import tweepy
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta

consumer_key = os.getenv('TWITTER_CONSUMER_KEY')
consumer_secret = os.getenv('TWITTER_CONSUMER_SECRET')
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
bearer_token = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')


# twitter で#松岡茉優の画像検索してURL取得
def search_tweets():
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)

    yesterday = datetime.strftime(
        datetime.today() - relativedelta(days=1), f"%Y-%m-%d")

    q = f'#松岡茉優 OR 松岡茉優 -松岡茉優似 filter:media exclude:retweets min_faves:10 since:{yesterday} min_retweets:2'

    cric_tweet = tweepy.Cursor(
        api.search, q=q, tweet_mode='extended', include_entities=True).items(20)

    contents = []
    for tweet in cric_tweet:
        print(tweet.full_text)
        try:
            media = tweet.extended_entities['media']
            print(media)
            for m in media:
                print(m)
                preview = m['media_url_https']
                if m['type'] == 'video':
                    origin = [variant['url'] for variant in m['video_info']
                              ['variants'] if variant['content_type'] == 'video/mp4'][0]
                else:
                    origin = m['media_url_https']

                content = {'preview': preview,
                           'origin': origin, 'type': m['type']}
                contents.append(content)

            print('--------------------------------------------')
        except:
            print('noUrl')
            print('--------------------------------------------')
    return contents


if __name__ == "__main__":
    search_tweets()

media_deliver.py

# -*- coding:utf-8 -*-
import logging
from linebot import LineBotApi

from linebot.exceptions import LineBotApiError
from linebot.models import (
    TextSendMessage, ImageSendMessage, VideoSendMessage
)

import requests
import os


import search_tweets

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def send_media(event, context):

    logger.info("Authentication OK.")
    # LineBotAPIオブジェクトを作成する
    token = os.getenv('LINE_ACCESS_TOKEN')

    line_bot_api = LineBotApi(token)

    try:
        # twitter 検索で画像と動画の URL オブジェクトの配列を取得
        all_media_list = search_tweets.search_tweets()
        print('all_media_list')
        print(all_media_list)

        messages = []

        for media in all_media_list:
            item = VideoSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview']) if media['type'] == 'video' else ImageSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview'])

            messages.append(item)

        print('messages')
        print(messages)
        if os.getenv('STAGE') == 'prod':
            line_bot_api.broadcast(
                messages[0:3]
            )
            line_bot_api.broadcast(
                messages[4:8]
            )
        else:
            user_id = os.getenv('USER_ID')
            line_bot_api.push_message(
                user_id,
                messages[0:3]
            )
            line_bot_api.push_message(
                user_id,
                messages[4:8]
            )

    except LineBotApiError as e:
        print(e.status_code)
        print(e.error.message)
        print(e.error.details)

    return {"stautsCode": 200, "body": "OK"}


if __name__ == '__main__':
    send_media(None, None)

requirements.py

import os
import sys


requirements = os.path.join(
    os.path.split(__file__)[0],
    '.requirements',
)

if requirements not in sys.path:
    sys.path.append(requirements)

requirements.txt

requests
line-bot-sdk
print_function
tweepy

serverless.yml

service: mayu-delivery

provider:
  name: aws
  runtime: python3.7
  region: ap-northeast-1
  stage: stg
  deploymentBucket: sls-deps
  environment: ${file(./env/${opt:stage, self:provider.stage}.yml)}

plugins:
  - serverless-python-requirements

custom:
  scheduleEnabled:
    prod: true
    stg: false
    local: false

functions:
  send_media:
    handler: media_deliver.send_media
    timeout: 300
    events:
      - http:
          path: linebot/send_media
          method: post
      - schedule:
          rate: cron(30 3 * * ? *)
          enabled: ${self:custom.scheduleEnabled.${opt:stage, self:provider.stage}}

env/stg.yml

STAGE: stg
LINE_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
LINE_CHANNEL_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
USER_ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_KEY: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_BEAR_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TZ: Asia/Tokyo

おわりに

まだまだ自分の中では物足りないLINE Bot ですが、これからももっと改善・機能追加を続けていきます。

間違っている部分とか、もっとこうした方が良いとかありましたら、コメントよろしくお願いします！！

そして、友達追加もよろしくお願いします！！！！

参考記事