Go to Qiita Advent Calendar 2024 Top
LoginSignup
3
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 5 years have passed since last update.

@Hisan_twi

PythonとSQLiteでRaspberry Piから定期ツィートするbotを作る

Posted at

#はじめに
前回の記事で定期ツィートをデータベースファイルに登録するアプリを作りました。
今回はそのデータベースファイルを使用して、Raspberry Piをサーバーとした定期ツィートbotを作成します。

#環境

  • Raspberry Pi 3 Model B+
  • OSX: 10.14.6 (メインPC)
  • python: 3.7.4

使用ライブラリ:

  • SQLite3
  • Twython

使用ソフトウェア:

  • Samba

事前にRaspberry PiにTwythonをインストールしておく必要があります。

pip3 install twython

####前提条件
Raspberry PiはOSとPython3がインストールされており、メインPCとSSH接続が行える設定、および、セキュリティ設定が完了しているものとします。

また、メインPCでは前回の記事で作成したアプリにより定期ツィートの登録を行ったデータベースファイルが既にあるものとします。

#botの作成
##Twitter APIキーの取得
まず、Twitter APIに登録申請してAPIキーを取得する必要があります。
(申請のやり方はhttps://www.torikun.com/entry/twitter-developer-api/ 等を参照してください。）

APIキーを取得したらauth.pyというファイルを作成して４つのキーを記述します。
(''には自分のアカウントで取得したAPIキーを入れてください。）

consumer_key = ''
consumer_secret = ''
access_token = ''
access_token_secret = ''

##ツィートを行うPythonプログラム

ツィートを投稿するプログラムのコードを以下に示します。

twitter-bot.py
import sqlite3
import unicodedata
import os
import sys
import datetime
from logging import getLogger, StreamHandler, FileHandler, DEBUG, Formatter

from twython import Twython, TwythonError
from auth import (
    consumer_key,
    consumer_secret,
    access_token,
    access_token_secret
)

twitter = Twython(
    consumer_key,
    consumer_secret,
    access_token,
    access_token_secret
)

dbname = "tweet-db.db"          #データベースファイルの名前
tablename = "tweettable"        #テーブルの名前
savefilename = "lasttweet.txt"      #設定ファイルの名前
logfilename = "tweetlog.txt"        #ログファイルの名前

setting_hours = datetime.timedelta(hours = 5)       #何時間置きにツィートするかを指定する
time_reduntance = datetime.timedelta(minutes = 5)   #時間余裕の設定
aborttime_start = 1     #ツィート停止開始時刻
aborttime_end = 5       #ツィート停止終了時刻

#ログ出力の設定
logger = getLogger(__name__)
handler1 = StreamHandler()
handler1.setFormatter(Formatter("%(asctime)s %(levelname)8s %(message)s"))
handler2 = FileHandler(logfilename)
handler2.setLevel(DEBUG)
handler2.setFormatter(Formatter("%(asctime)s %(levelname)8s %(message)s"))
logger.setLevel(DEBUG)
logger.addHandler(handler1)
logger.addHandler(handler2)
logger.propagate = False

def main():
    try:
        textid = getID()
        text, imgpath = callText(textid)
    except sqlite3.Error:
        sys.exit(0)
    else:
        doTweet(text, imgpath)
        saveTweet(textid, text, imgpath)

#テキストのIDを指定する関数
def getID():
    erroralert = False
    now_time = datetime.datetime.now()
    if not (aborttime_start <= now_time.hour <= aborttime_end): #ツィート停止期間でないかどうかの判定
        if os.path.exists(savefilename):
            with open(savefilename, mode='r') as f:
                l = f.readlines()
            last_tweet_time = datetime.datetime.fromisoformat(l[1].rstrip())
            td = now_time - last_tweet_time
            if td >= setting_hours - time_reduntance:           #設定した時間を経過しているかどうかの判定
                try:
                    conn = sqlite3.connect(dbname)
                    cur = conn.cursor()
                    cur.execute(u"select max(id) from {0}".format(tablename))   #最大のIDを確認する
                    tupletext = cur.fetchone()
                    maxid = tupletext[0]
                    if int(l[0]) >= maxid:
                        cur.execute(u"select min(id) from {0}".format(tablename))   #最小のIDを選ぶ
                        tupletext = cur.fetchone()
                        textid = str(tupletext[0])
                    else:
                        cur.execute(u"select min(id) from {0} where id > '{1}'".format(tablename, l[0]))  #前回のツィートのIDの次のIDを選ぶ
                        tupletext = cur.fetchone()
                        textid = str(tupletext[0])
                except sqlite3.Error as e:
                    logger.error(e.message)
                    erroralert = True
                    pass
                finally:
                    conn.close()
            else:
                sys.exit(0)     #設定した時間が経過していなければ何もしない
        else:
            try:
                conn = sqlite3.connect(dbname)
                cur = conn.cursor()
                cur.execute(u"select min(id) from {0}".format(tablename))   #最小のIDを選ぶ
                tupletext = cur.fetchone()
                textid = str(tupletext[0])
            except sqlite3.Error as e:
                logger.error(e.message)
                erroralert = True
                textid = "1"
                pass
            finally:
                conn.close()
    else:
        sys.exit(0)             #ツィート停止期間内であれば何もしない
    if erroralert == True:
        raise sqlite3.Error     #エラーフラグが立っていたら改めてエラーを出す
    else:
        return textid

#ツィートを行うメソッド
def doTweet(text, imgpath):
    if imgpath is None:
        try:
            twitter.update_status(status=text) 
            logger.info("Tweetしました: \n{0}\n添付画像: 無し".format(text))
        except TwythonError as e:
            logger.error(e.msg)
            sys.exit(1)
    else:
        try:
            image = open(imgpath, "rb")
            response = twitter.upload_media(media=image)
            twitter.update_status(status=text,  media_ids=[response['media_id']])
            logger.info("Tweetしました: \n{0}\n添付画像: {1}".format(text, imgpath))
        except TwythonError as te:
            logger.error(te.msg)
            sys.exit(1)
        except FileNotFoundError as fe:
            logger.error(fe.msg)
            sys.exit(1)

#データベースから指定したIDのテキストと画像アドレスを呼び出す関数
def callText(textid):
    erroralert = False
    conn = sqlite3.connect(dbname)
    cur = conn.cursor()
    try:
        cur.execute(u"select * from {0} where id = '{1}'".format(tablename, textid))    #指定したIDのテキストを呼び出す
        tupletext = cur.fetchone()
        text = tupletext[2]
        imgpath = tupletext[3]
    except sqlite3.Error:
        erroralert = True               #エラーフラグを立てる
        pass                            #一旦エラーをパス
    finally:
        conn.close()
    if erroralert == True:
        raise sqlite3.Error      #エラーフラグが立っていたら改めてエラーを出す
    else:
        return text, imgpath

#ログを出力する
def saveTweet(textid, text, imgpath):
    dt_now = datetime.datetime.now()    #現在の日時をdatetimeオブフェクトとして取得
    dt_now_str = dt_now.isoformat()     #datetimeオブフェクトを文字列に変換
    if imgpath is None: imgpath = str("None")
    l = [textid, dt_now_str, text, imgpath]
    with open(savefilename, mode='w') as f:
        f.write('\n'.join(l))

if __name__ == '__main__':
    main()

何時間置きにツィートするかを指定するsetting_hoursと、ツィート停止開始時刻aborttime_startおよびツィート停止終了時刻aborttime_endは自分の好みに合わせて変更してください。
（上記のコードでは5時間置きにツィート、1時から5時までツィート停止という設定になっています。）

##メインPCからRaspberry Piへのファイル転送
作成したファイルをメインPCからRaspberry Piへ転送します。

PCからRaspberry Piへのファイル転送はSambaを使用するのが楽です。

まず、Raspberry Pi側でSambaの設定を行います。

Raspberry PiにSambaをインストールします。

sudo apt-get install samba

次に、以下のコマンドからSambaの設定ファイルを編集します。

sudo nano /etc/samba/smb.conf

設定ファイルの末尾に以下の記述を追記します。
(ここではユーザー名をpiとしています。ユーザー名を変更している場合は自分のユーザー名に置き換えてください。)

[raspberrypi]
   comment = Users home
   path = /home/pi
   guest ok = no
   read only = no
   writable = yes
   browsable = yes
   valid user = pi
   create mask = 0664
   directry mask = 0775

Sambaへのユーザー登録を行います。以下のコマンドを実行すると、パスワードの新規登録も求められます。

sudo smbpasswd -a pi

ユーザー登録を行なったらSambaを再起動します。

sudo service smbd restart

これでRaspberry Pi側でSambaを使用する準備ができました。

次に、メインPCからRaspberry Piへの接続を行います。
OSXではFinderの「移動」メニューから「サーバへ接続...」を選択し、接続先をsmb://raspberrypi.localとすれば接続できます。詳しい接続方法、および、Windowsでの接続方法についてはhttps://tool-lab.com/make/raspberrypi-startup-14/ 等を参照してください。

上述のauth.pyとtwitter-bot.py、および定期ツィートを登録したデータベースファイルtweet-db.dbと登録した画像ファイルを内包する画像フォルダをTweet_botフォルダにまとめ、メインPCからRaspberry Piにコピーします。
Raspberry Piにコピーしたフォルダは以下のような構成となっています。

フォルダ構成
/home/pi/Twitter_bot
├── auth.py
├── tweet-bot.py
├── tweet-db.db
└── 画像
    └── 画像フォルダ
        └── 画像ファイル

##crontabの設定
作成したtweet-bot.pyを定期実行することで登録した内容を定期ツィートすることができます。
定期実行はOSの機能であるcron機能を使用します。
crontabファイルを編集することでcronの実行設定ができます。以下のコマンドを実行するとcrontabを編集できます。

crontab -e

末尾に以下の記述します。

crontab
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
LANG=ja_jp.UTF-8
00 * * * * cd /home/pi/Twitter_bot; python3 ./twitter-bot.py >> /home/pi/Twitter_bot/pylog.log 2>&1

変更を保存してcrontabを閉じます（変更を保存してもcrontabを閉じないと設定が有効にならないようです）。
この設定では毎時00分にtweet-bot.pyが実行されます。実行する時間は好みに合わせて変更してください。
この設定では一時間毎にしかプログラムが実行されません。ツィート間隔を一時間未満にする場合、例えば30分毎に設定する場合は00の部分を*/30にするか、もしくは、

crontab
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
LANG=ja_jp.UTF-8
00 * * * * cd /home/pi/Twitter_bot; python3 ./twitter-bot.py >> /home/pi/Twitter_bot/pylog.log 2>&1
30 * * * * cd /home/pi/Twitter_bot; python3 ./twitter-bot.py >> /home/pi/Twitter_bot/pylog.log 2>&1

のように行を追加して設定してください。

これでbotができあがり、Raspberry Piが定期ツィートを行うサーバーとなりました。

#botの動作
プログラムの初回実行時、正しく動作していればツィートが行われるとともにlasttweet.txt、tweetlog.txt、pylog.logの３つのファイルが生成されます。
lasttweet.txtは前回のツィートの情報が記録されます。以下は出力例です。

lasttweet.txt
1
2019-08-12T05:52:40.068169
試験投稿

このツィートは自作botの試験投稿です
画像/テスト/試験投稿1.png

１行目はツィートのID、２行目はツィートを行った日時、３行目以降はツィートの内容で、最後の行は添付した画像ファイルのパスが書かれています。
プログラムはこのファイルに書かれたIDと日時を参照して次にツィートする時間と内容を決定しています。

tweetlog.txtはプログラムからのログ出力、pylog.logはcron実行時のログが出力されます。実のところ、この２つのファイルはほとんど同じ内容が記録されます。tweetlog.txtへの不要な出力が不要な場合は、twitter-bot.pyのログ出力の設定部分のhandler2が関わる行をすべて削除してください。

登録ツィートを更新する場合は前回作成したアプリでデータベースファイルに定期ツィートの登録・変更を行い、データベースファイルと画像フォルダを差し替えてください。

##おわりに
TwythonとSQLite3を使用してRaspberry Piから定期ツィートするbotを作ることができました。

twittbotやMAKEBOT等のwebサービスと比較して、ツィート登録数の上限なしで利用できることに加え、事前に手動で画像付きツィートしなくても添付画像を登録できるのが嬉しい点であります。
(twittbotやMAKEBOTで画像付きツィートを登録する場合は、事前に手動で画像を添付したツィートを行い、投稿したツィートからpic.twitter.com/で始まるリンクをコピーしてきて貼り付けるしかありませんでした。)

返信機能などは実装できてないので今後の課題としたいと思います。

以上、お読みいただきましてありがとうございました。
ご意見やご指摘がございましたらコメントいただければ幸いです。

#参考

3
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?