自分の書いたものを4媒体ぶん、1本のスクリプトで集計しようとしました。書き終えて実行したら、こう出ました。
全4893記事 / 反響合計 11420
zenn 4800本 反響11379 中央値 0.0 0本2809 上位1本の占有 7.2%
qiita 93本 反響 41 中央値 0.0 0本 62 上位1本の占有 12.2% ストック 32
Qiita の93本は合っています。Zenn の4,800本は、わたしの記事ではありません。dev.to は1行も出ていません。
エラーは1つも出ていません。
名前が3つとも違いました
原因は単純でした。コマンドライン引数でユーザー名を1つ受け取って、それを3つの関数にそのまま渡していました。
user = sys.argv[1]
rows = qiita(user) + devto(user) + zenn(user)
渡した名前は Qiita のもので、頭が大文字です。Zenn と dev.to のURLを見ると、そちらは小文字でした。同じ人間なのに、媒体ごとに表記が違います。自分で登録したはずですが、覚えていませんでした。
ここまでは、よくある取り違えです。問題は、その先の反応が3通りに分かれたことでした。
3通りの壊れ方
Qiita は正しく返しました。名前が合っていたので当然です。
dev.to は空の配列を返しました。該当するユーザーがいないので、記事もゼロ件。集計する側から見ると「取れなかった」と分かるので、これは安全な壊れ方です。実際、出力にも dev.to の行が出ていません。気づけます。
Zenn は、絞り込みをやめて全件を返しました。
4,800本です。他人の記事が入っています。そして、そのまま集計が通りました。中央値0.0、反響ゼロの記事が2,809本、上位1本の占有7.2%。どれも計算としては正しく、形としては何もおかしくありません。
もしわたしが Zenn の記事数を把握していなかったら、この数字を眺めて「Zenn では読まれていない」と結論づけていたはずです。中央値0.0という数字は、それらしく見えます。
入力の間違いから、正しい形の出力が出てくる。これがいちばん困ります。
絞り込みを、APIに任せないことにした
直し方は2つ考えました。
ひとつは、名前を媒体ごとに持つことです。引数で受け取るのをやめて、辞書に書きます。
USERS = {
'qiita': 'Rapls',
'zenn': 'rapls',
'devto': 'rapls',
}
これで今回の間違いは消えます。ただし、また別の名前を書き間違えたら、同じことが起きます。
なので、もうひとつ足しました。返ってきたデータを、こちら側でもう一度絞ります。
path = it.get('path', '')
if not path.startswith(f'/{user}/'):
continue
Zenn の記事のパスには、ユーザー名が入っています。そこを見れば、APIの絞り込みが効いていなくても弾けます。パラメータを渡して、返ってきたものを検算する形です。
APIに渡した条件が効いているかどうかは、返ってきたデータからしか分かりません。効いている前提で数えると、今回のような数字が出ます。
件数そのものを、先に疑う
もう1つ足したのが、集計の前に件数を見る処理です。
def sanity(rows):
c = Counter(r['media'] for r in rows)
ng = []
for m in USERS:
n = c.get(m, 0)
if n == 0:
ng.append(f'{m}: 0件。ユーザー名か認証を確認')
elif n > 1000:
ng.append(f'{m}: {n}件。多すぎます。絞り込みが効いていません')
for line in ng:
print('⚠ ' + line, file=sys.stderr)
return not ng
ゼロ件と、多すぎる件数の両方を見ます。上限の1000は根拠のある数字ではありません。自分が1,000本書いていないので、超えたらおかしい、というだけです。
閾値としては雑ですが、今回の壊れ方はこれで両方止まります。dev.to のゼロ件も、Zenn の4,800件も、集計に入る前に警告が出ます。
修正後の実行結果です。
qiita: 93件 取得
devto: 24件 取得
zenn: 69件 取得
全186記事 / 反響合計 634
zenn 69本 反響 381 中央値 1.0 0本 22 上位1本の占有 61.7%
devto 24本 反響 212 中央値 5.5 0本 5 上位1本の占有 10.8%
qiita 93本 反響 41 中央値 0.0 0本 62 ストック 32
手で数えた結果と一致しました。
認証が要るところ、要らないところ
ついでに分かったことを書いておきます。
Qiita の一覧APIは、未認証でも stocks_count を返します。今回32件で、これも手元の把握と合っています。一方 page_views_count は入りません。空のまま返ってきます。PVを取るならトークンが要ります。
dev.to の公開APIも、PVは入りませんでした。自分の記事のPVを取るなら /api/articles/me/published に api-key を付けます。
どちらも、環境変数から読むようにしました。
token = os.environ.get('QIITA_TOKEN')
headers = {'Authorization': f'Bearer {token}'} if token else None
なお、トークンを付けた状態での実行は、まだ試していません。未認証でPVが入らないことだけ確認しています。ここは確かめてから追記します。
Zenn の /api/articles は非公式です。仕様が変わる前提で使っています。今回の「絞り込みが効かないと全件が返る」も、公式に決められた挙動というより、たまたまそうなっているだけかもしれません。だからこそ、返ってきたデータで検算する必要があると思っています。
全体
"""Qiita / Zenn / dev.to の反応を1本で集める。標準ライブラリのみ。
使い方:
python3 collect.py
環境変数（任意）:
QIITA_TOKEN Qiita の PV を取るのに必要
DEVTO_API_KEY dev.to の PV を取るのに必要
"""
import json
import os
import sys
import time
import urllib.error
import urllib.request
from collections import Counter, defaultdict
import statistics
# 媒体ごとにユーザー名が違う。1つの名前を使い回さない。
USERS = {
'qiita': 'Rapls',
'zenn': 'rapls',
'devto': 'rapls',
}
UA = 'article-stats/2.0'
TIMEOUT = 20
def get(url, headers=None):
h = {'User-Agent': UA}
if headers:
h.update(headers)
req = urllib.request.Request(url, headers=h)
for attempt in range(3):
try:
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
return json.loads(r.read().decode('utf-8'))
except urllib.error.HTTPError as e:
if e.code == 429 and attempt < 2:
time.sleep(2 ** attempt)
continue
raise
return None
def qiita(user):
token = os.environ.get('QIITA_TOKEN')
headers = {'Authorization': f'Bearer {token}'} if token else None
out, page = [], 1
while True:
d = get(f'https://qiita.com/api/v2/users/{user}/items'
f'?per_page=100&page={page}', headers)
if not d:
break
for it in d:
out.append({
'media': 'qiita', 'title': it['title'], 'url': it['url'],
'reactions': it.get('likes_count', 0),
'comments': it.get('comments_count', 0),
'stocks': it.get('stocks_count'),
'pv': it.get('page_views_count'),
})
if len(d) < 100:
break
page += 1
time.sleep(1)
return out
def devto(user):
key = os.environ.get('DEVTO_API_KEY')
out, page = [], 1
while True:
if key:
url = ('https://dev.to/api/articles/me/published'
f'?per_page=100&page={page}')
d = get(url, {'api-key': key})
else:
url = (f'https://dev.to/api/articles?username={user}'
f'&per_page=100&page={page}')
d = get(url)
if not d:
break
for it in d:
out.append({
'media': 'devto', 'title': it['title'], 'url': it['url'],
'reactions': it.get('public_reactions_count', 0),
'comments': it.get('comments_count', 0),
'stocks': None,
'pv': it.get('page_views_count'),
})
if len(d) < 100:
break
page += 1
time.sleep(1)
return out
def zenn(user):
out, page = [], 1
while True:
d = get(f'https://zenn.dev/api/articles?username={user}'
f'&order=latest&page={page}')
if not d:
break
arts = d.get('articles', [])
if not arts:
break
for it in arts:
path = it.get('path', '')
# username が効かず全件が返ることがある。パスで必ず絞る。
if not path.startswith(f'/{user}/'):
continue
out.append({
'media': 'zenn', 'title': it['title'],
'url': f'https://zenn.dev{path}',
'reactions': it.get('liked_count', 0),
'comments': it.get('comments_count', 0),
'stocks': None, 'pv': None,
})
nxt = d.get('next_page')
if not nxt:
break
page = nxt
time.sleep(1)
if page > 60:
print('zenn: ページ数が多すぎます。打ち切りました。', file=sys.stderr)
break
return out
def sanity(rows):
"""明らかにおかしい結果を、集計の前に止める。"""
c = Counter(r['media'] for r in rows)
ng = []
for m in USERS:
n = c.get(m, 0)
if n == 0:
ng.append(f'{m}: 0件。ユーザー名（{USERS[m]}）か認証を確認')
elif n > 1000:
ng.append(f'{m}: {n}件。多すぎます。絞り込みが効いていません')
for line in ng:
print('⚠ ' + line, file=sys.stderr)
return not ng
def summarize(rows):
by = defaultdict(list)
for r in rows:
by[r['media']].append(r)
print(f'全{len(rows)}記事 / 反響合計 {sum(r["reactions"] for r in rows)}')
print()
for m, v in sorted(by.items(),
key=lambda x: -sum(r['reactions'] for r in x[1])):
re_ = [r['reactions'] for r in v]
pv = [r['pv'] for r in v if r['pv']]
st = [r['stocks'] for r in v if r['stocks'] is not None]
top1 = max(re_) / sum(re_) * 100 if sum(re_) else 0
line = (f'{m:<7} {len(v):>4}本 反響{sum(re_):>5} '
f'中央値{statistics.median(re_):>5.1f} '
f'0本{sum(1 for x in re_ if x == 0):>4} '
f'上位1本の占有{top1:>5.1f}%')
line += f' PV{sum(pv):>7}' if pv else ' PV(取得なし)'
if st:
line += f' ストック{sum(st):>4}'
print(line)
if __name__ == '__main__':
rows = []
for name, fn in (('qiita', qiita), ('devto', devto), ('zenn', zenn)):
try:
got = fn(USERS[name])
print(f'{name}: {len(got)}件 取得', file=sys.stderr)
rows += got
except Exception as e:
print(f'{name}: 取得できませんでした ({e})', file=sys.stderr)
print(file=sys.stderr)
sanity(rows)
print()
summarize(rows)
with open('stats.json', 'w', encoding='utf-8') as f:
json.dump(rows, f, ensure_ascii=False, indent=2)
依存はありません。Python 3.9 以降なら動くはずです。手元は 3.14 と 3.12 で確認しました。
学んだこと
APIに条件を渡したとき、その条件が効かなかった場合に何が返るかは、媒体ごとに違いました。エラーを返すもの、空を返すもの、条件を無視して全部を返すもの。
このうち、最後がいちばん危ないと思っています。データは返ってくるし、集計も通るし、出てきた数字はもっともらしい形をしています。
条件を渡したら、効いたかどうかを返り値で確かめる。件数が想定と桁違いなら、計算に入る前に止める。今回はこの2つを足しました。
集計スクリプトを書くときは、集計そのものより、この手前のほうが大事だと思うようになりました。
ふだんはraplsworks.comで、WordPressプラグイン開発やClaude Codeまわりのことを書いています。