【Python】Twilog + Pythonで他人の過去の全ツイートを取得するコード

statsbeginner.hatenablog.com

を書いている方がいたのですが、BeautifulSoupとリスト内包表記を使えばもう少し楽に書けるということを示すために書きました。

参考資料 PythonとBeautiful Soupでスクレイピング - Qiita

import time
from urllib import request
from datetime import datetime as dt

import pandas as pd
from bs4 import BeautifulSoup

class TwilogParser:
    def __init__(self):
        self.account = None
        
    def set_account(self, target_account):
        self.account = target_account

    def get_page_df(self, num):
        url = self.make_twilog_url(num)
        soup = self.make_soup(url)
        return self.soup_to_df(soup)

    def make_twilog_url(self, num):
        # /allascで古い順にデータを取る必要性を感じなかったため、新しい順のURLを利用
        base_url = "http://twilog.org/%s/"%self.account
        if num <= 1:
            return base_url
        else:
            return base_url + "%d"%num

    def make_soup(self, url):
        # Pythonではファイルの読み書き等ではwith構文を使う
        # .closeしなくても、ブロックを越えると自動でファイルを閉じてくれるので安心
        with request.urlopen(url) as response:
            html = response.read()
        return BeautifulSoup(html)

    def soup_to_df(self, soup):
        daily_titles = soup.findAll('h3', class_='title01')
        daily_dfs = [self._daily_titles_to_df(d) for d in daily_titles]
        return pd.concat(daily_dfs) # 各日付のDataFrameが入ったリストを結合する

    def _get_user_ids(self, tweets):
        names = [t.find(class_ = 'tl-name').span for t in tweets]
        return [self._get_str(n) for n in names]

    def _get_tweet_ids(self, tweets):
        return [self._convert_tweet_id(t) for t in tweets] #t.get('id')は非公開ツイートだとNoneを返す

    def _get_tweet_texts(self, tweets):
        return [t.find(class_ = 'tl-text').text for t in tweets]

    def _get_tweet_times(self, tweets):
        atags = [t.find(class_ = 'tl-posted').find('a') for t in tweets]
        return [self._get_str(a) for a in atags]
    
    def _daily_titles_to_df(self, daily_title):
        date = self._to_date(daily_title.find('a'))
        tweets = daily_title.findNext('section').findAll(class_ = 'tl-tweet') 
        return self._tweets_to_df(tweets, date)

    def _tweets_to_df(self, tweets, date):
        dailydata = {
            'id': self._get_tweet_ids(tweets),
            'user': self._get_user_ids(tweets),
            'date': [date for i in tweets], #日付は全て同じ値
            'time': self._get_tweet_times(tweets),
            # 'rt_flg': ユーザ名が別の人になっていることで判別する
            'text': self._get_tweet_texts(tweets)
        }
        df = pd.DataFrame(dailydata)
        df['rt_flg'] = (df['user'] != ('@' + self.account))
        return df
    
    def _get_str(self, tag):
        if tag is None:
            return None
        else:
            return tag.string
    
    def _convert_tweet_id(self, tag):
        raw_tweet_id = tag.get('id')
        if raw_tweet_id is None:
            return None
        else:
            return raw_tweet_id[2:] #文字列のスライスで最初の二文字(tw)を取る

    def _to_date(self, soup):
        try:
            string = soup.string
            datetime = dt.strptime(string[:-3], '%Y年%m月%d日')
            return dt.date(datetime)
        except:
            return None

# この.pyファイルが実行ファイルの実行される部分
if __name__ == '__main__':
    parser = TwilogParser()
    parser.set_account('takeshi0406')
    # とりあえず5ページ分取得
    # 時間はかかるが、取得するツイートが無くなるまでループを回し続けれるようにすれば全件取得できる
    for i in range(5):
        if i == 0:
            df = parser.get_page_df(i+1)
        else:
            df = pd.concat([df, parser.get_page_df(i+1)])
        # 一気にアクセスするとサーバーに負荷をかけてしまうため、1秒ずつスリープさせる
        time.sleep(1)
    df.to_csv('./output.csv', index=False)