よくわからないエンジニア

よく分からないエンジニア(無音鈴鹿)の日々の記録

よくわからないエンジニア

pythonでLINEBOT(バファローズポンタ)を作る

CentOS7 python LINE ポンタ 技術

プロ野球が開幕したので、去年作成したバファローズポンタのツイートをLINEに投稿する機能を改良する事にした。

www.unknownengineer.net

www.unknownengineer.net

目次

目的

以下目的を実現する。

1.バファローズポンタがつぶやくとLINEグループにつぶやきが投稿される
2.つぶやきに画像/動画が添付されていた場合、それも投稿する

上記2点を最低限の目標として作って行きます。言語はpythonで書きます。

1.バファローズポンタがつぶやくとLINEグループにつぶやきが投稿される

まず第一段階だが、ここも細かくは複数の機能から出来ている。

1-1.Twitterのつぶやきを監視する
1-2.Twitterのつぶやきを取得する
1-3.LINEに投稿する

旧ポンタシステムではcronで5分事にシェルスクリプトを実行していたが、TwitterのストリーミングAPIを用いればリアルタイムでタイムラインを追えるっぽい。
尚、6月で廃止になる模様…
<4月10日追記>
6月廃止は延期になりました！

Twitterの“流れるタイムライン”なくなる？ 「User Streams API」が来年6月20日に廃止 - ITmedia NEWS

いきなり出鼻をくじかれそうになりましたが、とりあえず今動くやつを作ります。
ググるとTweepyを使えば簡単に出来そうとの情報を入手したので、Tweepy入れます。

# /usr/local/python3.6/bin/pip3.6 install tweepy
# /usr/local/python3.6/bin/pip3.6 list |grep tweepy
tweepy (3.6.0)

とりあえず、指定したアカウントのつぶやきを標準出力に出すようにします。
この辺りのサイトを参考にしました。

特定のTwitterアカウントを監視してログを吐き出すPythonスクリプト · GitHub

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from tweepy.streaming import StreamListener, Stream
from tweepy.auth import OAuthHandler
from tweepy.api import API
from datetime import timedelta

def get_oauth():
    consumer_key = "***"
    consumer_secret = "***"
    access_key = "***"
    access_secret = "***"
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_key, access_secret)
    return auth

class AbstractedlyListener(StreamListener):
    """ Let's stare abstractedly at the User Streams ! """
    def on_status(self, status):
        if status.author.screen_name == "<screen name>" :
            status.created_at += timedelta(hours=9)
            tweet = "{text}".format(text=status.text)
            print(u"{text}".format(text=status.text))
            print(u"{name}({screen}) {created} via {src}\n".format(
                name=status.author.name, screen=status.author.screen_name,
                created=status.created_at, src=status.source))

if __name__ == '__main__':
    auth = get_oauth()
    stream = Stream(auth, AbstractedlyListener(), secure=True)
    stream.filter(follow=['<Twitter UID>'])

keyとsecretは事前に準備して下さい。screen name,Twitter UIDをとりあえず自分のものを指定しておけば、自分がつぶやく度にprintでなにか表示されると思います。

1-1/1-2はとりあえずこれでクリア。次はLINEに通知します。Line notifyでもメッセージの通知は出来るのですが、画像/動画が送信出来ないので、今回からMessaging APIのFor Developerを使います。
LINE DevelopersでMessaging APIのアカウント作ったりして、アクセストークンと入手して下さい。
後は公式でLine bot用のapi用意されているので、以下コマンドで突っ込みます。

# pip list line-bot-sdk
# pip list |grep line-bot
line-bot-sdk (1.5.0)

上記入れたら、とりあえずテスト的に通知だけするコードは以下です。

from linebot import LineBotApi
from linebot.models import TextSendMessage
from linebot.models import ImageSendMessage
from linebot.models import VideoSendMessage
from linebot.exceptions import LineBotApiError

line_bot_api = LineBotApi('<ACCESS TOKEN>')

try:
    line_bot_api.push_message('<userId>', TextSendMessage(text='Hello World!'))
    line_bot_api.push_message('<userId>', ImageSendMessage(original_content_url='https://****.jpg',preview_image_url='https://***.jpg'))
    line_bot_api.push_message('<userId>', VideoSendMessage(original_content_url='https://**.mp4',preview_image_url='https://***.jpg'))
except LineBotApiError as e:
    print('You can not do this operation!')

上から順番にテキスト/画像/動画を指定のuserIdに送りつけます。
これらを合わせると、とりあえずメッセージを通知するだけのものは作れます。

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from tweepy.streaming import StreamListener, Stream
from tweepy.auth import OAuthHandler
from tweepy.api import API
from datetime import timedelta
from linebot import LineBotApi
from linebot.models import TextSendMessage
from linebot.exceptions import LineBotApiError
from linebot.models import ImageSendMessage
from linebot.models import VideoSendMessage

line_bot_api = LineBotApi('<ACCESS TOKEN>')

def get_oauth():
    consumer_key = "***"
    consumer_secret = "***"
    access_key = "***"
    access_secret = "***"
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_key, access_secret)
    return auth

class AbstractedlyListener(StreamListener):
    """ Let's stare abstractedly at the User Streams ! """
    def on_status(self, status):
        if status.author.screen_name == "<screen name>" :
            status.created_at += timedelta(hours=9)
            tweet = "{text}".format(text=status.text)
            print(u"{text}".format(text=status.text))
            print(u"{name}({screen}) {created} via {src}\n".format(
                name=status.author.name, screen=status.author.screen_name,
                created=status.created_at, src=status.source))
            line_bot_api.push_message('<userId>', TextSendMessage(text=tweet))

if __name__ == '__main__':
    auth = get_oauth()
    stream = Stream(auth, AbstractedlyListener(), secure=True)
    stream.filter(follow=['<Twitter UID>'])

自分のアカウントを指定している場合、適当になにかつぶやくと、ラインにも通知が来るはずです。

2.つぶやきに画像/動画が添付されていた場合、それも投稿する

ここからが今回の新機能で、Twitter側に画像/動画が添付されていたらそちらもLINEに投稿します。
と言ってもこちらも対してやることはなく、動画と画像のURLを取り出すだけです。
動画のURLはextended_entities['media'][0]['video_info']['variants']の video_info[1]['url']で取得出来ます。
gif動画はextended_entities['media'][0]['video_info']['variants']の video_info[0]['url']で取得出来ます。
画像URLはextended_entities['media'][0]['media_url_https']で取得出来ます。
注意点はmedia_urlから引っ張ると、httpのURLが記載されており投稿出来ません。ちゃんとmedia_url_httpsから取得して下さい。

#!/usr/bin/env python
#-*- coding:utf-8 -*-

from tweepy.streaming import StreamListener, Stream
from tweepy.auth import OAuthHandler
from tweepy.api import API
from datetime import timedelta
from linebot import LineBotApi
from linebot.models import TextSendMessage
from linebot.exceptions import LineBotApiError
from linebot.models import ImageSendMessage
from linebot.models import VideoSendMessage

line_bot_api = LineBotApi('<ACCESS TOKEN>')

def get_oauth():
    consumer_key = "***"
    consumer_secret = "***"
    access_key = "***"
    access_secret = "***"
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_key, access_secret)
    return auth

class AbstractedlyListener(StreamListener):
    """ Let's stare abstractedly at the User Streams ! """
    def on_status(self, status):
        if status.author.screen_name == "<screen name>" :
            status.created_at += timedelta(hours=9)
            tweet = "{text}".format(text=status.text)
            print(u"{text}".format(text=status.text))
            print(u"{name}({screen}) {created} via {src}\n".format(
                name=status.author.name, screen=status.author.screen_name,
                created=status.created_at, src=status.source))
            line_bot_api.push_message('<userId>', TextSendMessage(text=tweet))
            try:
                video_info = status.extended_entities['media'][0]['video_info']['variants']
                video_url = video_info[1]['url']
                url=status.extended_entities['media'][0]['media_url_https']
                line_bot_api.push_message('<userId>', VideoSendMessage(original_content_url=video_url,preview_image_url=url))
                try:
                    video_info = status.extended_entities['media'][0]['video_info']['variants']
                    video_url = video_info[0]['url']
                    url=status.extended_entities['media'][0]['media_url_https']
                    line_bot_api.push_message('<userId>', VideoSendMessage(original_content_url=video_url,preview_image_url=url))
                except:
                    try:
                        url=status.extended_entities['media'][0]['media_url_https']
                        line_bot_api.push_message('<userId>', ImageSendMessage(original_content_url=url,preview_image_url=url))
                    except:
                        pass

if __name__ == '__main__':
    auth = get_oauth()
    stream = Stream(auth, AbstractedlyListener(), secure=True)
    stream.filter(follow=['<Twitter UID>'])

上から順番に動画、gif動画、画像を送信するようにtry:exceptで記述しています。

3.一部上手く取得できない動画/画像への対策

expanded_urlにeng.mgで入ってたり、長めのgif動画だとtweepyに渡ってくる情報に動画/画像のURLが入ってない事があります。これだと上記コードでは上手く動きません。そういった場合でもWebサイトのソースの中にはURLが埋め込まれているので、BeautifulSoupを使ってURLを取得します。とりあえずBeautifulSoupをインストールします。

#pip3.6 install beautifulsoup4
Collecting beautifulsoup4
  Downloading beautifulsoup4-4.6.0-py3-none-any.whl (86kB)
    100% |????????????????????????????????| 92kB 3.4MB/s
Installing collected packages: beautifulsoup4
Successfully installed beautifulsoup4-4.6.0

使い方は色んなサイトで解説しているので、そちらに任せます。
とりあえず下手なコードですが、以下でtweepyで取得できない動画/画像のURLを引っ張ってこれました。

target_url = 'https://twitter.com/bs_ponta/media?lang=ja'
r = requests.get(target_url)
soup = BeautifulSoup(r.text, "html.parser")
div = soup.find('div', class_="AdaptiveMediaOuterContainer")
match = div.find("div", class_="PlayableMedia-player ")
if match is None:
    match = div.find("div", class_="AdaptiveMedia-photoContainer js-adaptive-photo ")
    url = match.attrs['data-image-url']
    line_bot_api.push_message('<userId>', ImageSendMessage(original_content_url=url,preview_image_url=url))
else:
    match = match.attrs['style']
    pattern = re.compile("(http.+jpg)")
    urlg = pattern.search(match)
    url = urlg.group(0)
    ulist = url.split("/")
    fname = ulist[len(ulist)-1]
    filename = os.path.splitext(fname)
    fid = filename[0]
    murl = 'https://video.twimg.com/tweet_video/' + fid  + '.mp4'
    line_bot_api.push_message('<userId>', VideoSendMessage(original_content_url=video_murl,preview_image_url=url))

上のコードの末尾にこれをくっつけてとりあえずポンタのTwitterとの連携は完成。自動応答とかはそのうち搭載していきます。

unknownengineer 読者になる

スポンサードリンク

読者になる
コメントを書く