Zenn 숨은 통계 API를 매시간 스크랩해 PV를 SQLite에 저장하고 고의로 충돌시키는 파이썬 프로브를 만들었다.
출처: Dev.to
Zenn에 기술 글을 올린다면 이미 겪어봤을 고통을 알고 있을 겁니다. 대시보드에는 페이지 뷰가 표시되지만 공개된 통계 엔드포인트도 없고, 데이터를 내보내는 버튼도 없습니다. 이 프로브를 3주 동안 운영한 결과, 40 KB 정도 되는 SQLite 파일에 각 글별 시간당 PV 시계열 데이터를 확보했고, 글이 500뷰를 돌파하면 Discord 알림을 받으며, 수동 점검은 전혀 필요 없게 되었습니다.
이 글을 끝까지 읽으면 다음을 할 수 있게 됩니다.
1️⃣ Zenn 대시보드가 사용하는 동일한 비공개 JSON 엔드포인트를 호출하기
2️⃣ 문제가 감지되면 경고를 남기고 계속 진행하는 것이 아니라 바로 예외를 발생시켜 종료하는 프로브 작성하기
3️⃣ GitHub Actions에서 실행해 크래시가 발생하면 다음 크론 실행 시 자동으로 재시작되도록 만들기
아래 내용은 모두 복사‑붙여넣기만으로 바로 실행할 수 있습니다.
/api/me/articles 엔드포인트가 HTML 스크래핑보다 뛰어난 이유
첫 번째 버전에서는 BeautifulSoup으로 공개 글 페이지를 파싱해 “❤️ N” 배지를 읽어냈습니다. 9일째에 Zenn이 CSS를 바꾸면서 선택자가 None을 반환했고, KPI 테이블에 2일 동안 0이 채워졌습니다. 숫자는 그럴듯해 보였기 때문에 가장 끔찍한 실패 형태였습니다.
해결책은 HTML 스크래핑을 완전히 포기하는 것이었습니다. DevTools → Network 탭을 열고 Zenn 대시보드를 로드하면 페이지가 https://zenn.dev/api/me/articles?page=1을 요청합니다. 이 JSON에는 각 글마다 page_view_count, liked_count, comments_count, published_at이 포함되어 있어 대시보드가 바로 렌더링하는 필드와 일치합니다. 이 요청에는 세션 쿠키(connect.sid)가 필요하므로 DevTools → Application → Cookies에서 한 번 추출해 비밀 변수로 저장하면 됩니다.
핵심 설계 결정: 이 프로브는 절대 오류를 무시하지 않습니다. 필드 누락, 빈 리스트, HTTP 401, 혹은 이전 실행 이후 PV가 감소한 경우 모두 예외를 발생시킵니다. 프로브는 죽고 재시작되는 것이 목표이며, 가짜 KPI 데이터를 만들어 내며 버티는 것이 아닙니다. 복구 역할은 스케줄러가 담당하고, try/except: pass 같은 코드는 없습니다.
프로브: zenn_probe.py – SQLite에 커밋하거나 바로 종료
아래는 전체 수집기 코드입니다. 모든 페이지를 순회하면서 응답 형태를 검증하고, 실행당 한 번씩 각 글에 대한 행을 기록합니다. 문제가 있으면 예외를 발생시키고, GitHub Actions 작업이 실패 처리되므로 다음 시간대에 깨끗하게 재시도됩니다.
# zenn_probe.py
import os
import sqlite3
import sys
from datetime import datetime, timezone
import requests
ZENN_API = "https://zenn.dev/api/me/articles"
DB_PATH = os.environ.get("KPI_DB", "kpi.sqlite3")
SESSION_COOKIE = os.environ["ZENN_SESSION"] # KeyError on purpose if unset
class ProbeError(RuntimeError):
"""Raised when the response is structurally untrustworthy."""
def fetch_all_articles() -> list[dict]:
headers = {
"Cookie": f"connect.sid={SESSION_COOKIE}",
"User-Agent": "kpi-probe/1.0 (+personal-analytics)",
}
articles, page = [], 1
while True:
resp = requests.get(f"{ZENN_API}?page={page}", headers=headers, timeout=10)
if resp.status_code == 401:
raise ProbeError("401: ZENN_SESSION expired — refresh connect.sid")
resp.raise_for_status()
payload = resp.json()
batch = payload.get("articles")
if batch is None:
raise ProbeError(f"no 'articles' key on page {page}: {list(payload)[:5]}")
if not batch:
break
articles.extend(batch)
if not payload.get("next_page"):
break
page += 1
if page > 50: # guard against an infinite next_page loop
raise ProbeError("pagination exceeded 50 pages — aborting")
if not articles:
raise ProbeError("zero articles returned — auth or API shape changed")
return articles
def validate(article: dict) -> dict:
required = ("id", "slug", "title", "page_view_count", "liked_count")
missing = [k for k in required if k not in article]
if missing:
raise ProbeError(f"missing fields {missing} on {article.get('slug', '?')}")
pv = article["page_view_count"]
if not isinstance(pv, int) or pv None:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS pv_history (
slug TEXT NOT NULL,
title TEXT NOT NULL,
page_views INTEGER NOT NULL,
likes INTEGER NOT NULL,
captured_at TEXT NOT NULL,
PRIMARY KEY (slug, captured_at)
)
"""
)
def last_pv(conn: sqlite3.Connection, slug: str) -> int | None:
row = conn.execute(
"SELECT page_views FROM pv_history WHERE slug=? "
"ORDER BY captured_at DESC LIMIT 1",
(slug,),
).fetchone()
return row[0] if row else None
def commit_run() -> None:
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
articles = [validate(a) for a in fetch_all_articles()]
conn = sqlite3.connect(DB_PATH)
try:
init_db(conn)
for a in articles:
prev = last_pv(conn, a["slug"])
pv = a["page_view_count"]
# Zenn PV is monotonic. A drop means cache/garbage — fail loud.
if prev is not None and pv {pv}, refusing