管理员想每月看一次：谁在用 Grammarly、用了多少次、改进的会话占多少

管理员想每月看一次：谁在用 Grammarly、用了多少次、改进的会话占多少。这篇做 analytics_report.py： 先在本地校验日期范围（最早一年前、最晚两天前），然后按游标翻页拉取全部成员，导出 Excel 能直接打开的 CSV，并输出汇总。

先看清成品

• 日期校验（真实运行）：今天是 2026-10-21，date_to 最晚只能是 2026-10-19；2025-10-20 早于一年前，被拒绝；参数有错时不发送任何请求。
• 替身环境：1000 个成员按 400 + 400 + 200 分三页拉取，写入 CSV。
• 修复了一个只在 2 月 29 日才出现的 bug。pytest 9 项通过。

合规提醒：这些 API 只开放给 Grammarly Enterprise 和 Grammarly for Education 机构授权的管理员，凭据在管理后台 Organization → OAuth 2.0 credentials 中创建。Client Secret 只放在环境变量或密钥管理系统里，不提交到仓库，不共享账号或凭据。只给凭据勾选需要的 API 和 scope。成员统计、邮箱属于个人数据，导出前请确认机构的隐私政策，并告知成员；文件只在内部保存。本文不列价格，购买请联系 Grammarly 销售。

运行说明：box 上没有 Grammarly 企业版凭据，所以没有拿到真实结果。真实运行过的部分是用无效凭据访问正式接口（2026-10-21 19:34，UTC+8）：token 接口返回 HTTP 404 {{"status":404,"error":"Not Found","message":"Invalid client"}}；直接用无效 token 调用 scores、analytics、users 接口，都返回 HTTP 401 {{"title":"Unauthorized","status":401,"detail":"Invalid JWT token"}}。成功路径用的是我自己写的本地替身 fake_grammarly.py，字段格式依据官方文档中的示例。

做完之后你会得到什么？

需要准备哪些环境？

官方事实

来源：Analytics API 文档。

文件结构

65-grammarly-B-src/
├── analytics_report.py
├── gclient.py
├── fake_grammarly.py
└── test_analytics.py

共用客户端 gclient.py

三个项目都使用同一个客户端：获取 token；遇到 429 时按 2、4、8 秒指数退避；遇到 401 时重新获取一次 token。

"""Grammarly 企业版 API 的最小客户端（官方 REST 接口，只用 requests）。
- 用 OAuth client_credentials 获取 token（https://auth.grammarly.com/v4/api/oauth2/token，scope 用逗号分隔）
- 429：指数退避，基数 2 秒（官方文档的建议）
- 401：重新获取一次 token 再重试（文档没有写 token 有效期）
- 实测：client_id 或 secret 错误时，token 接口返回的是 HTTP 404 {"message":"Invalid client"}，不是 401"""
import os, time, requests

UA = "binbho-tutorial/1.0"

class GrammarlyError(Exception):
    def __init__(self, status, body, hint=""):
        super().__init__(f"HTTP {status}: {body}{(' — ' + hint) if hint else ''}"); self.status = status; self.body = body

class Client:
    def __init__(self, scopes, client_id=None, client_secret=None, sleep=time.sleep, max_429=5):
        self.scopes = ",".join(scopes); self.cid = client_id or os.environ["GRAMMARLY_CLIENT_ID"]
        self.secret = client_secret or os.environ["GRAMMARLY_CLIENT_SECRET"]
        self.s = requests.Session(); self.s.headers["User-Agent"] = UA; self.token = None
        self.sleep = sleep; self.max_429 = max_429
        self.token_url = os.environ.get("GRAMMARLY_TOKEN_URL", "https://auth.grammarly.com/v4/api/oauth2/token")
        self.api = os.environ.get("GRAMMARLY_API", "https://api.grammarly.com/ecosystem/api")

    def _get_token(self):
        r = self.s.post(self.token_url, data={"grant_type": "client_credentials", "client_id": self.cid,
                                         "client_secret": self.secret, "scope": self.scopes}, timeout=30)
        if r.status_code in (400, 401, 404):
            raise GrammarlyError(r.status_code, r.text, "检查 Client ID、Secret，以及管理后台给这个凭据勾选的 API 和 scope")
        r.raise_for_status(); self.token = r.json()["access_token"]

    def request(self, method, path, **kw):
        if not self.token: self._get_token()
        refreshed, n429 = False, 0
        while True:
            r = self.s.request(method, self.api + path, headers={"Authorization": f"Bearer {self.token}", "Accept": "application/json"}, timeout=30, **kw)
            if r.status_code == 401 and not refreshed:
                self._get_token(); refreshed = True; continue
            if r.status_code == 429 and n429 < self.max_429:
                self.sleep(2 * 2 ** n429); n429 += 1; continue
            if r.status_code >= 400:
                raise GrammarlyError(r.status_code, r.text)
            return r

    def paged(self, path, params, limit):
        """按 paging.next_cursor 翻页，直到 has_more 为 false。"""
        cursor = None
        while True:
            p = dict(params, limit=limit, **({"cursor": cursor} if cursor else {}))
            j = self.request("GET", path, params=p).json()
            yield from j.get("data", [])
            pg = j.get("paging") or {}
            if not pg.get("has_more") or not pg.get("next_cursor"): return
            cursor = pg["next_cursor"]

第 1 步：报表脚本

"""把 Grammarly Analytics API 的成员使用统计导出成 CSV，并给出汇总。
接口：GET /ecosystem/api/v2/analytics/users?date_from=YYYY-MM-DD&date_to=YYYY-MM-DD&limit<=400&cursor=...
官方限制：date_from 最早为一年前，date_to 最晚为两天前。需要 scope analytics-api:read。"""
import argparse, csv, datetime as dt, statistics, sys
from gclient import Client, GrammarlyError

FIELDS = ["id", "name", "email", "days_active", "sessions_count", "sessions_improved_percent", "prompt_count"]

def one_year_ago(today):
    try: return today.replace(year=today.year - 1)
    except ValueError: return today.replace(year=today.year - 1, day=28)     # 2 月 29 日

def check_range(date_from, date_to, today=None):
    today = today or dt.date.today()
    f, t = dt.date.fromisoformat(date_from), dt.date.fromisoformat(date_to)
    errs = []
    if f > t: errs.append("date_from 晚于 date_to")
    if t > today - dt.timedelta(days=2): errs.append(f"date_to 最晚只能是 {today - dt.timedelta(days=2)}（两天前）")
    if f < one_year_ago(today):
        errs.append(f"date_from 不能早于一年前（{one_year_ago(today)}）")
    return errs

def summarize(rows):
    active = [r for r in rows if (r.get("days_active") or 0) > 0]
    return {"members": len(rows), "active": len(active), "inactive": len(rows) - len(active),
            "sessions_total": sum(r.get("sessions_count") or 0 for r in rows),
            "prompt_total": sum(r.get("prompt_count") or 0 for r in rows),
            "median_improved_pct": statistics.median([r["sessions_improved_percent"] for r in active if r.get("sessions_improved_percent") is not None] or [0])}

def main(argv=None, client=None):
    ap = argparse.ArgumentParser(); ap.add_argument("--from", dest="f", required=True); ap.add_argument("--to", dest="t", required=True)
    ap.add_argument("--out", default="analytics.csv"); ap.add_argument("--limit", type=int, default=400)
    a = ap.parse_args(argv)
    errs = check_range(a.f, a.t) + ([] if 1 <= a.limit <= 400 else ["limit 必须是 1～400"])
    if errs:
        for e in errs: print("[ERROR]", e)
        return 2
    client = client or Client(["analytics-api:read"])
    rows = list(client.paged("/v2/analytics/users", {"date_from": a.f, "date_to": a.t}, a.limit))
    with open(a.out, "w", newline="", encoding="utf-8-sig") as fh:      # utf-8-sig：Excel 打开中文名不乱码
        w = csv.DictWriter(fh, FIELDS, extrasaction="ignore"); w.writeheader(); w.writerows(rows)
    print(summarize(rows)); print("写入", a.out, len(rows), "行")
    return 0

if __name__ == "__main__":
    try: sys.exit(main())
    except GrammarlyError as e:
        print("API 错误：", e); sys.exit(1)

第 2 步：测试

import csv, datetime as d, pytest
import fake_grammarly as F, gclient, analytics_report as A

T = d.date(2026, 10, 21)
@pytest.mark.parametrize("f,t,ok", [("2026-09-01", "2026-10-19", True), ("2026-09-01", "2026-10-20", False),
    ("2025-10-20", "2026-10-01", False), ("2025-10-21", "2026-10-01", True), ("2026-10-05", "2026-10-01", False)])
def test_range(f, t, ok):
    assert (A.check_range(f, t, T) == []) is ok

def test_leap_day():
    # 调试时发现的 bug：条件表达式优先级写错，2 月 29 日这一天任何 date_from 都会被判为“早于一年前”
    assert A.check_range("2023-03-01", "2023-12-31", d.date(2024, 2, 29)) == []
    assert A.check_range("2023-02-27", "2023-12-31", d.date(2024, 2, 29))

@pytest.fixture
def env(monkeypatch):
    st = F.State(); srv = F.make_server(st)
    for k, v in F.env_for(srv).items(): monkeypatch.setenv(k, v)
    monkeypatch.setenv("GRAMMARLY_CLIENT_ID", "demo"); monkeypatch.setenv("GRAMMARLY_CLIENT_SECRET", "good")
    st.analytics = [{"id": i, "name": f"成员{i}", "email": f"u{i}@example.com", "days_active": i % 4, "sessions_count": i * 3,
                     "sessions_improved_percent": 50 + i % 30, "prompt_count": i % 5} for i in range(1, 1001)]
    yield st; srv.shutdown()

def test_pagination_and_csv(env, tmp_path):
    out = tmp_path / "a.csv"
    assert A.main(["--from", "2026-09-01", "--to", "2026-10-19", "--out", str(out)]) == 0
    rows = list(csv.DictReader(open(out, encoding="utf-8-sig")))
    assert len(rows) == 1000 and rows[0]["name"] == "成员1"
    assert [c for c in env.calls if c[1].endswith("analytics/users")].__len__() == 3     # 400 + 400 + 200

def test_bad_args_no_api_call(env):
    assert A.main(["--from", "2026-09-01", "--to", "2026-10-21", "--limit", "500"]) == 2 and env.calls == []

def test_summary():
    s = A.summarize([{"days_active": 0, "sessions_count": 0}, {"days_active": 3, "sessions_count": 10, "sessions_improved_percent": 40, "prompt_count": 2}])
    assert s == {"members": 2, "active": 1, "inactive": 1, "sessions_total": 10, "prompt_total": 2, "median_improved_pct": 40}

第 3 步：运行

$ python -c "import analytics_report as A, datetime as d; print(A.check_range('2026-09-01','2026-10-20', d.date(2026,10,21)))"
['date_to 最晚只能是 2026-10-19（两天前）']
$ GRAMMARLY_CLIENT_ID=invalid-demo GRAMMARLY_CLIENT_SECRET=invalid python analytics_report.py --from 2026-09-01 --to 2026-10-19
API 错误： HTTP 404: {... "message":"Invalid client"} — 检查 Client ID、Secret，以及管理后台给这个凭据勾选的 API 和 scope
$ python -m pytest -q
9 passed

有真实凭据时运行 python analytics_report.py --from 2026-09-01 --to 2026-09-30，会得到 analytics.csv 和一行汇总，例如 {'members': ..., 'active': ..., ...}。

怎么检查结果？

• CSV 的行数应该等于管理后台显示的成员数。
• 用 Excel 打开时，中文姓名不乱码（文件用 utf-8-sig 编码）。
• inactive 表示所选时间段内 days_active 为 0 的人数。

调试示例

只在 2 月 29 日出现的 bug。 第一版用一行条件表达式计算“一年前”：

if f < today.replace(year=today.year - 1) if not (today.month == 2 and today.day == 29) else today - dt.timedelta(days=365):

我原意是“和一年前的日期比较”，但 Python 把它解析成了 (f < X) if 条件 else Y。平时结果正常；到了 2 月 29 日，表达式的值变成一个日期对象，总是为真，于是任何 date_from 都会被判为“早于一年前”。 我用 2024-02-29 作为“今天”测试时发现了这个问题：2024-03-01 起的范围也报错。改成单独的 one_year_ago() 函数（2 月 29 日回退到前一年的 2 月 28 日），并加了 test_leap_day。

常见问题

模板规则

• 参数在本地校验，错误参数不发请求。
• 翻页以 has_more 和 next_cursor 为准。
• CSV 只在内部共享，不用于考核个人。

项目资源

每月跑一次报表，就能看清 Grammarly 在团队里的实际使用情况。