2
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 3 years have passed since last update.

Python Tweepyを使ってツイートを自動化

Last updated at Posted at 2021-02-06

Python Tweepyを使ってツイートを自動化

はじめに

Tweepyインストール、OAuth認証の詳細は以下の投稿を参考にしてください
Python Tweepyを使ってキーワードを含むツイートを取得

概要

自動ツイートには Cloud Functions と Cloud Scheduler を使う
1分毎にCloud Functionsを実行し対象があれば自動ツイートを行う
ツイートしたい時間、投稿はCSVで管理する

tweepy インストール

Twitter API の操作はtweepyが便利、是非使おう。

$ pip install tweepy

OAuth 認証

Twitter API を操作するには OAuth認証が必要
Developerサイトで取得したAPI KEYを使い認証を行う。

import tweepy


# 取得したAPI KEY、TOKENを入力
API_KEY = ""
API_SECRET_KEY = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""


def twitter_api() -> tweepy.API:
    auth = tweepy.OAuthHandler(API_KEY, API_SECRET_KEY)
    auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
    return tweepy.API(auth)

ツイート

def tweet(event, context):

    api = twitter_api()

    try:
        api.update_status(
            status="今日も1日頑張るぞい #週末ハッカソン https://weekend-hackathon.connpass.com/event/203873/",
        )
        time.sleep(4)
    except tweepy.TweepError as e:
        print(f"tweet_failure: {e.reason}")

実行結果
スクリーンショット 2021-02-06 14.04.52.png

メディア付きツイート

def tweet(event, context):

    api = twitter_api()

    try:
        api.update_with_media(
            status="今日も1日頑張るぞい #週末ハッカソン https://weekend-hackathon.connpass.com/event/203873/",
            filename="./toyscreation.png",
        )
        time.sleep(4)
    except tweepy.TweepError as e:
        print(f"tweet_failure: {e.reason}")

実行結果
スクリーンショット 2021-02-06 14.09.11.png

複数画像付きツイート

def tweet(event, context):

    api = twitter_api()

    try:
        file_names = ["./marron1.jpg", "./marron2.jpg"]
        media_ids = []
        for filename in file_names:
            res = api.media_upload(filename)
            media_ids.append(res.media_id)
        
        api.update_status(
            status="今日も1日頑張るぞい #週末ハッカソン \nhttps://weekend-hackathon.connpass.com/event/203873/",
            media_ids=media_ids
        )
        time.sleep(4)
    except tweepy.TweepError as e:
        print(f"tweet_failure: {e.reason}")

実行結果
スクリーンショット 2021-02-06 14.30.33.png

CSV管理

ツイートしたい時間、投稿はCSVで管理する
run_time ... 投稿時間
start_at ... 開始期間
end_at ... 終了期間
weekday ... 投稿週(0:日曜〜6:土曜)
message ... 投稿内容
filename ... 投稿画像(カンマ区切りで複数)

下記の例では2021-01-01から2022-01-01の間、毎週、21:30にツイートを行う

run_time,start_at,end_at,weekday,message,filename
21:30,2021-01-01,2022-01-01,"'0123456","今日も1日頑張るぞい #週末ハッカソン \nhttps://weekend-hackathon.connpass.com/event/203873/",
21:30,2021-01-01,2022-01-01,"'0123456","マロン画像","./marron1.jpg,./marron2.jpg"

PandasでCSVデータ取得し条件で絞り込み
ループしてツイートを実行する

def tweet(event, context):

    # CSV取得
    tweet_df = pd.read_csv(
        os.getcwd() + "/tweet.csv",
        thousands=",",
        engine="python",
        encoding="UTF-8",
        index_col=False,
    )

    now = datetime.now()
    weekday = now.weekday()
    run_time = now.strftime("%H:%M")

    tweet_df["start_at"] = pd.to_datetime(tweet_df["start_at"])
    tweet_df["end_at"] = pd.to_datetime(tweet_df["end_at"])
    tweet_df = tweet_df[
        (now >= tweet_df["start_at"])
        & (tweet_df["end_at"] >= now)
        & (tweet_df["weekday"].str.contains(str(weekday)))
        & (tweet_df["run_time"] == run_time)
    ]

    api = twitter_api()

    try:

        for _, row in tweet_df.iterrows():

            if pd.isnull(row["filename"]):
                api.update_status(
                    status=row["message"].replace("\\n", "\n"),
                )
                print(f"tweet_success: {row['message']}")
            else:
                file_names = row["filename"].split(",")
                media_ids = []
                for filename in file_names:
                    res = api.media_upload(filename)
                    media_ids.append(res.media_id)

                api.update_status(
                    status=row["message"].replace("\\n", "\n"), media_ids=media_ids
                )
                print(f"tweet_success: {row['message']}")

            time.sleep(4)
    except tweepy.TweepError as e:
        print(f"tweet_failure: {e.reason}")

Cloud Functions 登録

作成したスクリプトをCloud Functionsに登録
詳細はこちらを参考 → Cloud Functions Pythonを使ったローカル開発環境とデプロイ方法

$ gcloud functions deploy tweet --timeout=300 --entry-point=tweet --source=./functions/tweet \
--runtime python37 --trigger-topic tweet_action --allow-unauthenticated --set-env-vars TZ=Asia/Tokyo

Cloud Scheduler 登録

コンソールから登録したFunctionsを1分置きに実行
詳細はこちらを参考 → Google cloud scheduler で Functionsを定期実行する
スクリーンショット 2021-02-06 22.27.53.png

CSV管理した投稿時間でツイートされれば完了

実行結果
スクリーンショット 2021-02-06 22.52.25.png


シンプルですが定期的に投稿したいメッセージがある場合に有効活用できるのではないでしょうか:thumbsup:
いいね!と思ったら LGTM お願いします :clap::clap::clap:

【PR】週末ハッカソンというイベントやってます! → https://weekend-hackathon.toyscreation.jp/about/

2
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
2
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?