運用の設計
非公式APIは、いつか必ず壊れる。問われるのは、壊れたときに原因が分かるかどうかだ。
切り分けの手順と、備えの実装をまとめます
📋 この記事でわかること
- ステータスコードから原因を特定する手順
- 403が返る典型的な条件の一覧
- 429とレート制限への対処
- 指数バックオフとサーキットブレーカーの実装
- レスポンスを保存して仕様変更を検出する仕組み
- 復旧のためのチェックリスト
昨日まで動いていたスクリプトが、今朝から動かない——非公式APIを使っていれば必ず経験することです。原因は認証切れかもしれないし、フィールド名の変更かもしれないし、単に送りすぎただけかもしれません。重要なのは、壊れないようにすることではなく、壊れたときに素早く原因が分かる作りにしておくことです。この記事では、その設計と実装をまとめます。シリーズの締めくくりとして、これまでの記事で断片的に触れてきたエラー対策を1箇所に集約します。
この記事の位置づけ
- 個別のAPIの使い方は各記事で解説済みです
- ここでは、共通して起きる問題への対処をまとめます
- 基本的な考え方は入門記事で触れています
- 認証まわりの詳細はCookie認証の記事を参照してください
ステータスコード別の原因
まずは全体像です。返ってきたコードで、原因の範囲がかなり絞れます。
| コード | 意味 | 主な原因 | 対処の方向 |
|---|---|---|---|
| 200 | 成功 | — | 中身も確認する |
| 401 | 未認証 | Cookieの失効・未設定 | 認証を取り直す |
| 403 | 拒否 | 権限なし・操作条件を満たさない | 条件を確認する |
| 404 | 存在しない | id/keyの取り違え・パスの誤り | 識別子を確認する |
| 422 | 内容が不正 | 必須フィールドの欠落 | ペイロードを見直す |
| 429 | リクエスト過多 | 間隔が短すぎる | 設計から見直す |
| 500 | サーバーエラー | 送信内容の形式が想定外 | MIME typeなどを確認 |
| 502 / 503 | 一時的な障害 | サーバー側の問題 | 時間をおいて再試行 |
200でも失敗していることがある
もっとも厄介なのがこのケースです。ログインAPIは、reCAPTCHAトークンが無くても2xxを返しますが、セッションCookieは発行されません。「エラーが出ないから成功した」と判断すると、原因不明のまま時間を溶かします。認証処理の後は必ず、目的の結果が得られたかを検証してください。
403が返る典型条件
403はもっとも原因が分かりにくいコードです。「権限がない」と言われても、何の権限が足りないのかは分かりません。実際に遭遇しやすい条件を整理します。
| 操作 | 403になる条件 |
|---|---|
| 記事を下書きに戻す | 有料記事で販売実績がある |
| 記事を下書きに戻す | 有料マガジンに追加済み |
| 記事を下書きに戻す | メンバーシップ限定に設定済み |
| 記事を編集する | 自分の記事ではない |
| コメントを投稿する | X-Note-Client-Code ヘッダーが無い |
| コメントを投稿する | 記事側でコメントが無効化されている |
| マガジンを操作する | 有料マガジンの制約に該当 |
| 画像をアップロードする | 署名の有効期限切れ |
| 各種操作 | アカウントに制限がかかっている |
最後の「アカウント制限」は深刻です。正常だった操作が突然403になる場合、送りすぎによる一時的な制限を疑ってください。この場合、しばらく時間をおいてから、実行頻度を落として再開するのが対処です。
エラーを分類する例外設計
エラーの種類ごとに例外クラスを分けておくと、呼び出し側で対応を変えられます。
note_errors.py
"""
note API 用の例外階層。
呼び出し側が対応を切り替えられるよう、種類ごとに分ける。
"""
class NoteAPIError(Exception):
"""すべてのAPIエラーの基底"""
def __init__(self, message: str, status: int = None,
path: str = "", body: str = ""):
super().__init__(message)
self.status = status
self.path = path
self.body = body[:500] if body else ""
def __str__(self):
base = super().__str__()
if self.status:
return f"[HTTP {self.status}] {base} ({self.path})"
return base
class AuthError(NoteAPIError):
"""認証が通っていない（401）。Cookieの更新が必要"""
class PermissionError_(NoteAPIError):
"""操作が許可されていない（403）。条件を確認する"""
class NotFoundError(NoteAPIError):
"""対象が存在しない（404）。id/keyの取り違えが多い"""
class ValidationError(NoteAPIError):
"""送信内容が不正（422）。必須フィールドを確認する"""
class RateLimitError(NoteAPIError):
"""リクエストが多すぎる（429）。設計を見直す"""
class ServerError(NoteAPIError):
"""サーバー側の問題（5xx）。時間をおいて再試行"""
class SchemaChangedError(NoteAPIError):
"""レスポンス構造が想定と違う。仕様変更の可能性"""
ERROR_MAP = {
401: AuthError,
403: PermissionError_,
404: NotFoundError,
422: ValidationError,
429: RateLimitError,
}
HINTS = {
401: "Cookieが失効しています。ブラウザで取り直してください",
403: "権限がないか、操作が許可されない条件に該当しています",
404: "対象が見つかりません。idとkeyを取り違えていませんか",
422: "送信内容に不備があります。必須フィールドを確認してください",
429: "リクエストが多すぎます。間隔を広げてください",
500: "サーバーエラーです。MIME typeの指定漏れなどを確認してください",
}
def raise_for_status(status: int, path: str, body: str = ""):
"""ステータスコードに応じた例外を投げる"""
if 200 <= status < 300:
return
hint = HINTS.get(status, "想定外のエラーです")
if status >= 500:
raise ServerError(hint, status, path, body)
error_class = ERROR_MAP.get(status, NoteAPIError)
raise error_class(hint, status, path, body)
if __name__ == "__main__":
for code in (401, 403, 404, 422, 429, 500):
try:
raise_for_status(code, "/api/v3/notes/nXXXX")
except NoteAPIError as e:
print(f"{type(e).__name__}: {e}")リトライとサーキットブレーカー
一時的な障害には再試行が有効ですが、無条件のリトライは危険です。失敗が続いているのに叩き続けると、状況を悪化させます。
resilient_client.py
"""
リトライとサーキットブレーカーを備えたクライアント。
サーキットブレーカー = 失敗が続いたら、一定時間リクエストを止める仕組み。
壊れている相手を叩き続けないための安全装置。
"""
import logging
import random
import time
import requests
from note_errors import (raise_for_status, NoteAPIError, AuthError,
RateLimitError, ServerError)
logger = logging.getLogger(__name__)
BASE_URL = "https://note.com"
DEFAULT_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36")
# サーキットブレーカーの設定
FAILURE_THRESHOLD = 5 # 連続failure回数
COOLDOWN_SEC = 300 # 開いている時間（5分）
class CircuitBreaker:
"""連続失敗したら、一定時間リクエストを止める"""
def __init__(self, threshold: int = FAILURE_THRESHOLD,
cooldown: float = COOLDOWN_SEC):
self.threshold = threshold
self.cooldown = cooldown
self.failures = 0
self.opened_at = 0.0
@property
def is_open(self) -> bool:
if self.failures < self.threshold:
return False
elapsed = time.time() - self.opened_at
if elapsed > self.cooldown:
logger.info("クールダウンが終わりました。再開します")
self.reset()
return False
return True
def record_success(self):
if self.failures:
logger.info("成功しました。失敗カウントをリセットします")
self.failures = 0
def record_failure(self):
self.failures += 1
if self.failures == self.threshold:
self.opened_at = time.time()
logger.error(
"連続%s回失敗しました。%.0f秒間リクエストを停止します",
self.threshold, self.cooldown,
)
def reset(self):
self.failures = 0
self.opened_at = 0.0
def remaining(self) -> float:
if not self.is_open:
return 0.0
return max(0.0, self.cooldown - (time.time() - self.opened_at))
class ResilientClient:
def __init__(self, session_cookie: str = None,
interval: float = 1.2, max_retry: int = 3):
self.session = requests.Session()
self.session.headers.update({
"User-Agent": DEFAULT_UA,
"Accept": "application/json",
})
if session_cookie:
self.session.cookies.set(
"_note_session_v5", session_cookie.strip(),
domain="note.com",
)
self.interval = interval
self.max_retry = max_retry
self.breaker = CircuitBreaker()
self._last_request_at = 0.0
def _wait(self):
elapsed = time.time() - self._last_request_at
if elapsed < self.interval:
time.sleep(self.interval - elapsed)
self._last_request_at = time.time()
def _backoff(self, attempt: int, base: float = 1.5):
"""指数バックオフ（ジッター付き）"""
wait = base * (2 ** (attempt - 1)) + random.uniform(0, 1.5)
logger.info("%.1f 秒待機します", wait)
time.sleep(wait)
def request(self, method: str, path: str, **kwargs):
if self.breaker.is_open:
raise NoteAPIError(
f"サーキットブレーカーが開いています"
f"（あと {self.breaker.remaining():.0f} 秒）"
)
url = f"{BASE_URL}{path}"
kwargs.setdefault("timeout", 20)
last_error = None
for attempt in range(1, self.max_retry + 1):
self._wait()
try:
res = self.session.request(method, url, **kwargs)
except requests.RequestException as e:
logger.warning("通信エラー(%s/%s): %s",
attempt, self.max_retry, e)
last_error = e
self._backoff(attempt)
continue
# 認証エラーはリトライしても無駄
if res.status_code == 401:
self.breaker.record_failure()
raise AuthError(
"認証が切れています。Cookieを更新してください",
401, path, res.text,
)
# レート制限は長めに待つ
if res.status_code == 429:
logger.warning("レート制限(429)")
last_error = RateLimitError(
"リクエストが多すぎます", 429, path
)
self._backoff(attempt, base=10.0)
continue
# サーバーエラーは再試行の価値がある
if res.status_code >= 500:
logger.warning("サーバーエラー(%s)", res.status_code)
last_error = ServerError(
"サーバー側の問題", res.status_code, path
)
self._backoff(attempt)
continue
# それ以外はリトライしない
self.breaker.record_success()
return res
self.breaker.record_failure()
if isinstance(last_error, NoteAPIError):
raise last_error
raise NoteAPIError(
f"リトライ上限に達しました: {method} {path}"
)
def get_json(self, path: str, params: dict = None) -> dict:
res = self.request("GET", path, params=params)
raise_for_status(res.status_code, path, res.text)
try:
return res.json().get("data", {})
except ValueError:
raise NoteAPIError("JSONとして解釈できません", path=path)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
client = ResilientClient()
try:
data = client.get_json("/api/v3/notes/n6a10366298b0")
print("タイトル:", data.get("name"))
except NoteAPIError as e:
print("エラー:", e)リトライすべきエラー・すべきでないエラー
認証エラー（401）は、何度試しても結果が変わりません。リトライして意味があるのは、時間経過で解消する可能性があるものだけです。429とサーバーエラーは待てば直る可能性がありますが、401・403・404・422はコード側かデータ側の問題なので、即座に諦めて原因を報告するほうが親切です。
▶ 認証まわりの詳細は、専用の記事で解説しています。
Cookieの取得から失効検知までの実装です。
仕様変更を検出する
非公式APIで最も気づきにくいのが、レスポンス構造の変更です。エラーにならず、値が取れなくなるだけなので発見が遅れます。
schema_watcher.py
"""
レスポンスの構造を記録し、変化を検出する。
仕様変更に早く気づくための仕組み。
"""
import json
import os
from datetime import datetime
SCHEMA_DIR = "schemas"
def extract_schema(obj, depth: int = 0, max_depth: int = 3):
"""
値ではなく「構造」を抽出する。
{"name": "記事名", "price": 500} → {"name": "str", "price": "int"}
"""
if depth > max_depth:
return "..."
if isinstance(obj, dict):
return {
k: extract_schema(v, depth + 1, max_depth)
for k, v in sorted(obj.items())
}
if isinstance(obj, list):
if not obj:
return ["empty"]
return [extract_schema(obj[0], depth + 1, max_depth)]
return type(obj).__name__
def schema_path(name: str) -> str:
safe = name.replace("/", "_").strip("_")
return os.path.join(SCHEMA_DIR, f"{safe}.json")
def save_schema(name: str, data: dict):
"""現在の構造を保存する"""
os.makedirs(SCHEMA_DIR, exist_ok=True)
record = {
"saved_at": datetime.now().isoformat(),
"schema": extract_schema(data),
}
with open(schema_path(name), "w", encoding="utf-8") as f:
json.dump(record, f, ensure_ascii=False, indent=2)
def compare_schema(name: str, data: dict) -> list:
"""
保存済みの構造と比較して、差分を返す。
初回は保存だけして空リストを返す。
"""
path = schema_path(name)
if not os.path.exists(path):
save_schema(name, data)
print(f" 構造を記録しました: {name}")
return []
with open(path, encoding="utf-8") as f:
saved = json.load(f)
old = saved.get("schema", {})
new = extract_schema(data)
diffs = []
walk_diff(old, new, "", diffs)
return diffs
def walk_diff(old, new, prefix: str, diffs: list):
"""構造を再帰的に比較する"""
if isinstance(old, dict) and isinstance(new, dict):
for key in old:
path = f"{prefix}.{key}" if prefix else key
if key not in new:
diffs.append(f"消えた: {path}")
else:
walk_diff(old[key], new[key], path, diffs)
for key in new:
if key not in old:
path = f"{prefix}.{key}" if prefix else key
diffs.append(f"増えた: {path}")
return
if isinstance(old, list) and isinstance(new, list):
if old and new:
walk_diff(old[0], new[0], f"{prefix}[]", diffs)
return
if old != new:
diffs.append(f"型が変わった: {prefix} ({old} → {new})")
def check(name: str, data: dict) -> bool:
"""構造をチェックする。変化があればFalse"""
diffs = compare_schema(name, data)
if not diffs:
return True
print(f"\n【仕様変更の可能性】{name}")
for d in diffs[:15]:
print(f" - {d}")
print(" → コードの修正が必要かもしれません")
return False
if __name__ == "__main__":
from resilient_client import ResilientClient
client = ResilientClient()
targets = [
("notes_detail", "/api/v3/notes/n6a10366298b0"),
("hashtag", "/api/v2/hashtags/Python"),
]
print("=== レスポンス構造をチェックします ===")
for name, path in targets:
try:
data = client.get_json(path)
ok = check(name, data)
if ok:
print(f" {name}: 変化なし")
except Exception as e:
print(f" {name}: 取得できません（{e}）")この仕組みを日次で回しておけば、スクリプトが壊れる前に仕様変更に気づけます。値ではなく構造だけを保存しているので、記事の内容が変わっても差分は出ません。
ログの設計
問題が起きたときに追える情報が残っていなければ、原因の特定はできません。
logging_setup.py
"""
ログの設定をまとめる。
- ファイルは日付ごとにローテーション
- コンソールには重要なものだけ
- 認証情報は絶対に出さない
"""
import logging
import os
import re
from logging.handlers import TimedRotatingFileHandler
LOG_DIR = "logs"
# 認証情報が混ざったら伏せ字にする
SENSITIVE_PATTERNS = [
(re.compile(r"(_note_session_v5=)[\w\-%.]+"), r"\1***"),
(re.compile(r"(X-Note-Client-Code[\"']?\s*[:=]\s*[\"']?)[0-9a-f]{16,}"),
r"\1***"),
(re.compile(r"(password[\"']?\s*[:=]\s*[\"']?)[^\s,\"']+"), r"\1***"),
]
class MaskingFilter(logging.Filter):
"""ログから認証情報を除去するフィルタ"""
def filter(self, record):
message = str(record.getMessage())
for pattern, replacement in SENSITIVE_PATTERNS:
message = pattern.sub(replacement, message)
record.msg = message
record.args = ()
return True
def setup(name: str = "note", level: int = logging.INFO,
console_level: int = logging.WARNING) -> logging.Logger:
"""ロガーを設定して返す"""
os.makedirs(LOG_DIR, exist_ok=True)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
# 既存のハンドラを消す（重複防止）
logger.handlers.clear()
formatter = logging.Formatter(
"%(asctime)s [%(levelname)-7s] %(name)s: %(message)s"
)
# ファイル: 日次ローテーション、7世代
file_handler = TimedRotatingFileHandler(
os.path.join(LOG_DIR, f"{name}.log"),
when="midnight",
backupCount=7,
encoding="utf-8",
)
file_handler.setLevel(level)
file_handler.setFormatter(formatter)
file_handler.addFilter(MaskingFilter())
# コンソール: 警告以上のみ
console = logging.StreamHandler()
console.setLevel(console_level)
console.setFormatter(formatter)
console.addFilter(MaskingFilter())
logger.addHandler(file_handler)
logger.addHandler(console)
return logging.getLogger(name)
if __name__ == "__main__":
log = setup("test")
log.info("通常のログ")
log.warning("警告のログ")
log.error("Cookie: _note_session_v5=abcdef123456789 を使用")
log.info("logs/test.log を確認してください（伏せ字になっているはず）")ログに認証情報を出さない
デバッグのつもりでリクエストヘッダーをそのままログに出すと、Cookieがファイルに残ります。ログファイルは意外と共有されやすく、流出経路になります。この記事のフィルタのように、出力前に伏せ字化する仕組みを入れておくのが確実です。
▼ RECOMMENDATION ▼
保守が要らない選択肢もある
ここまで読んで分かるとおり、非公式APIを使った仕組みは作って終わりではありません。仕様変更のたびに直す保守作業が続きます。アメブロであれば、アメプレスProがいいね・フォロー・アクセス獲得を担い、仕様変更への対応は提供元が行います。自分でコードを保守する必要がありません。ASPアフィリエイトも使えるため、収益化まで含めた設計ができます。月額2,980円、365日LINEサポート付きです。
✦ 保守は提供元が対応
✦ 自動いいね・フォロー
✦ ASPアフィリOK
✦ 365日LINEサポート
アメプレスPro 公式ページを確認する →
※ 当サイトはアフィリエイト広告を含みます。
ヘルスチェックと通知
定期実行の仕組みでは、失敗に気づかないまま止まっていることが最大の問題です。1日1回、状態を確認して通知する仕組みを入れます。
healthcheck.py
"""
API周りの健全性をチェックして、問題があれば通知する。
毎日1回、定期実行する想定。
"""
import os
import sys
import time
from datetime import datetime
import requests
from resilient_client import ResilientClient
from note_errors import NoteAPIError, AuthError
from schema_watcher import check as check_schema
from logging_setup import setup
WEBHOOK_URL = os.getenv("NOTIFY_WEBHOOK", "")
LOG_DIR = "logs"
STALE_HOURS = 30
log = setup("healthcheck")
def notify(message: str):
print(message)
if not WEBHOOK_URL:
return
try:
requests.post(WEBHOOK_URL, json={"content": message}, timeout=10)
except requests.RequestException as e:
log.warning("通知の送信に失敗: %s", e)
def check_auth(client: ResilientClient) -> list:
"""認証が生きているか"""
try:
data = client.get_json("/api/v2/current_user")
if not data.get("id"):
return ["認証: ユーザー情報が空です"]
log.info("認証OK: @%s", data.get("urlname"))
return []
except AuthError:
return ["認証: Cookieが失効しています。更新してください"]
except NoteAPIError as e:
return [f"認証: 確認できません（{e}）"]
def check_endpoints(client: ResilientClient) -> list:
"""主要エンドポイントが応答するか"""
targets = [
("記事取得", "/api/v3/notes/n6a10366298b0"),
("タグ情報", "/api/v2/hashtags/Python"),
]
problems = []
for label, path in targets:
try:
data = client.get_json(path)
if not data:
problems.append(f"{label}: 空のレスポンス")
time.sleep(2.0)
except NoteAPIError as e:
problems.append(f"{label}: {e}")
return problems
def check_schemas(client: ResilientClient) -> list:
"""レスポンス構造に変化がないか"""
targets = [
("notes_detail", "/api/v3/notes/n6a10366298b0"),
]
problems = []
for name, path in targets:
try:
data = client.get_json(path)
if not check_schema(name, data):
problems.append(f"構造変化: {name}")
time.sleep(2.0)
except NoteAPIError:
pass
return problems
def check_logs() -> list:
"""ログが更新されているか（スクリプトが動いているか）"""
if not os.path.isdir(LOG_DIR):
return ["ログディレクトリがありません"]
logs = [
f for f in os.listdir(LOG_DIR) if f.endswith(".log")
]
if not logs:
return ["ログファイルがありません"]
newest = max(
os.path.getmtime(os.path.join(LOG_DIR, f)) for f in logs
)
age_hours = (time.time() - newest) / 3600
if age_hours > STALE_HOURS:
return [f"ログが {age_hours:.0f} 時間更新されていません"]
return []
def main() -> int:
client = ResilientClient(
session_cookie=os.getenv("NOTE_SESSION", "")
)
problems = []
problems += check_auth(client)
problems += check_endpoints(client)
problems += check_schemas(client)
problems += check_logs()
stamp = datetime.now().strftime("%Y-%m-%d %H:%M")
if not problems:
log.info("ヘルスチェック: 問題なし")
print(f"{stamp} 問題ありません")
return 0
message = f"【note API】{stamp}\n" + "\n".join(
f"- {p}" for p in problems
)
log.error("問題を検出: %s件", len(problems))
notify(message)
return 1
if __name__ == "__main__":
sys.exit(main())壊れたときのチェックリスト
実際に動かなくなったとき、上から順に確認してください。多くの場合、最初の3つで解決します。
- ステータスコードを確認する——ログに残っているはずです。401なら認証、429なら送りすぎ、404なら識別子の問題です。
- 認証を確認する——
/api/v2/current_userを叩いて、自分の情報が返るか確認します。返らなければCookieを取り直します。 - ブラウザで同じ操作をしてみる——画面上でできない操作は、APIでもできません。仕様変更や制限を切り分けられます。
- レスポンスを保存して構造を見る——200が返っているのに値が取れない場合は、フィールド名の変更を疑います。
- DevToolsで実際の通信を確認する——ブラウザが送っているリクエストと、自分のコードを比較します。
- 時間をおいて再実行する——一時的な制限やサーバー側の問題であれば、これで解決します。
ブラウザでの確認が最強の切り分け
手順3が意外と効きます。画面上で同じ操作をしてみれば、APIの問題なのか、アカウントや記事側の条件の問題なのかがすぐ分かります。たとえば有料記事を下書きに戻そうとして403が出た場合、画面上でも同じくできないなら、それは仕様です。コードを直しても解決しません。
壊れにくい書き方
最後に、そもそも壊れにくくするための設計指針をまとめます。
| 指針 | 理由 |
|---|---|
| フィールド取得は必ず .get() を使う | キーが消えてもクラッシュしない |
| 複数のキー名候補を試す | camelCase / snake_case の揺れに対応 |
| 処理を取得・変換・保存に分ける | 壊れた箇所だけ直せる |
| ループには必ず上限を設ける | 無限ループとリクエスト暴走を防ぐ |
| 投稿系は dry_run を既定にする | 誤実行による事故を防ぐ |
| 正常時のレスポンスを保存しておく | 壊れたときに差分を比較できる |
| 手作業でもできる状態を保つ | 自動化が止まっても運用が続く |
| 業務の根幹に据えない | 非公式APIは保証がない |
最後の項目が最も重要です。非公式APIは、いつ使えなくなってもおかしくありません。それが止まると仕事が回らない、という状態にしないでください。あくまで手作業を減らす補助として位置づけるのが、現実的な距離感です。
エラーの傾向を集計する
ログが溜まってきたら、どのエラーが多いかを集計してみてください。頻出するエラーには、たいてい構造的な原因があります。
error_stats.py
"""
ログファイルからエラーの傾向を集計する。
どのエラーが多いかが分かれば、対策の優先順位が決まる。
"""
import os
import re
from collections import Counter, defaultdict
from datetime import datetime
LOG_DIR = "logs"
# ログ行から情報を拾うパターン
TIMESTAMP = re.compile(r"^(\d{4}-\d{2}-\d{2})\s+(\d{2}):\d{2}:\d{2}")
STATUS = re.compile(r"HTTP\s+(\d{3})")
LEVEL = re.compile(r"\[(DEBUG|INFO|WARNING|ERROR|CRITICAL)\s*\]")
PATH = re.compile(r"(/api/v\d/[\w/{}.-]+)")
HINTS = {
"401": "Cookieの更新頻度を上げる。失効検知の通知を入れる",
"403": "操作条件を確認する。制限がかかっていないか疑う",
"404": "id と key の使い分けを見直す",
"422": "必須フィールドの送信漏れを確認する",
"429": "リクエスト間隔を広げる。1回の処理件数を減らす",
"500": "送信データの形式（MIME typeなど）を確認する",
}
def read_logs(directory: str = LOG_DIR) -> list:
"""ログディレクトリ内の全行を読む"""
if not os.path.isdir(directory):
return []
lines = []
for filename in sorted(os.listdir(directory)):
if not filename.endswith(".log") and ".log." not in filename:
continue
path = os.path.join(directory, filename)
try:
with open(path, encoding="utf-8", errors="replace") as f:
lines.extend(f.readlines())
except OSError as e:
print(f"読めません（{filename}）: {e}")
return lines
def analyze(lines: list):
if not lines:
print("ログがありません")
return
by_status = Counter()
by_level = Counter()
by_date = defaultdict(Counter)
by_path = Counter()
by_hour = Counter()
for line in lines:
level_match = LEVEL.search(line)
level = level_match.group(1) if level_match else ""
if level:
by_level[level] += 1
time_match = TIMESTAMP.match(line)
date = time_match.group(1) if time_match else ""
hour = time_match.group(2) if time_match else ""
status_match = STATUS.search(line)
if status_match:
code = status_match.group(1)
by_status[code] += 1
if date:
by_date[date][code] += 1
if hour:
by_hour[hour] += 1
path_match = PATH.search(line)
if path_match:
by_path[path_match.group(1)] += 1
print(f"=== ログ {len(lines):,} 行を解析 ===\n")
print("--- ログレベル別 ---")
for level in ("CRITICAL", "ERROR", "WARNING", "INFO"):
count = by_level.get(level, 0)
if count:
print(f" {level:<9} {count:>6,}")
if not by_status:
print("\nHTTPステータスの記録が見つかりませんでした")
return
print("\n--- ステータスコード別 ---")
total = sum(by_status.values())
for code, count in by_status.most_common():
pct = count / total * 100
bar = "|" * int(min(pct / 2, 40))
print(f" {code} {count:>5,} 件 ({pct:>5.1f}%) {bar}")
print("\n--- エラーが多いパス ---")
for path, count in by_path.most_common(10):
print(f" {count:>5,} {path}")
if by_hour:
print("\n--- 時間帯別のエラー数 ---")
for hour in sorted(by_hour):
count = by_hour[hour]
bar = "|" * int(min(count, 40))
print(f" {hour}時 {count:>4} {bar}")
if by_date:
print("\n--- 日別（直近14日）---")
for date in sorted(by_date)[-14:]:
counts = by_date[date]
summary = " ".join(
f"{code}:{n}" for code, n in counts.most_common(4)
)
print(f" {date} {summary}")
print("\n=== 対策の優先順位 ===")
for code, count in by_status.most_common(5):
if code.startswith("2"):
continue
hint = HINTS.get(code, "原因を個別に確認してください")
print(f" [{code}] {count} 件")
print(f" → {hint}")
def main():
lines = read_logs()
analyze(lines)
print(f"\n解析日時: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
if __name__ == "__main__":
main()時間帯別の集計が有効なこともあります。特定の時刻にエラーが集中している場合、複数のスクリプトが同時に走っている可能性があります。実行時刻をずらすだけで解決するケースです。
まとめ：壊れる前提で設計する
このシリーズを通じて扱ってきたAPIは、すべて非公式です。ドキュメントもサポートもなく、予告なく変わります。それでも役に立つのは、壊れたときの備えができている場合だけです。
この記事の要点
- ステータスコードで原因の範囲が絞れる
- 200でも失敗していることがある。結果を検証する
- 403は有料記事・販売実績・権限の条件を疑う
- リトライすべきなのは429と5xxのみ
- 連続失敗したら止まるサーキットブレーカーを入れる
- レスポンスの構造を記録し、仕様変更を検出する
- ログに認証情報を残さない。伏せ字化する
- 非公式APIを業務の根幹に据えない
ヘルスチェックのスクリプトを1日1回動かしておくだけでも、状況は大きく変わります。壊れてから慌てるのではなく、壊れたことにすぐ気づける状態を作ってください。
FAQ｜note APIのエラー対策についてよくある質問
Q. 突然403が返るようになりました
まず、その操作を画面上で試してください。画面でもできないなら仕様や条件の問題です。画面ではできるのにAPIで403が返る場合、送りすぎによる一時的な制限を疑ってください。数時間から1日おいてから、実行頻度を落として再開すると解消することがあります。それでも駄目なら、リクエストの内容をDevToolsで確認したものと比較してください。
Q. 429が返らなければ問題ないですか？
そうとは限りません。429が返るのは明確な超過ですが、その手前でも負荷はかけています。また、429を返さずに黙って制限をかける実装も一般的です。429が出ていないから大丈夫、ではなく、そもそも人間の操作として自然な頻度かで判断してください。1リクエストあたり1〜2秒が目安です。
Q. サーキットブレーカーは必要ですか？
定期実行する仕組みなら入れる価値があります。認証が切れた状態で15分おきに実行され続けると、1日で100回近い失敗リクエストが飛びます。これは相手にとって迷惑ですし、制限を招く原因にもなります。連続失敗したら一定時間止まる仕組みがあれば、こうした暴走を防げます。
Q. 仕様変更にはどれくらいの頻度で遭遇しますか？
予測できません。半年以上変わらないこともあれば、短期間に続けて変わることもあります。だからこそ、頻度を予測するのではなく、変わったときに気づける仕組みを持つほうが確実です。この記事のスキーマ監視を日次で回しておけば、変化に早く気づけます。
Q. エラーが出たとき、どこまでログに残すべきですか？
日時、リクエストのメソッドとパス、ステータスコード、レスポンス本文の冒頭数百文字があれば、大半の問題は追えます。逆に、リクエストヘッダーの全体を出すのは避けてください。Cookieが残ります。この記事のログ設定は、認証情報を自動で伏せ字にする仕組みを入れています。
Q. リトライは何回まで設定すべきですか？
3回程度が適切です。それ以上リトライしても成功する見込みは低く、待ち時間だけが伸びます。指数バックオフを使っていれば、3回で合計10秒前後は待つことになります。それでも駄目なら、時間をおいて再実行するほうが確実です。定期実行の仕組みなら、次回の実行に任せるという判断もあります。
Q. アカウントに制限がかかったか確認できますか？
明示的に知らせるAPIはありません。判断材料としては、これまで通っていた操作が403になる、通常より高い頻度でエラーが出る、画面上でも一部の操作ができないといった症状です。心当たりがある場合は、しばらく自動実行を止めて、手作業での利用に戻すのが賢明です。制限が一時的なものであれば、時間経過で解除されます。
Q. どこまで自動化するのが適切ですか？
読み取り系は積極的に自動化して構いません。投稿系は確認モードを挟んだうえで、自分のアカウント・自分のデータの範囲で。他人への働きかけ（スキ・フォロー・コメント）の機械的な実行は避けてください。判断が必要な操作は自動化せず、候補の提示までに留めるのが安全な設計です。
Q. 仕組みが複雑になりすぎました
非公式APIを使う仕組みは、シンプルなほど保守しやすくなります。この記事で紹介した対策も、すべてを一度に入れる必要はありません。まずログと認証チェックだけを入れて、実際に問題が起きてから対策を足していくほうが現実的です。使っていない機能の保守は、純粋な負債になります。
Q. 結局、非公式APIは使うべきですか？
目的次第です。自分のデータのバックアップや分析であれば、得られるものが大きく、リスクも小さいので使う価値があります。一方、収益や業務の根幹に関わる部分を依存させるのは危険です。いつ止まっても困らない範囲で、手作業を減らす補助として使う——この距離感が適切です。保守の手間を負いたくないなら、既製のツールがある選択肢を選ぶという判断もあります。