【Python】Qiitaのユーザーの全記事についたLGTM・ストックをcsvに出力するやつ

Qiita APIをたたいて特定ユーザーの全記事についてLGTM・ストックを収集してcsvに吐き出すpythonコードを作成しました。

できること

Qiitaの指定ユーザーの投稿した全記事に対して

• 記事のタイトル・LGTM数・URL等の情報のcsvファイル
• 全LGTMの付いた時刻・付けたユーザー等の情報のcsvファイル
• 全ストックの付いた時刻・付けたユーザー等の情報のcsvファイル

をQiita APIから取得し出力します。

使い方・python or Docker

こちらのGithubリポジトリからコードを取得してください。

• https://github.com/k8shiro/qiita-lgtm-graph

準備するもの

• 取得したいユーザー名 例) ko-he-8
• Qiitaのアクセストークン
• 結果のcsv出力用ディレクトリ

注意

• Qiita APIの回数制限(トークンありの場合1000回/時間)回避のため処理中にsleepが入っています。
• --tokenを指定しない場合も動きますが、Qiita APIの回数制限に引っかかる可能性があります。

pythonを使う場合

$ pip install -r requirements.txt
$ python src/to_csv.py <ユーザー名> <出力ディレクトリ> --token=<Qiitaトークン>

Dockerを使う場合

$ docker build -t qiita-lgtm .
$ docker run -v $(pwd)/src:/src -v $(pwd)/<出力ディレクトリ>:/data qiita-lgtm python to_csv.py <ユーザー名> /data --token=<Qiitaトークン>

コードとやっていること

1. itemsAPIからユーザー全記事を取得
   1. ページネーションに対応してるため複数回呼ばれることがある
   2. 取得情報をpandas.DataFrameに変換してcsv出力
2. item idごとにlikesAPIから記事ごとに全LGTM取得
   1. ページネーションに対応してるため複数回呼ばれることがある
   2. 取得情報をpandas.DataFrameに変換して連結後csv出力
3. item idごとにstockersAPIから記事ごとに全ストック取得
   1. ページネーションに対応してるため複数回呼ばれることがある
   2. 取得情報をpandas.DataFrameに変換して連結後csv出力

import argparse
import urllib.request
import json
import os
import time
import pandas as pd


def get_user_items(user, token):
    url = 'https://qiita.com/api/v2/items'
    counter = 0

    items = []
    for idx in range(1,101):
        time.sleep(3.6)
        print('.', end='', flush=True)
        counter += 1
        params = {
            'page': idx,
            'per_page': 100,
            'query': 'user:{}'.format(user)
        }

        headers = {'Authorization': 'Bearer {}'.format(token)} if token else {}
        req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(params)), headers=headers)
        with urllib.request.urlopen(req) as res:
            page_items = json.loads(res.read().decode('utf8'))
        if len(page_items) == 0:
            break
        items.extend(page_items)
    return items, counter

def get_items_likes(item_id, token):
    url = 'https://qiita.com/api/v2/items/{}/likes'.format(item_id)
    counter = 0

    likes = []
    for idx in range(1,101):
        time.sleep(3.6)
        print('.', end='', flush=True)
        counter += 1
        params = {
            'page': idx,
            'per_page': 100,
        }
        headers = {'Authorization': 'Bearer {}'.format(token)} if token else {}
        req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(params)), headers=headers)
        with urllib.request.urlopen(req) as res:
            page_likes = json.loads(res.read().decode('utf8'))
        if len(page_likes) == 0:
            break
        likes.extend(page_likes)
    return likes, counter

def get_items_stockers(item_id, token):
    url = 'https://qiita.com/api/v2/items/{}/stockers'.format(item_id)
    counter = 0
    
    stockers = []
    for idx in range(1,101):
        time.sleep(3.6)
        print('.', end='', flush=True)
        counter += 1
        params = {
            'page': idx,
            'per_page': 100,
        }
        headers = {'Authorization': 'Bearer {}'.format(token)} if token else {}
        req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(params)), headers=headers)
        with urllib.request.urlopen(req) as res:
            page_stockers = json.loads(res.read().decode('utf8'))
        if len(page_stockers) == 0:
            break
        stockers.extend(page_stockers)
    return stockers, counter

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('user', help='user name')
    parser.add_argument('data_dir', help='data export directory')
    parser.add_argument('--token', help='access token')
    
    args = parser.parse_args()
    user = args.user
    data_dir = args.data_dir
    token = args.token

    items, counter = get_user_items(user, token)
    items_df = pd.DataFrame(data=items, columns=['created_at', 'id', 'likes_count', 'title', 'url'])
    items_df.to_csv(os.path.join(data_dir, '{}_items.csv'.format(user)))
    print("items API: called {} times".format(counter))
    print(items_df)

    likes_df = pd.DataFrame()
    counter = 0
    for item_id in items_df['id']:
        item_likes, _counter = get_items_likes(item_id, token)
        counter += _counter
        item_likes_df = pd.DataFrame(data=item_likes, columns=['created_at', 'user'])
        item_likes_df['id'] = item_likes_df['user'].apply(lambda x: x['id'])
        item_likes_df['name'] = item_likes_df['user'].apply(lambda x: x['name'])
        item_likes_df['permanent_id'] = item_likes_df['user'].apply(lambda x: x['permanent_id'])
        item_likes_df['organization'] = item_likes_df['user'].apply(lambda x: x['organization'])
        item_likes_df = item_likes_df.drop('user', axis=1)
        item_likes_df['item_id'] = item_id
        likes_df = pd.concat([likes_df, item_likes_df])
    likes_df = likes_df.reset_index()
    likes_df.to_csv(os.path.join(data_dir, '{}_likes.csv'.format(user)))
    print("likes API: called {} times".format(counter))
    print(likes_df)

    stockers_df = pd.DataFrame()
    counter = 0
    for item_id in items_df['id']:
        item_stockers, _counter = get_items_stockers(item_id, token)
        counter += _counter
        item_stockers_df = pd.DataFrame(data=item_stockers, columns=['id', 'permanent_id', 'title', 'name', 'organization'])
        item_stockers_df['item_id'] = item_id
        stockers_df = pd.concat([stockers_df, item_stockers_df])
    stockers_df = stockers_df.reset_index()
    stockers_df.to_csv(os.path.join(data_dir, '{}_stockers.csv'.format(user)))
    print("stockers API: called {} times".format(counter))
    print(stockers_df)