twint

date: 2021-02-06 excerpt: twintの使い方

twintの使い方

twintとはtwitterのAPIの制限を超えて使うためにapiのifのような項目を集められるスクレイパーである
いつtwitterにブロックされてもしょうがないソフトウェアなので、あまり期待しないで用いるのが良い
followerやfollowしているユーザを取得する機能もあったが、twitterによってブロックされた
token.pyでエラーが出る -> トークンを取得するIPがブロックされている

install

$ python3 -m pip install twint

アドホックな対応がある場合はgithubからインストールする

help

$ twint -h | less

特定のキーワードを検索する

$ twint -s ${KEYWORD}

特定の期間で検索する

$ twint -s ${KEYWORD} --since "%Y-%m-%d %H:%M:%S" --until "%Y-%m-%d %H:%M:%S"

proxyを使用する

http proxyの例

$ twint -s ${KEYWORD} --proxy-type http --proxy-host ${IP_ADDRESS} --proxy-port ${PORT_NUMBER}

取得件数を制限する

$ twint -s ${KEYWORD} --limit ${LIMIT_NUM}

具体的な例(キーワードを日毎に検索する)

import datetime
import os
from pathlib import Path
import random
import pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor

proxy_df = pd.read_csv("index.html", sep=":", header=None)
proxy_df.columns = ["host", "port"]

def scan(term, years=2):
    Path(term).mkdir(exist_ok=True)
    start = datetime.datetime.strptime("2021-02-01", "%Y-%m-%d")

    ts = []
    for i in range(365 * years):
        t = (start - datetime.timedelta(days=i))
        ts.append(t)

    def _worker(t):
        filename_str = t.strftime("%Y-%m-%d") + ".csv"
        if Path(f"{term}/{filename_str}").exists():
            return

        start_str = t.strftime("%Y-%m-%d") + " 00:00:00"
        end_str = t.strftime("%Y-%m-%d") + " 23:59:59"

        row = proxy_df.sample(frac=1).iloc[0]
        host = row.host
        port = row.port

        cmd = f'twint -s {term} --proxy-type http --proxy-host {host} --proxy-port {port} --limit 1000 --since "{start_str}" --until "{end_str}" --csv -o {term}/{filename_str} '
        print(cmd)
        os.system(cmd)
        time.sleep(10)

    ts = random.sample(ts, len(ts))
    '''
    for t in ts:
        _worker(t)
    '''
    with ThreadPoolExecutor(max_workers=5) as exe:
        for _ in exe.map(_worker, ts):
            _


if __name__ == "__main__":
    while True:
        scan('維新の会', years=2)
        scan('N国', years=2)
        scan('れいわ新選組', years=2)
        scan('電通', years=6)
        scan('セブンイレブン', years=3)
        scan('くら寿司', years=3)
        scan("東京五輪")
        scan('プリキュア', years=5)
        scan('けものフレンズ', years=5)
        scan('ドコモ口座')
        scan('ドコモ')
        scan('テラスハウス')
        scan('ゆうちょ銀行')
        scan('自民党', years=2)
        scan('公明党', years=2)
        scan('立憲', years=2)
        scan('共産党', years=2)
        scan('国民民主', years=2)
        scan('中核派', years=2)
        scan('山本太郎', years=2)
        scan('コロワイド')

具体的な例(特定のユーザのタイムラインを取得する)

import twint
import pandas as pd
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

proxy_df = pd.read_csv("index.html", sep=":", header=None)
proxy_df.columns = ["host", "port"]

def run(screen_name):
    if (Path(f"user_tweets") / screen_name).exists():
        return
    row = proxy_df.sample(frac=1).iloc[0]
    host = row.host
    port = row.port

    print(screen_name)

    config = twint.Config()
    config.Proxy_host = host
    config.Proxy_port = port
    config.Proxy_type = "http"
    config.Limit = 1000

    config.Username = screen_name
    config.Store_csv = True
    config.Output = "user_tweets/" + screen_name

    try:
        twint.run.Search(config)
    except ValueError as exc: # 存在しないユーザだと例外が出る
        print(exc)
        pass

df = pd.read_csv("screen_name.csv")
# for screen_name in df.sample(frac=1).screen_name:
#    run(screen_name)

screen_names = df.sample(frac=1).screen_name.tolist()
with ThreadPoolExecutor(max_workers=5) as exe:
    for _ in exe.map(run, screen_names):
        _

具体的な例(キーワードで検索しその結果を保存する)

import twint
import pandas as pd
from pathlib import Path

proxy_df = pd.read_csv("index.html", sep=":", header=None)
proxy_df.columns = ["host", "port"]


def run(arg):
    row = proxy_df.sample(frac=1).iloc[0]
    host = row.host
    port = row.port

    config = twint.Config()
    config.Proxy_host = host
    config.Proxy_port = port
    config.Proxy_type = "http"
    config.Limit = 30000
    # config.Username = screen_name
    config.Search = 'youtube "花粉" lang:ja'
    config.Store_csv = True
    config.Output = f"sample.csv"
    try:
        twint.run.Search(config)
    except ValueError as exc: # 存在しないユーザだと例外が出る
        print(exc)
        pass
    except Exception as exc:
        print(exc)

run(None)

トラブルシューティング

`token.py`でエラーが出る時

site-packages/twint/token.pyのtoken取得のコードを変更する

変更点
- sessionをproxyを利用するように変更

    def __init__(self, config):
        proxy_df = pd.read_csv("index.html", sep=":", header=None)
        proxy_df.columns = ["host", "port"]
        row = proxy_df.sample(frac=1).iloc[0]
        host = row.host
        port = row.port

        proxies=dict(http=f"http://{host}:{port}", https=f"http://{host}:{port}")

        self._session = requests.Session()
        self._session.proxies.update(proxies)
        self._session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:78.0) Gecko/20100101 Firefox/78.0'})
        self.config = config
        self._retries = 5
        self._timeout = 10
        self.url = 'https://twitter.com'