Edited at

Qiitaの投稿をガーッと取得するバッチ処理用QiitaAPI Pythonラッパー

More than 3 years have passed since last update.

QiitaAPIで「ほぼ」全投稿を取得する で作ったバッチ用のQiitaAPIのPythonラッパーです。

あまりちゃんと作ってなかったのでブラッシュアップしようと思ってたら結局何もせず一カ月放置。。。

APIもバージョンアップされたので、v2対応だけしてとりあえず公開します。


特徴


  • バッチ処理専用です。

  • ページングの処理は隠蔽しているので無駄にネストしたループを書く必要がありません。

  • リクエスト毎にウェイトを入れているのでリクエスト回数制限を気にする必要がありません。

  • リクエストでエラーが発生した際はリトライを行います。

新着投稿を100件ずつ、5ページまで取得してファイルに保存するコード例です。

ページングの処理は内部で行っているので2重のループを書く必要はありません。

qiita2.wait_seconds = 0

for item in qiita2.items(100, 5):
print(item["title"])
qiita2.save_item(item)

items()を呼ぶと投稿のjsonオブジェクトのイテレータを返すので、for文やsortでそのまま使用できます。


対応API

必要なものだけつくったのでこれだけです。


  • GET /api/v2/items (新着投稿一覧取得)

  • GET /api/v2/tags (タグ一覧取得)

  • GET /api/v2/tags/:id/items (特定タグ投稿一覧取得)


使い方


アクセストークンの指定

qiita2のモジュール変数で設定します。

qiita2.auth_token = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"


設定

qiita2のモジュール変数で設定します。

変数名
デフォルト
内容

default_per_page
100
1ページ当たりの取得件数(各APIでper_pageを省略した時のデフォルト)

default_max_page
100
最大ページ数(各APIでmax_pageを省略した時のデフォルト)

wait_seconds
12
リクエスト送信前のウェイト秒数

retry_wait_min
1
エラー発生時のリトライまでのウェイト(分)

retry_limit
10
エラー発生時のリトライ回数制限


API


新着投稿一覧の取得

新着投稿一覧のイテレータを返す。

#デフォルトの取得件数と最大ページ数で取得

items = qiita2.items()

#取得件数と最大ページ数を指定して取得

items = qiita2.items(per_page=20, max_page=5)


タグ一覧の取得

タグ一覧のイテレータを返す。

items = qiita2.tags()

items = qiita2.tags(per_page, max_page)


特定タグの投稿一覧の取得

特定タグ投稿一覧のイテレータを返す。

items = qiita2.tag_items(tag_url)

items = qiita2.tag_items(tag_url, per_page, max_page)


トータル件数の取得

Total-Countレスポンスヘッダから取得します。

len(qiita2.items())


投稿をファイルへ保存

ファイル名はdata/items/<投稿ID>.jsonです。←保存先固定かよ!

qiita2.save_item(item)


ソース

1ファイルなのでそのままコピペで使えます。


qiita2.py

import time

import codecs
import json
from logging import getLogger
import requests
from urllib.parse import urlparse, parse_qs

logger = getLogger(__name__)

URL_ITEMS = "https://qiita.com/api/v2/items"
URL_TAG_ITEMS = "https://qiita.com/api/v2/tags/%s/items"
URL_TAGS = "https://qiita.com/api/v2/tags"

HEADER_TOTAL = "Total-Count"
LINK_NEXT = "next"
LINK_LAST = "last"

default_per_page = 100
default_max_page = 100
wait_seconds = 12
retry_wait_min = 1
retry_limit = 10

auth_token = None

def items(per_page = default_per_page, max_page = default_max_page):
req = QiitaRequest(URL_ITEMS, per_page, max_page)
return QiitaIterator(req)

def tag_items(tag_url, per_page = default_per_page, max_page = default_max_page):
req = QiitaRequest(URL_TAG_ITEMS % tag_url, per_page, max_page)
return QiitaIterator(req)

def tags(per_page = default_per_page, max_page = default_max_page):
req = QiitaRequest(URL_TAGS, per_page, max_page)
return QiitaIterator(req)

class QiitaIterator:
def __init__(self, req):
self.req = req
self.items = req.request().__iter__()

def __iter__(self):
return self

def __next__(self):
if self.items == None: raise StopIteration
try:
val = self.items.__next__()
return val
except StopIteration:
if self.req.has_next():
self.items = self.req.next().__iter__()
return self.__next__()
else:
raise StopIteration

def __len__(self):
return self.req.total_count()

class QiitaRequest:

last_request_time = None

retry_num = 0

def __init__(self, url, per_page = default_per_page, max_page = default_max_page, page = 1):
self.url = url
self.per_page = per_page
self.max_page = max_page
self.page = page
self.res = None
self.current_page = None

def request(self):
self.links = dict()
params = {"per_page": self.per_page, "page": self.page}
return self.__request__(self.url, params)

def __request__(self, url, params = None):
self.__wait__()
logger.info("url:%s" % url)

headers = {"Authorization": "Bearer " + auth_token} if auth_token != None else None
self.res = requests.get(url, params = params, headers = headers)
status = self.res.status_code

while status != 200 and QiitaRequest.retry_num <= retry_limit:
logger.warning("status:%d" % status)
logger.warn(u"%d分待機します。" % retry_wait_min)
time.sleep(retry_wait_min * 60)
QiitaRequest.retry_num = QiitaRequest.retry_num + 1
self.res = requests.get(url, params = params)
status = self.res.status_code

if status != 200:
logger.warning("status:%d" % status)
logger.warning(self.res.text)
return None

QiitaRequest.retry_num = 0
return self.res.json()

def next(self):
if not self.has_next(): raise Exception()
# v2でLinkレスポンスヘッダでper_pageが欠落している不具合の対応
params = {"per_page": self.per_page}
return self.__request__(self.res.links[LINK_NEXT]["url"], params)

def retry(self):
pass
def has_error(self):
pass
def has_next(self):
if not LINK_NEXT in self.res.links: return False
url = self.res.links[LINK_NEXT]["url"]
page = self.__get_page__(url)
return page <= self.max_page

def last_page(self):
url = self.res.links[LINK_LAST]["url"]
return self.__get_page__(url)

def total_count(self):
return int(self.res.headers[HEADER_TOTAL])

def __get_page__(self, url):
query = urlparse(url).query
page = parse_qs(query)["page"][0]
return int(page)

def __wait__(self):
if QiitaRequest.last_request_time != None:
last = QiitaRequest.last_request_time
now = time.clock()
wait = wait_seconds - (now - last)
if 0 < wait:
time.sleep(wait)
QiitaRequest.last_request_time = time.clock()

def save_item(item):
item_id = item["id"]
filename = "data/items/%s.json" % item_id
with codecs.open(filename, "w", "utf-8") as f:
f.write(json.dumps(item, indent = 4, ensure_ascii=False))