自分の書いたものを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まわりのことを書いています。