In both the browser version and the app version, you can only see the trend of Twitter top 30. It's a hassle to stick to the trend page all the time, and it's hard to see trends in __any region __.
Therefore, let's create a Twitter bot that acquires trends using the Twitter API and notifies when any hashtag is in the trend. You can use the Twitter API to get the top 50 trends for any region __. Maybe it can be operated for free. I think it works just by rewriting a part of the code covered in the article.
If you just want to try it locally, please skip it.
Please make it.
Create a directory to save your bot's files.
Save all the files created below here.
Here, it is vtuber-twitter-bot. In the end, it looks like this.
vtuber-twitter-bot
├── dics.py
├── tools.py
├── main.py
├── index.py
├── runtime.txt
├── requirements.txt
├── Procfile
heroku This article is detailed. Let's register and install the CLI. Since it operates for free, you do not need to register your credit card.
Python 3.x I will write the bot code in Python 3.x. The library to be installed is as follows. Please let me know if there are any omissions. Maybe everything can be installed with pip.
Twitter API This article is detailed. Make a note of the API key, Consumer_secret, Access_token, Access_secret.
-Summary of steps from Twitter API registration (account application method) to approval
Run the code below.
You will see a list of cities where you can get trends.
Find the city you want to get by relying on country (country name) and name (city name), and make a note of the woeid.
import tweepy
def twitter_api():
CONSUMER_KEY = 'YOUR API KEY'
CONSUMER_SECRET = 'YOUR CONSUMER SECRET'
ACCESS_TOKEN = 'YOUR ACCESS TOKEN'
ACCESS_SECRET = 'YOUR ACCESS SECRET'
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
api = tweepy.API(auth)
return api
api = twitter_api()
api.trends_available()
#>>[{'country': '',
#>> 'countryCode': None,
#>> 'name': 'Worldwide',
#>> 'parentid': 0,
#>> 'placeType': {'code': 19, 'name': 'Supername'},
#>> 'url': 'http://where.yahooapis.com/v1/place/1',
#>> 'woeid': 1},
#>> {'country': 'Canada',
#>> 'countryCode': 'CA',
#>> 'name': 'Winnipeg',
#>> 'parentid': 23424775,
#>> 'placeType': {'code': 7, 'name': 'Town'},
#>> 'url': 'http://where.yahooapis.com/v1/place/2972',
#>> 'woeid': 2972},
#>> .
#>> .
#>> .
I think the following woeid is enough to get the world trend, Japan trend, and trend of major cities in Japan. If there is a leak, please look for it by the above method.
woeid_dic = {'world': 1, 'Japan': 23424856,
'Tokyo': 1118370, 'Kyoto': 15015372, 'Osaka': 15015370,
'Sapporo': 1118108, 'Sendai': 1118129, 'Nagoya': 1117817,
'Kobe': 1117545, 'Hiroshima': 1117227, 'Fukuoka': 1117099,
'Saitama': 1116753, 'Chiba': 1117034, 'Yokohama': 1118550,
'Kawasaki': 1117502, 'Sagamihara': 1118072, 'Kitakyushu': 1110809,
'Okayama': 90036018, 'Niigata': 1117881, 'Takamatsu': 1118285,
'Hamamatsu': 1117155, 'Kumamoto': 1117605, 'Okinawa': 2345896}
If you refrain from woeid, it is finally time to get the trend.
city = 'Tokyo'
woeid = woeid_dic[city]
trends = api.trends_place(woeid)
#Print only trend ranking and content
for trend in trends:
trend_l = 0
for trend_id in trend['trends']:
trend_n = trend_id['name']
trend_l += 1
print(city, trend_l, trend_n)
In the following, as an example, we will create a bot that notifies you when a suitable hashtag enters the trend. The design is as follows. Please change the acquisition frequency and notification frequency as appropriate. __ At that time, be aware of API limit. __ (GET trends / place is 75 requests / 15 minutes)
--Get trends every 5 minutes --Notify every 15 minutes --Cancel if the ranking is lower than the past notification
If you rewrite only dics.py, it should be your favorite bot. [](
If you change woeid_dic, also change now_trend, tmp_record, recent_trends in tools.py. )
For your understanding, we recommend that you try running each code locally.
Be careful when running locally, as os.environ ['ENV_NAME'] is the code that receives the value of the environment variable ENV_NAME.
dics.py
dics.py
#A list of hashtags you want to be notified of if you're on a trend
check_trend = ['#hashtag', '#trend']
#Woeid of the region to get the trend
woeid_dic = {'world': 1, 'Japan': 23424856}
tools.py
tools.py
import pickle
import os
import tweepy
from dics import check_trend, woeid_dic
#Preparing to tweet and retrieve data on Twitter
def twitter_api():
CONSUMER_KEY = os.environ['API_KEY']
CONSUMER_SECRET = os.environ['API_SECRET_KEY']
ACCESS_TOKEN = os.environ['ACCESS_TOKEN']
ACCESS_SECRET = os.environ['ACCESS_TOKEN_SECRET']
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
api = tweepy.API(auth)
return api
def twitter_trend_notification(tw_switch):
api = twitter_api()
trend_log = './trend_log.pickle'
now_trend = {}
tmp_record = {}
recent_trends = {}
for city in woeid_dic.keys():
now_trend.update({city: {}})
tmp_record.update({city: {}})
recent_trends.update({city: {}})
#Get trends and extract regions / rankings / trend words
for city, woeid in woeid_dic.items():
trends = api.trends_place(woeid)
for trend in trends:
trend_l = 0
for trend_id in trend['trends']:
trend_l += 1
trend_name = trend_id['name']
if trend_name in check_trend:
print(city, trend_l, trend_name)
now_trend[city].update({trend_name: trend_l})
#Extract only the words you want to notify from the acquired trends
for city in now_trend.keys():
for c_trend in check_trend:
if c_trend in now_trend[city].keys():
in_dic_tmp = {c_trend: now_trend[city][c_trend]}
tmp_record[city].update(in_dic_tmp)
if not os.path.exists(trend_log):
with open(trend_log, 'wb') as pi:
pickle.dump(tmp_record, pi)
with open(trend_log, 'rb') as pi:
old_record = pickle.load(pi)
#print(tmp_record)
#print(old_record)
#Update if the ranking is higher than the past record
#Add if there is a trend that is not in the past record
new_record = old_record
for city in now_trend.keys():
for t_trend in tmp_record[city].keys():
if t_trend in old_record[city].keys():
if tmp_record[city][t_trend] < old_record[city][t_trend]:
new_record[city][t_trend] = tmp_record[city][t_trend]
else:
in_dic_tmp = {t_trend: tmp_record[city][t_trend]}
new_record[city].update(in_dic_tmp)
with open(trend_log, 'wb') as pi:
pickle.dump(new_record, pi)
#if new_record != {'world': {}, 'Japan': {}}:#, 'Tokyo': {}}:
# print('trend : ', new_record)
#Extract the ranking of 10 past notifications
recent_tweets = api.user_timeline(count=10, tweet_mode='extended')
#recent_tweets = [tweet.full_text for tweet in recent_tweets]
for tweet in recent_tweets:
sp_tw = tweet.full_text.split('\n')
if '[Trend notification]' in sp_tw[0]:
hashtag = '#'+ tweet.entities['hashtags'][0]['text']
for key_city in recent_trends.keys():
try:
re_lev = sp_tw[2].replace('In the last 15 minutes, "{}"But{}Trends'.format(hashtag, key_city), '').replace('It seems that it has reached the rank.', '')
re_lev = int(re_lev)
try:
if recent_trends[key_city][hashtag] > re_lev:
recent_trends[key_city][hashtag] = re_lev
except:
recent_trends[key_city].update({hashtag: re_lev})
except:
pass
#When you run a notification, tweet if it ranks higher than past notifications
if tw_switch:
for city in new_record.keys():
for trend_name in new_record[city].keys():
if not trend_name in recent_trends[city].keys():
tw_flag = 1
elif recent_trends[city][trend_name] > new_record[city][trend_name]:
tw_flag = 1
else:
tw_flag = 0
if tw_flag and (trend_name in check_trend):
tw_str = '[Trend notification]\n\n'
#tw_str = '[Trend notification] During trial operation\n\n'
tw_str += 'In the last 15 minutes, "{}"But{}Trends{}It seems that it has reached the rank.'.format(trend_name, city, new_record[city][trend_name])
print(tw_str)
api.update_status(tw_str)
#Erase the record after tweeting
os.remove(trend_log)
main.py
main.py
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
from tools import twitetr_trend_notification
#Don't tweet
@sched.scheduled_job('cron', minute='5,10,20,25,35,40,50,55', hour='*')
def main_wo_tw():
try:
tw_switch = 0
twitter_trend_notification(tw_switch)
except Exception as e:
print('ERROR on twitter_trend_notification')
print(e)
#Tweet
@sched.scheduled_job('cron', minute='0,15,30,45', hour='*')
def main_w_tw():
try:
tw_switch = 1
twitter_trend_notification(tw_switch)
except Exception as e:
print('ERROR on twitter_trend_notification')
print(e)
if __name__ == '__main__':
sched.start()
index.py
#empty. I feel like I don't need it.
Create runtime.txt, requirements.txt, Procfile.
runtime.txt
python-3.6.2
requirements.txt
tweepy==3.6.0
APScheduler==3.0.3
Procfile
web: python index.py
clock: python main.py
app-name is a good idea to name the directory where you want to save your code.
The following commands should be executed in this directory.
heroku login
heroku create app-name
You can set environment variables with heroku config: set ENV_NAME =" VALUE ". The last four are executed as they are.
app-name is the one from earlier.
appname=app-name
heroku config:set ACCESS_TOKEN="YOUR TWITTER ACCESS TOKEN" --app $appname
heroku config:set ACCESS_TOKEN_SECRET="YOUR TWITTER ACCESS TOKEN SECRET" --app $appname
heroku config:set API_KEY="YOUR TWITTER API KEY" --app $appname
heroku config:set API_SECRET_KEY="YOUR TWITTER API SECRET KEY" --app $appname
heroku config:set TZ="Asia/Tokyo" --app $appname
app-name is the one from earlier.
appname=app-name
git init
git add .
git commit -m "new commit"
git remote add heroku https://git.heroku.com/$appname.git
git push heroku master
You can see the printed logs by running the command heroku logs -t in the app-name directory.
For debugging, it is easier to put the following shell script in the directory one level above the app-name directory.
After modifying the code, execute ../heroku_deploy.sh in the app-name directory. If it fails, set it to executable with chmod u + x ../heroku_deploy.sh.
heroku_deploy.sh
git add .
git commit -m "new commit"
git push heroku master
heroku ps:scale web=0 clock=1
heroku ps
heroku logs --tail
Recommended Posts