Grammarly：许可证数量有限，有人离职或长期不用，过期的邀请也一直占着名额

许可证数量有限，有人离职或长期不用，过期的邀请也一直占着名额。这篇做 license_plan.py：

• 第一步只读，生成 plan.csv，列出长期不活动的非管理员用户和 EXPIRED 邀请；
• 人工审核后，带上 --apply --confirm 行数 才会删除；
• 删除时，409（属于其他机构）跳过，404（已被删除）记录下来，403（权限不足）立即停止。

先看清成品

• 替身环境：6 个用户、2 个邀请，计划移除 4 个；管理员和保留名单中的人被跳过。
• 执行时：2 个删除成功，1 个过期邀请删除成功，属于其他机构的用户返回 409，被跳过。
• 用无效凭据访问正式接口（真实运行）：404 Invalid client，没有生成 plan.csv。pytest 9 项通过。

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

风险提醒：删除许可证会让对方立即失去企业版功能。先和 HR 或部门负责人核对名单，再执行。

运行说明：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，字段格式依据官方文档中的示例。

做完之后你会得到什么？

需要准备哪些环境？

官方事实

来源：License Management API 文档。

文件结构

65-grammarly-C-src/
├── license_plan.py
├── gclient.py
├── fake_grammarly.py
└── test_license_plan.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 许可证清理：先生成计划（默认只读），人工审核 plan.csv 后，再用 --apply 执行。
依据 License Management API：
- GET /v1/users（limit 1～250）返回 user_id、email、last_activity_at、is_admin；暂不支持移除管理员许可证
- GET /v1/invitees 中 status=EXPIRED 的邀请“在被移除前仍占用一个许可证”
- DELETE /v1/users/{id} 和 /v1/invitees/{id}：成功返回 204；403 表示权限不足；404 表示不存在；409 表示用户属于其他机构，不能移除
需要 scope：计划阶段 users-api:read，执行阶段另加 users-api:write。"""
import argparse, csv, datetime as dt, re, sys
from gclient import Client, GrammarlyError

PLAN_FIELDS = ["kind", "id", "email", "reason"]

def parse_ts(s):
    """官方示例中有 3 位（.834Z）和 9 位（.588388505Z）小数秒；Python 3.11 之前的 fromisoformat 只接受 3 位或 6 位，所以统一截成 6 位。"""
    if not s: return None
    s = re.sub(r"\.(\d{1,6})\d*", lambda m: "." + m.group(1).ljust(6, "0"), s.replace("Z", "+00:00"))
    return dt.datetime.fromisoformat(s)

def build_plan(users, invitees, inactive_days, now=None, keep=()):
    now = now or dt.datetime.now(dt.timezone.utc); cutoff = now - dt.timedelta(days=inactive_days)
    keep = {k.lower() for k in keep}; plan, skipped = [], []
    for u in users:
        if (u.get("email") or "").lower() in keep: skipped.append((u["email"], "在保留名单中")); continue
        if u.get("is_admin"): skipped.append((u["email"], "管理员（API 不支持移除）")); continue
        last = parse_ts(u.get("last_activity_at"))
        if last is None or last < cutoff:
            plan.append({"kind": "user", "id": u["user_id"], "email": u["email"],
                         "reason": f"最后活动 {u['last_activity_at']}，超过 {inactive_days} 天" if last else "没有任何活动记录"})
    for i in invitees:
        if (i.get("email") or "").lower() in keep or i.get("is_admin"): continue
        if i.get("status") == "EXPIRED":
            plan.append({"kind": "invitee", "id": i["id"], "email": i["email"], "reason": f"邀请已过期（创建于 {i.get('created_at')}），仍占用许可证"})
    return plan, skipped

def apply_plan(client, rows):
    out = []
    for r in rows:
        path = f"/v1/{'users' if r['kind'] == 'user' else 'invitees'}/{int(r['id'])}"
        try:
            client.request("DELETE", path); res = "deleted"
        except GrammarlyError as e:
            if e.status == 404: res = "not_found（可能已被删除）"
            elif e.status == 409: res = "skipped_409（属于其他机构）"
            elif e.status == 403:
                out.append(dict(r, result="403 权限不足，已停止")); print("403：凭据缺少 users-api:write 或权限不足，停止执行"); break
            else: res = f"error_{e.status}"
        out.append(dict(r, result=res)); print(r["kind"], r["id"], r["email"], "→", res)
    return out

def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("--inactive-days", type=int, default=90); ap.add_argument("--keep", default="", help="逗号分隔的保留邮箱")
    ap.add_argument("--plan", default="plan.csv"); ap.add_argument("--apply", action="store_true")
    ap.add_argument("--confirm", type=int, help="执行时必须填写 plan.csv 的行数，防止误操作")
    a = ap.parse_args(argv)
    if not a.apply:
        c = Client(["users-api:read"])
        users = list(c.paged("/v1/users", {}, 250)); inv = list(c.paged("/v1/invitees", {}, 250))
        plan, skipped = build_plan(users, inv, a.inactive_days, keep=[k for k in a.keep.split(",") if k])
        with open(a.plan, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, PLAN_FIELDS); w.writeheader(); w.writerows(plan)
        print(f"成员 {len(users)}，邀请 {len(inv)}；计划移除 {len(plan)} 个（用户 {sum(p['kind'] == 'user' for p in plan)}，过期邀请 {sum(p['kind'] == 'invitee' for p in plan)}）")
        for e, why in skipped: print("  跳过", e, why)
        print(f"已写入 {a.plan}。这是演练，没有删除任何内容。审核后运行：--apply --confirm {len(plan)}")
        return 0
    rows = list(csv.DictReader(open(a.plan, encoding="utf-8")))
    if a.confirm != len(rows):
        print(f"--confirm 必须等于 {a.plan} 中的行数 {len(rows)}，未执行"); return 2
    res = apply_plan(Client(["users-api:read", "users-api:write"]), rows)
    with open("result.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, PLAN_FIELDS + ["result"]); w.writeheader(); w.writerows(res)
    print("结果写入 result.csv"); 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, license_plan as L

NOW = d.datetime(2026, 10, 21, tzinfo=d.timezone.utc)
USERS = [{"user_id": 1, "institution_id": 9, "email": "admin@example.com", "name": "A", "last_activity_at": "2025-01-01T00:00:00Z", "is_admin": True},
         {"user_id": 2, "institution_id": 9, "email": "old@example.com", "name": "B", "last_activity_at": "2026-05-01T08:00:00.123Z", "is_admin": False},
         {"user_id": 3, "institution_id": 9, "email": "new@example.com", "name": "C", "last_activity_at": "2026-10-15T00:00:00Z", "is_admin": False},
         {"user_id": 4, "institution_id": 9, "email": "never@example.com", "name": "D", "last_activity_at": None, "is_admin": False},
         {"user_id": 5, "institution_id": 9, "email": "Boss@example.com", "name": "E", "last_activity_at": "2025-01-01T00:00:00Z", "is_admin": False},
         {"user_id": 6, "institution_id": 8, "email": "other@example.com", "name": "F", "last_activity_at": "2025-01-01T00:00:00Z", "is_admin": False}]
INV = [{"id": 70, "email": "x@example.com", "name": "X", "created_at": "2026-01-01T00:00:00Z", "is_admin": False, "status": "EXPIRED"},
       {"id": 71, "email": "y@example.com", "name": "Y", "created_at": "2026-10-01T00:00:00Z", "is_admin": False, "status": "INVITED"}]

def test_plan_rules():
    plan, skipped = L.build_plan(USERS, INV, 90, now=NOW, keep=["boss@example.com"])
    assert [(p["kind"], p["id"]) for p in plan] == [("user", 2), ("user", 4), ("user", 6), ("invitee", 70)]
    assert {e for e, _ in skipped} == {"admin@example.com", "Boss@example.com"}

@pytest.fixture
def env(monkeypatch, tmp_path):
    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.users = [dict(u) for u in USERS] + [{"user_id": 100 + i, "institution_id": 9, "email": f"u{i}@example.com", "name": "", "last_activity_at": "2026-10-20T00:00:00Z", "is_admin": False} for i in range(300)]
    st.invitees = [dict(i) for i in INV]; st.conflict_ids = {6}
    monkeypatch.chdir(tmp_path); yield st; srv.shutdown()

def test_dry_run_deletes_nothing(env):
    assert L.main(["--keep", "boss@example.com"]) == 0
    assert not [c for c in env.calls if c[0] == "DELETE"]
    assert len(list(csv.DictReader(open("plan.csv")))) == 4
    assert sum(1 for c in env.calls if c[1] == "/api/v1/users") == 2                 # 306 个用户，250 + 56

def test_apply_requires_confirm(env):
    L.main(["--keep", "boss@example.com"])
    assert L.main(["--apply"]) == 2 and L.main(["--apply", "--confirm", "3"]) == 2
    assert not [c for c in env.calls if c[0] == "DELETE"]

def test_apply_handles_409_404(env):
    L.main(["--keep", "boss@example.com"])
    env.users = [u for u in env.users if u["user_id"] != 4]                         # 审核期间已被别人删除
    assert L.main(["--apply", "--confirm", "4"]) == 0
    res = {r["id"]: r["result"] for r in csv.DictReader(open("result.csv"))}
    assert res["2"] == "deleted" and res["4"].startswith("not_found") and res["6"].startswith("skipped_409") and res["70"] == "deleted"

def test_403_stops(env):
    rows = [{"kind": "user", "id": "1", "email": "admin@example.com", "reason": "t"}, {"kind": "user", "id": "2", "email": "old@example.com", "reason": "t"}]
    import gclient
    out = L.apply_plan(gclient.Client(["users-api:write"], sleep=lambda s: None), rows)
    assert len(out) == 1 and "403" in out[0]["result"] and any(u["user_id"] == 2 for u in env.users)

@pytest.mark.parametrize("s", ["2024-10-09T23:00:35.834Z", "2024-04-02T17:09:20.588388505Z", "2024-04-02T17:09:20Z", "2024-04-02T17:09:20.5Z"])
def test_parse_ts_variants(s):
    assert L.parse_ts(s).tzinfo is not None

第 3 步：运行（替身环境）

$ python license_plan.py --keep boss@example.com
成员 6，邀请 2；计划移除 4 个（用户 3，过期邀请 1）
  跳过 admin@example.com 管理员（API 不支持移除）
  跳过 Boss@example.com 在保留名单中
已写入 plan.csv。这是演练，没有删除任何内容。审核后运行：--apply --confirm 4

$ cat plan.csv
kind,id,email,reason
user,2,old@example.com,最后活动 2026-05-01T08:00:00.123Z，超过 90 天
user,4,never@example.com,没有任何活动记录
user,6,other@example.com,最后活动 2025-01-01T00:00:00Z，超过 90 天
invitee,70,x@example.com,邀请已过期（创建于 2026-01-01T00:00:00Z），仍占用许可证

$ python license_plan.py --apply --confirm 4
user 2 old@example.com → deleted
user 4 never@example.com → deleted
user 6 other@example.com → skipped_409（属于其他机构）
invitee 70 x@example.com → deleted
结果写入 result.csv

$ GRAMMARLY_CLIENT_ID=invalid-demo GRAMMARLY_CLIENT_SECRET=invalid python license_plan.py     # 正式接口，真实运行
API 错误： HTTP 404: {... "message":"Invalid client"} — ...
$ python -m pytest -q
9 passed

（替身输出中的第 4 行原本显示为“最后活动 无记录，超过 90 天”，读起来不通，已改为“没有任何活动记录”；上面是修改后的结果。）

怎么检查结果？

• plan.csv 中不应该出现管理员或保留名单中的人。
• --confirm 的数字和 plan.csv 行数不一致时，脚本拒绝执行（测试 test_apply_requires_confirm）。
• 执行后对照管理后台的已用席位数。

调试示例

1. 时间戳格式不统一。 官方示例中，last_activity_at 是 3 位小数（.834Z），而 Writing Score 的 updated_at 是 9 位（.588388505Z）。 box 上的 Python 3.13 两种都能解析（9 位会被截断到微秒）。但按 Python 文档，3.11 之前的 fromisoformat 只接受 3 位或 6 位小数，也不接受 Z。box 上只有 3.13，我没有在旧版本上实测，所以 parse_ts 统一截成 6 位并替换 Z，测试覆盖了 0、1、3、9 位四种情况。

2. 文档中的接口地址不一致。 汇总接口的标题是 /v1/institutions-summary，示例却少了 /v1。我无法用真实凭据确认哪个正确，所以本工具不调用它；如果你需要，两个都试一下，以返回 200 的为准。

3. 计划阶段放进了其他机构的用户。 测试数据中 user 6 属于另一个机构，计划阶段仍会把它列出来，因为本地无法确定哪个机构是“本机构”。执行时，服务器返回 409，脚本记为跳过，不会中断。

常见问题

模板规则

• 默认只读；删除需要显式开关加行数确认。
• 执行时只读取审核过的 plan.csv，不重新计算名单。
• 403 立即停止，409 和 404 记录后继续。
• 保留 result.csv 作为审计记录。

项目资源

先演练、再审核、最后执行，许可证清理就不会误删同事。