できたBOT
メンションで時刻と文を送ると、その時刻にその文を送り返してくれるTwitterBOT
LINEのリマインくん的な（便利ですよねあれ。
仕組み
2つ以上のリマインドを作れるようにしたいのでStream関数は使えない。
→ １０分の頻度で送られたメンションを記録する
メンションで指定した時刻が現在時刻よりも前だった場合にリプライを返す。
→ ２回目以降の実行は前回実行からの１０分間の中に指定時刻があった場合リプライを返すようになる。
リプライを返したらファボをつける。
ファボが付いたメンションにはリプライを送らないようにする。
１０分毎の時間設定しかできないようにしている。
(小さくするとAPIのリクエストが多すぎてエラーを起こす可能性があるため)
ちなみに自分用なのでメンションのフォーマット部分は手を抜いてがちがちにしてある。
ソースコード
GitHub
各関数の説明
設定部分
# 実行する頻度 分単位
do_minute = 10
# 書き方が悪い時のエラーメッセージ
ErrorText_makejson = "文の構成がオカシイです\n固定ツイートを確認してください。"
# 時間の設定が悪い時のエラーメッセージ
ErrorText_Time = "時間の設定がオカシイです\n固定ツイートを確認してください。"
# ツイート間の休止時間 秒単位 小さすぎるとAPIエラーになる
sleep_time = 5
make_json()
make_json()
def make_json(): # リプライの情報をJSONファイルに出力
mention_data = []
mention_dict = {}
mention_list = api.mentions_timeline(count=200) # メンションを２００個取得
for mention in mention_list:
if mention.user.screen_name in follow_list: # フォローしていない人を除く
if mention.favorited == 0: # ファボしてあるモノは除く
Tweet_time = mention.created_at.astimezone(
timezone('Asia/Tokyo')) # ツイートされた時間を UTC → JST に変換
Tweet_time = datetime.datetime.strftime(
Tweet_time, '%Y-%m-%d %H:%M') # フォーマットを変換
Tweet_text = mention.text
try:
spilit_text = Tweet_text.split('\n', maxsplit=2) #指定時間と文を分ける
time = spilit_text[1]
text = spilit_text[2]
except:
print("エラー")
time = 1
text = ErrorText_makejson
# ツイートID,ツイートされた時間,本文,ユーザーIDを取得
mention_dict = dict(
Tweet_id=mention.id, # ツイートID
user_id=mention.user.screen_name, # ユーザーID
Tweet_time=str(Tweet_time), # ツイートされた時間（文字列に変換)
Time=time, # 指定時間
Text=text, # 返送する文
)
mention_data.append(mention_dict)
json.dumps(mention_data)
with open('mention_data.json', 'w') as f: # ← ローカル環境時
# with open('/tmp/mention_data.json', 'w') as f: # ← GCP上で起動時
json.dump(mention_data, f, ensure_ascii=False, indent=2)
print(json.dumps(mention_data, ensure_ascii=False, indent=2)) #作ったjsonファイルを出力
メンションを最大200まで取得してその情報をjsonファイルに出力する
メンション形式がフォーマットにあってない場合はTimeに1、分にエラー文を返す。
実行すると、こんなjsonファイルが作られる。
[
{
"Tweet_id": 160000000000,
"user_id": "Al***",
"Tweet_time": "2023-01-01 12:43",
"Time": "12:40",
"Text": "てすと"
}
]
get_strtime_conversion(str)
get_strtime_conversion(str)
def get_strtime_conversion(str_time):
time_pattern_list = [
r'^[0-9]{4}/[0-9]{2}/[0-9]{2} [0-9]{2}:[0-9]{2}', # 2022/12/29 03:03
r'^[0-9]{2}/[0-9]{2} [0-9]{2}:[0-9]{2}', # 12/29 03:03
r'^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}', # 2022-12-29 03:03
r'^[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}', # 12-29 03:03
r'^[0-9]{2}:[0-9]{2}', # 03:03
]
datetime_pattern = [
'%Y/%m/%d %H:%M',
'%m/%d %H:%M',
'%Y-%m-%d %H:%M',
'%m-%d %H:%M',
'%H:%M',
]
time = 0
Today_year = now_time.year
Today_month = now_time.month
Today_day = now_time.day
try:
for i, time_pattern in enumerate(time_pattern_list):
if re.compile(time_pattern).search(str_time):
time = datetime.datetime.strptime(
str_time, datetime_pattern[i])
if i == 0:
time = datetime.datetime.strftime(
time, '%Y-%m-%d %H:%M')
if i == 1:
time = datetime.datetime.strftime(
time, f'{Today_year}-%m-%d %H:%M')
if i == 2:
time = datetime.datetime.strftime(
time, '%Y-%m-%d %H:%M')
if i == 3:
time = datetime.datetime.strftime(
time, f'{Today_year}-%m-%d %H:%M')
if i == 4:
time = datetime.datetime.strftime(
time, f'{Today_year}-{Today_month}-{Today_day} %H:%M')
break
if time == 0: # どのフォーマットにも当てはまらなかったとき
return 0
time = datetime.datetime.strptime(
time, '%Y-%m-%d %H:%M').astimezone(tz.gettz('Asia/Tokyo'))
# time = time - datetime.timedelta(hours=9) #GCP上で起動時に有効にする 時差が発生するため??
return time
except:
time = 0
return time
jsonファイルからの指定の時刻(str型)を与える。
正規表現を使って5パターンの時刻を指定するフォーマットを統一してdatetime型にして返す。
どのフォーマットにも当てはまらない場合はtime=0を返す。
reply()
reply()
#エラーで時間設定ができていない時は代わりに古い時間を読み込む
old_time = datetime.datetime.strptime(
'2000-1-1 00:00', '%Y-%m-%d %H:%M').astimezone(tz.gettz('Asia/Tokyo'))
json_open = open('mention_data.json', 'r') # ← ローカル上で起動時
# json_open = open('/tmp/mention_data.json', 'r') # ← GCP上で起動時
json_load = json.load(json_open)
for mention_data in json_load:
Tweet_id = mention_data['Tweet_id']
user_id = mention_data['user_id']
Time = mention_data['Time']
Text = mention_data['Text']
if Time == 1: # json作成時にエラーが起きていた時
Time = old_time
else:
Time = get_strtime_conversion(Time)
if Time == 0: # 時間設定でエラーが起きたとき
Time = old_time
if not Text == ErrorText_makejson:
Text = ErrorText_Time
else :
#実行する間隔より前の時刻を設定していた際にエラーを返す
if now_time - datetime.timedelta(minutes=do_minute) > Time:
Time = old_time
if not Text == ErrorText_makejson:
Text = ErrorText_Time
if Time <= now_time:
# リプを送る
reply_text = '@' + user_id + '\n' + Text
api.update_status(
reply_text, in_reply_to_status_id=Tweet_id)
# ファボをする
api.create_favorite(Tweet_id)
print("\nツイートしました\n")
time.sleep(sleep_time)
実際にリプライを返す関数。
指定時刻が起動時刻の起動間隔時間(10分)よりも前の時間だった場合にエラー文をリプライ文に設定する。
ここまででエラーが起きた場合にはtime = 0,1となっており、それに合わせたエラー文をリプライ文に設定
その後timeには過去の時刻を設定する。(この実行時にリプライを返す)
ここまでで指定時刻とリプライ文を調整したので、調整した時刻が現在時刻より前だった場合にリプライ＆ファボを行う。
ハマったとこ
datetimeのタイムゾーン周りに苦戦しました。
ローカル環境だと問題なく動作するのに、GCP上だと一切エラー出てないのに動作しない...
原因がタイムゾーンの設定をしてないから。
if文の条件部分が、日本時とアメリカ時で比較し続けてました...
そして色々付け加えているため、そのあたりがゴチャゴチャしちゃってます。