Python Tweepyを使ってフォローを自動化

#Python Tweepyを使ってフォローを自動化

## はじめに

Tweepyインストール、OAuth認証、キーワード検索の詳細は以下の投稿を参考にしてください
[Python Tweepyを使ってキーワードを含むツイートを取得](https://qiita.com/morita-toyscreation/items/2abc084f73a9593db26c)
[Python Tweepyを使っていいねを自動化](https://qiita.com/morita-toyscreation/items/69cdefaf59b8e76b8acc)

## tweepy インストール

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

```
$ pip install tweepy
```

## OAuth 認証

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

```Python
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)
```


## キーワード検索

キーワード検索しフォロー処理に必要なパラメーターを取得

```Python
def get_search(api: tweepy.API, q: str, since_id: int, count: int = 10) -> pd.DataFrame:

    q = f"{q} -filter:retweets -from:{ACCOUNT}"

    df = pd.DataFrame(
        columns=[
            "user_id",
            "following",
            "follow_request_sent",
            "followers_count",
            "friends_count",
            "tweet_id",
            "tweet_created_at",
        ]
    )

    last_id = -1
    while len(df) < count:
        count = count - len(df)
        try:

            tweets = api.search(
                q=q,
                count=count,
                tweet_mode="extended",
                locale="ja",
                lang="ja",
                include_entities=False,
                max_id=str(last_id - 1),
                since_id=str(since_id),
            )

            if not tweets:
                break

            for tweet in tweets:
                df = df.append(
                    {
                        "user_id": tweet.user.id,
                        "user_name": tweet.user.name,
                        "user_screen_name": tweet.user.screen_name,
                        "following": tweet.user.following,
                        "follow_request_sent": tweet.user.follow_request_sent,
                        "followers_count": tweet.user.followers_count,
                        "friends_count": tweet.user.friends_count,
                        "tweet_id": tweet.id,
                        "tweet_created_at": tweet.created_at + timedelta(hours=+9),
                    },
                    ignore_index=True,
                )

            last_id = df.iloc[-1]["tweet_id"]

        except tweepy.TweepError as e:
            raise e

    return df
```


## フォローの処理
キーワード検索した結果をループしてUser IDを指定して`create_friendship`を実行することでいいねが実行される
今回はFOLLOWにフォロー実行回数を設定し10回でフォロー処理を終了させる

フォロー履歴はBigQueryに保存

`row.following`はフォロー済
`row.follow_request_sent`はフォロー申請済判定を行う




```Python
COUNT = 200
FOLLOW = 10
KEYWORDS = [
    "#週末ハッカソン",
]

BQ = bigquery.Client()


def follow(event, context):

    df = pd.DataFrame(
        columns=[
            "execution_at",
            "search",
            "since_id",
            "follow_success",
            "follow_failure",
        ]
    )

    api = twitter_api()
    
    # 直近フォロー履歴取得
    follow_history = _get_follow_history()

    for keyword in KEYWORDS:

        print(f"search: {keyword}")

        follow_success = 0
        follow_failure = 0
        execution_at = datetime.now()
        since_id = 0

        # 前回、TweetID
        tmp_df = follow.query(f"search == '{keyword}'")
        if len(tmp_df):
            since_id = tmp_df.iloc[0]["since_id"]

        search = get_search(api, keyword, since_id, COUNT)

        print(f"検索数:{COUNT} ヒット数:{len(search)}")

        if len(search):

            # 次回、TweetID
            since_id = search[0:1]["tweet_id"]

            for _, row in search.iterrows():

                # 合計10回フォローAPI実行終了
                if follow_success + follow_failure >= FOLLOW:
                    break

                try:
                    # 条件追加
                    if not row.following and not row.follow_request_sent:
                        # フォローの処理
                        api.create_friendship(id=row.user_screen_name)
                        follow_success += 1
                        print(f"follow_success: {row.user_id}")
                        time.sleep(4)

                except tweepy.TweepError as e:
                    follow_failure += 1
                    print(f"follow_failure: {row.tweet_id} {e.reason}")

        # 結果保存用のDataFrame
        df = df.append(
            {
                "execution_at": execution_at,
                "search": keyword,
                "since_id": since_id,
                "follow_success": follow_success,
                "follow_failure": follow_failure,
            },
            ignore_index=True,
        )

    print(df)

    # 結果をBigQueryに保存
    _add_follow_history(df)
```

## フォロー履歴をBigQueryに保存

### データセット、テーブル作成

保存用のデータセット作成

```
# データセット追加
bq --location=US mk --dataset --description "twitter_functions dataset using" weekend-hackathon:twitter_functions
```

保存用のテーブル作成

```
# テーブル作成
bq mk --table --description "favorite_history" \
weekend-hackathon:twitter_functions.follow_history schema/follow_history_schema.json
```
スキーマー

```
[
    {
        "description": "API実行日",
        "mode": "REQUIRED",
        "name": "execution_at",
        "type": "TIMESTAMP"
    },
    {
        "description": "検索",
        "mode": "REQUIRED",
        "name": "search",
        "type": "STRING"
    },
    {
        "description": "最終TweetID",
        "mode": "REQUIRED",
        "name": "since_id",
        "type": "INTEGER"
    },
    {
        "description": "成功フォロー数",
        "mode": "REQUIRED",
        "name": "follow_success",
        "type": "INTEGER"
    },
    {
        "description": "失敗フォロー数",
        "mode": "REQUIRED",
        "name": "follow_failure",
        "type": "INTEGER"
    }
]
```

フォロー履歴を取得して検索Tweetをsence_id以降で取得

```Python
def _get_follow_history():
    # 前回結果取得
    sql = f"""
    SELECT
      search,
      MAX(since_id) AS since_id,
    FROM
      `weekend-hackathon.twitter_functions.follow_history`
    GROUP BY
      search
    ORDER BY
      since_id DESC
    """

    return BQ.query(sql).to_dataframe()
```

フォロー履歴を保存

```Python
def _add_follow_history(df: pd.DataFrame):
    # 結果をBigQueryに保存
    table_id = "weekend-hackathon.twitter_functions.follow_history"
    job = BQ.load_table_from_dataframe(df, table_id)
    job.result()
```

---

いいね！と思ったら LGTM お願いします :clap::clap::clap:

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