Pythonでツイートを生成したい！ -マルコフ連鎖編-

環境

ubuntu 16.04 LTS
python 3.7.3

やること

マルコフ連鎖をpythonで簡単に実装し、ツイートを自動生成します。元データにはTwitter APIで取得したツイートを使用します。

マルコフ連鎖について

マルコフ連鎖とは一つ前の状態から次の状態が決まるような連鎖です。例えば「今日は晴れだった」「明日は雨が降るらしい」という2つの文があった場合、「今日」であれば次は「は」になる確率が100%です。その「は」については、「晴れ」と「雨」がそれぞれ50%の確率で連鎖していきます。また「らしい」はそれに続く単語がないので連鎖がストップします。
つまり、データに存在する任意の単語に対してどんな単語がどれくらいの確率で続くかを分かるようにすればいいということです。

①データの収集・前処理

Twitter APIのsearch/tweetsでツイートを取得します。皆様の悲痛な叫びを代弁するため、検索ワードは「月曜」にしました。ツイートを取得後、ユーザーIDやURL、空白の行を削除します。また検索結果にRTを含むと直近でバズっているツイートがある場合そればっかりになってしまうので検索オプションであらかじめ除きます。前処理まで完了したら、テキストファイルとして一旦保存します。

Twitter APIやpythonにおけるsearch/tweetsの基本的な使い方については前記事のhttps://qiita.com/h_tashiro/items/ed119c237f5595c3d7b8 をご覧ください。

コード

import urllib
import io
from requests_oauthlib import OAuth1Session, OAuth1
import requests
import sys
import re



def main():
    # APIの認証キー
    CK = 'xxxxxxxxxxxxxxxx'
    CKS = 'xxxxxxxxxxxxxxxx'
    AT = 'xxxxxxxxxxxxxxxx'
    ATS = 'xxxxxxxxxxxxxxxx'
    # 検索ワード
    word = '月曜'
    # 検索時のパラメーター
    count = 100 # 一回あたりの検索数(最大100/デフォルトは15)
    range = 180 # 検索回数の上限値(最大180/15分でリセット)
    # インスタンス作成
    get = Get_tweets()
    # ツイート検索
    tweets = get.search_tweets(CK, CKS, AT, ATS, word, count, range)
    # 前処理
    tweets = get.preprocess(tweets)
    # テキストファイルに保存
    path = 'tweets.txt'
    get.to_text(tweets, path)

class Get_tweets:

    def search_tweets(self, CK, CKS, AT, ATS, word, count, range):
        # 文字列設定
        word += ' exclude:retweets' # RTは除く
        word = urllib.parse.quote_plus(word)
        sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
        # リクエスト
        url = "https://api.twitter.com/1.1/search/tweets.json?lang=ja&q="+word+"&count="+str(count)
        auth = OAuth1(CK, CKS, AT, ATS)
        response = requests.get(url, auth=auth)
        data = response.json()['statuses']
        # ２回目以降のリクエスト
        cnt = 1
        tweets = []
        count_tweets = 0
        while True:
            if len(data) == 0:
                break
            if cnt > range:
                break
            cnt += 1
            for tweet in data:
                tweets.append(tweet['text'])
                count_tweets += 1
                maxid = int(tweet["id"]) - 1
            url = "https://api.twitter.com/1.1/search/tweets.json?lang=ja&q="+word+"&count="+str(count)+"&max_id="+str(maxid)
            response = requests.get(url, auth=auth)
            try:
                data = response.json()['statuses']
            except KeyError: # リクエスト回数が上限に達した場合のデータのエラー処理
                print('上限まで検索しました')
                break
        print('取得したツイート数 :', count_tweets)
        return tweets

    def preprocess(self, tweets):
        for i in range(len(tweets)):
            tweets[i] = re.sub(r'@\w+', '', tweets[i])  # ユーザーID
            tweets[i] = re.sub(r'https?://[\w/:%#\$&\?\(\)~\.=\+\-]+', '', tweets[i])  # URL
            tweets[i] = re.sub(r'\n{2,}', '\n', tweets[i])  # 空白の行]
        return tweets

    def to_text(self, tweets, path):
        with open(path, 'w') as f:
            for tweet in tweets:
                f.write(tweet + '\n')


if __name__ == '__main__':
    main()

②分かち書き

得られたツイートを品詞ごとに分かち書き（半角スペースで空ける）をします。Mecabを使い、①で保存したテキストファイルから分かち書き後のテキストファイルを出力して保存します。専用のライブラリやsubprocessを使ってpython上でも実行できますが、今回はサクッとコマンドラインでMecabを使います。

$ mecab tweets.txt -O wakati -o wakati.txt

③マルコフ連鎖の実装とツイート生成

マルコフ連鎖について

先ほどの「今日は晴れだった」「明日は雨が降るらしい」という文であれば、{'今日':['は'], 'は':['晴れ','雨'], ...}というように、一つ前の単語をキーとして、それに続く単語をリストにします。この時リスト内での単語の重複を許します。また辞書にする前に一度行ごとに単語をリスト化するのですが、行の始めには「0」を開始文字として挿入します。この「0」は終端文字としても機能します。

ツイート生成について

マルコフ連鎖に限らず、例えばRNNなどの時系列解析による文章自動生成でも一つ前（まで）の状態を考慮して単語をつなげていきます。ここで、数列において初項が大切なように、どんな単語から文章を始めるかというのはしばしば問題になります。今回は「月曜」を最初の単語として文章を生成しますが、例えば開始文字の直後＝文頭の単語をランダムに選ぶなどのやり方もあります。

最初の単語が決まったら、その単語をキーとして値であるリスト内から単語をランダムに選択します。リストをつくる際に単語の重複を許したので、元データにおける単語の出現確率をそのまま反映できます。この連鎖を終端文字にぶつかるまでつなげます。

また連鎖によってはかなり長い文章を生成してしまうのでlimitというパラメーターで単語数を制限します。今回は30で統一します。

コード

import random


def main():
    # インスタンス生成
    markov = Markov()
    # 分かち書きしたテキストファイルのパス
    path = 'wakati.txt'
    # 最初の単語
    first_word = '月曜'
    # 連鎖数の上限
    limit = 30
    # 単語ごとの辞書を作成
    dict = markov.to_dict(path)
    # ツイートを１０回生成
    for i in range(10):
        tweet = markov.gen_tweet(dict, first_word, limit)
        print(tweet)


class Markov:

    def to_dict(self, path):
        wordlist = []
        with open(path, 'r') as f:
            line = f.readline()
            while line:
                pre_wordlist = [0]
                pre_wordlist.extend(line.split())
                wordlist.extend(pre_wordlist)
                line = f.readline()
        wordlist.append(0)
        dict = {}
        pre_word = ''
        for word in wordlist:
            if pre_word:
                if pre_word in dict:
                    list = dict[pre_word]
                else:
                    list = []
                list.append(word)
                dict[pre_word] = list
            pre_word = word
        return dict

    def gen_tweet(self, dict, first_word, limit):
        tweet = ""
        word = first_word
        cnt = 0
        while cnt <= limit:
            if word == 0:
                break
            tweet += word
            word = random.choice(dict[word])
            cnt += 1
        return tweet


if __name__ == '__main__':
    main()

実行結果

月曜にされなくてたら肩が7/21時よりディナー予約表で快適そのもので早いなぁ…😴
月曜のまとめ
月曜の朝からバタバタなのに…
月曜は月曜で最も重要ライブできるから7月なことでよければ底過ぎる
月曜きらい。
月曜夕方ヒトカラスカポコチャ配信です今月会いたいですし親指のサポート！是非遊んで頑張れそうでは休みだというやる気がいませ
月曜の首に！！明日月曜一限で、何時31分の情報を想定内の朝
月曜じゃん。九州、本番であっ、月曜から頑張らなく母と李くんやっくんになりそうです
月曜なはずな
月曜でさえ日曜の朝は頑張れる！！

ヒトカラスカポコチャ配信をする方もいるようですが、どちらかというとやはり負の感情が強いですね。出来としてはかなり微妙です。特に固有名詞や特殊な単語が入ると汎用性を下げるので（そこが面白いところではありますが）、例えば単語の出現回数を二乗する、対数をとるなどして単語選択の確率をそれに応じたものにしたり、出現回数が1回の単語を除くなどの方法が考えられます。また一つ前の単語との関わりしか考慮されないので、全体としての意味に一貫性のない文章になりがちです。特に後者を解決するために、LSTMなどの深層学習での時系列解析を試してみたいです。