[PYTHON] I made a LINE Bot that sends recommended images every day on time
Introduction
I, a fan of Mayu Matsuoka (commonly known as Mayura), I made a LINE Bot just for me, just for me.
After about 5 years as a fan
→ Search for tweets related to Mayu Matsuoka on the Twitter timeline. → Save Mayu Matsuoka's images and videos
I don't enjoy this work as much as I used to, and I myself become a member of society, so I don't have time to surf Twitter every day.
** Let's automate everything! !! ** **
So, as the first step, search for tweets with images and videos that are popular recently on Twitter, get the image URL, and send it on time every day using LINE Bot! !! !! !!
Please add friends if you like.
The explanation of how to use twitter API and how to register LINE Developers is omitted here.
Tweet search, image / video acquisition with tweepy
Twitter Developers page https://developer.twitter.com/en/portal/dashboard
tweepy official documentation http://docs.tweepy.org/en/latest/
Setting environment variables when using tweepy
The environment variables are described in the Yaml file in the env folder.
search_tweets.py
import os
import tweepy
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta
#Set of environment variables
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')
#on twitter#Image search for Mayu Matsuoka and get 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
Tweet search and image / video acquisition
search_tweets.py
#Get yesterday's date
yesterday = datetime.strftime(datetime.today() - relativedelta(days=1), f"%Y-%m-%d")
#Twitter search word
q = f'#Mayu Matsuoka OR Mayu Matsuoka-Mayu Matsuoka similar filter:media exclude:retweets min_faves:10 since:{yesterday} min_retweets:2'
#Search
cric_tweet = tweepy.Cursor(
api.search, q=q,
tweet_mode='extended', #Get all omitted tweets
include_entities=True).items(20) #Get all omitted links
#image
contents = []
for tweet in cric_tweet:
print(tweet.full_text)
try:
#An array that stores URLs of images and videos in tweets
media = tweet.extended_entities['media']
print(media)
for m in media:
print(m)
#Image URL of the preview screen required when sending with LINE Bot
preview = m['media_url_https']
#For video
if m['type'] == 'video':
#When getting the video URL, content_type is video/If it is not mp4, the video will not play when sent by LINE Bot.
#Also, I wanted to write it in one line, so I used the comprehension notation forcibly and finally[0]I got the URL in.
origin = [variant['url'] for variant in m['video_info']
['variants'] if variant['content_type'] == 'video/mp4'][0]
#For images
else:
#URL of the content image displayed when the preview image is clicked
origin = m['media_url_https']
#Arrange it in the required form when sending with the LINE Bot Messaging API.
content = {'preview': preview,
'origin': origin, 'type': m['type']}
contents.append(content)
print('--------------------------------------------')
except:
print('noUrl')
print('--------------------------------------------')
return contents
I wrote in the article about when I was addicted to this process, so please do not miss it ↓
A story I was addicted to trying to get a video url with tweepy https://qiita.com/soma_sekimoto/items/65c664f00573284b0b74
Send a message from LINE Bot
LINE Bot Official Reference 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.")
#Create a LineBot API object
token = os.getenv('LINE_ACCESS_TOKEN')
line_bot_api = LineBotApi(token)
try:
#Get an array of image and video URL objects with twitter search
all_media_list = search_tweets.search_tweets()
print('all_media_list')
print(all_media_list)
messages = []
#Further transform the array obtained by tweepy for 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)
#Be sure to send messages only to yourself, except in a production environment.
if os.getenv('STAGE') == 'prod':
#For some reason, I can only send 4 images at a time, so I will send them in two parts. (Still, it is still a mystery that only 7 sheets can be sent in total)
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)
Deploy with 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 #Module that can use LINE Bot SDK in Python
tweepy #Module that can use Twitter API in Python
serverless.yml
service: mayu-delivery
provider:
name: aws
runtime: python3.7
region: ap-northeast-1
stage: stg
deploymentBucket: sls-deps #Specifying a deployment bucket
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
Don't forget to copy the URL of the above endpoints to the LINE Bot side.
Completion code
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')
#on twitter#Image search for Mayu Matsuoka and get 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'#Mayu Matsuoka OR Mayu Matsuoka-Mayu Matsuoka similar 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.")
#Create a LineBot API object
token = os.getenv('LINE_ACCESS_TOKEN')
line_bot_api = LineBotApi(token)
try:
#Get an array of image and video URL objects with twitter search
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
in conclusion
The LINE Bot is still unsatisfactory for me, but I will continue to improve and add functions.
If you find something wrong or something more like this, please leave a comment! !!
And please add friends! !! !! !!
Reference article
https://blog.serverworks.co.jp/sls-line-beginner
Recommended Posts