培训部门想批量看一批文档的写作质量。Grammarly Writing Score API 会给出总分，以及 engagement、correctness、delivery、clarity 四项分数。 这篇做
score_doc.py：先在本地预检（格式、4 MB、10 万字符、30 词），再走官方三步流程（创建请求 → 预签名地址上传 → 轮询），最后写出 CSV。
先看清成品
- 本地预检（真实运行）：
empty.txt和notes.md被跳过；short.txt只有 7 个词，收到警告；memo.txt和memo.docx都是 51 个词，通过。- 用无效凭据访问正式接口（真实运行）：得到 404 Invalid client 和一条明确的排查提示。
- 替身环境：总分 86，四项分数 86/89/86/83，写入 CSV。pytest 7 项通过。
合规提醒：这些 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，字段格式依据官方文档中的示例。
做完之后你会得到什么？
| 成果 | 说明 |
|---|---|
score_doc.py |
预检 + 打分 + CSV；--check-only 只做预检 |
gclient.py |
token、429 退避、401 重新获取 token |
fake_grammarly.py |
本地替身 |
test_score_doc.py |
7 项测试 |
需要准备哪些环境？
| 项目 | 说明 |
|---|---|
| Python | 3.9 以上（box 上是 3.13），pip install requests pytest |
| 凭据 | 管理员创建的 OAuth 凭据，勾选 Writing Score API（scores-api:read 和 scores-api:write） |
| 环境变量 | GRAMMARLY_CLIENT_ID、GRAMMARLY_CLIENT_SECRET |
官方事实
| 事实 | 内容 |
|---|---|
| 获取 token | POST https://auth.grammarly.com/v4/api/oauth2/token，表单参数 grant_type=client_credentials、client_id、client_secret、scope（逗号分隔） |
| 创建请求 | POST /ecosystem/api/v2/scores，请求体 {"filename": ...}，返回 score_request_id 和 file_upload_url |
| 上传 | 用 PUT 上传到预签名地址，不带 Authorization；必须在 120 秒内开始上传 |
| 结果 | GET /v2/scores/{id}，status 为 PENDING、FAILED 或 COMPLETED；分数是 0～1 的小数 |
| 格式 | doc、docx、odt、txt、rtf |
| 限制 | 4,194,304 字节；100,000 字符；少于 30 个词时结果不可靠，有 1 个词就 COMPLETED 但 score 为 null |
| 速率 | 创建请求每秒 10 次，查询每秒 50 次；遇到 429 用指数退避，基数 2 秒 |
| 保留 | 分数保留 30 天；文档最多保留 24 小时 |
来源：Writing Score API 文档。文档列出了失败原因（例如 document_size_exceeds_limit），但没有写明失败原因在响应中的字段名，所以脚本在失败时把整个响应原样写进 note 列。
文件结构
65-grammarly-A-src/
├── score_doc.py
├── gclient.py
├── fake_grammarly.py
├── test_score_doc.py
└── samples/{memo.txt,memo.docx,short.txt,empty.txt,notes.md}
（memo.docx 是我用 zipfile 拼出来的最小文件，只用于本地提取文字的测试，不一定能被 Word 打开。）
共用客户端 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 Writing Score API 给文档打分（需要 Grammarly Enterprise / Education 管理员创建的 OAuth 凭据）。
流程：本地预检 → POST /v2/scores 拿到 score_request_id 和预签名上传地址 → 120 秒内 PUT 上传文件（不带 Authorization）
→ GET /v2/scores/{id} 轮询，直到 COMPLETED 或 FAILED。
官方限制：格式 doc/docx/odt/txt/rtf；最大 4,194,304 字节；最多 100,000 字符；少于 30 个词时结果不可靠，有 1 个词就会 COMPLETED 但 score 为 null。"""
import argparse, csv, os, re, sys, time, zipfile
import requests
from gclient import Client, GrammarlyError
EXT = {".doc", ".docx", ".odt", ".txt", ".rtf"}
MAX_BYTES, MAX_CHARS, MIN_WORDS = 4_194_304, 100_000, 30
def extract_text(path):
ext = os.path.splitext(path)[1].lower()
if ext == ".txt":
return open(path, encoding="utf-8", errors="replace").read()
if ext in (".docx", ".odt"):
part = "word/document.xml" if ext == ".docx" else "content.xml"
with zipfile.ZipFile(path) as z:
xml = z.read(part).decode("utf-8", "replace")
xml = re.sub(r"</(w:p|text:p|text:h)>", "\n", xml)
return re.sub(r"<[^>]+>", "", xml)
return None # .doc / .rtf 本地不解析，只检查大小
def preflight(path):
"""返回 (errors, warnings, stats)。"""
errs, warns, st = [], [], {"bytes": os.path.getsize(path)}
ext = os.path.splitext(path)[1].lower()
if ext not in EXT: errs.append(f"不支持的格式 {ext}（只支持 doc、docx、odt、txt、rtf）")
if st["bytes"] == 0: errs.append("空文件：服务器会返回 file_upload_failed")
if st["bytes"] > MAX_BYTES: errs.append(f"超过 4 MB（{st['bytes']} 字节）：会 FAILED，document_size_exceeds_limit")
text = extract_text(path) if not errs else None
if text is not None:
st["chars"] = len(text); st["words"] = len(text.split())
if st["chars"] > MAX_CHARS: errs.append(f"{st['chars']} 字符超过 100,000：会 FAILED，content_length_exceeds_limit")
if st["words"] == 0: errs.append("没有提取到文字：服务器会返回 text_extraction_failed")
elif st["words"] < MIN_WORDS: warns.append(f"只有 {st['words']} 个词（少于 30），分数不可靠或为 null")
elif not errs:
warns.append("本地不解析 .doc/.rtf 的文字，只检查了大小")
return errs, warns, st
def score(client, path, poll_timeout=300, sleep=time.sleep):
j = client.request("POST", "/v2/scores", json={"filename": os.path.basename(path)}).json()
rid, url = j["score_request_id"], j["file_upload_url"]
with open(path, "rb") as f:
up = requests.put(url, data=f, timeout=60) # 预签名地址：不能带 Authorization
if up.status_code >= 300:
raise GrammarlyError(up.status_code, up.text[:300], "上传失败：预签名地址 120 秒后过期")
t0, delay = time.monotonic(), 1.0
while True:
r = client.request("GET", f"/v2/scores/{rid}").json()
if r["status"] in ("COMPLETED", "FAILED"): return r
if time.monotonic() - t0 > poll_timeout: return dict(r, status="TIMEOUT")
sleep(delay); delay = min(delay * 2, 10)
def main(argv=None):
ap = argparse.ArgumentParser(); ap.add_argument("files", nargs="+"); ap.add_argument("--csv", default="scores.csv")
ap.add_argument("--check-only", action="store_true", help="只做本地预检，不调用 API")
a = ap.parse_args(argv); rows, client = [], None
for p in a.files:
errs, warns, st = preflight(p)
for w in warns: print(f"[WARN] {p}: {w}")
if errs:
for e in errs: print(f"[SKIP] {p}: {e}")
rows.append({"file": p, "status": "SKIPPED", "note": "; ".join(errs)}); continue
if a.check_only:
print(f"[OK] {p}: {st}"); continue
client = client or Client(["scores-api:read", "scores-api:write"])
r = score(client, p); sc = r.get("score") or {}
row = {"file": p, "status": r["status"], **{k: round(v * 100) for k, v in sc.items() if v is not None}}
if r["status"] != "COMPLETED" or not sc: row["note"] = str(r)[:300] # 失败原因字段文档没写清，原样保留
print(row); rows.append(row)
if rows and not a.check_only:
keys = ["file", "status", "general_score", "engagement", "correctness", "delivery", "clarity", "note"]
with open(a.csv, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, keys); w.writeheader(); w.writerows(rows)
print("写入", a.csv)
return 0
if __name__ == "__main__":
try: sys.exit(main())
except GrammarlyError as e:
print("API 错误：", e); sys.exit(1)
"""本地替身服务器（我自己写的，不是 Grammarly 官方的）。字段格式和状态码依据官方文档的示例；
token 接口的“错误凭据返回 404”是对真实接口的实测结果。"""
import json, threading, uuid, re
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
class State:
def __init__(self):
self.calls = []; self.scores = {}; self.uploads = {}; self.polls_needed = 2
self.fail_429 = 0; self.expire_token_once = False; self.tokens = set()
self.users = []; self.invitees = []; self.analytics = []; self.conflict_ids = set()
def make_server(st):
class H(BaseHTTPRequestHandler):
def log_message(self, *a): pass
def _send(self, code, obj=None):
b = b"" if obj is None else json.dumps(obj).encode()
self.send_response(code); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
def _auth(self):
t = self.headers.get("Authorization", "")[7:]
if st.expire_token_once and t in st.tokens:
st.expire_token_once = False; st.tokens.discard(t)
if t not in st.tokens:
self._send(401, {"type": "about:blank", "title": "Unauthorized", "status": 401, "detail": "Invalid JWT token"}); return False
if st.fail_429 > 0:
st.fail_429 -= 1; self._send(429, {"status": 429}); return False
return True
def _body(self):
n = int(self.headers.get("Content-Length") or 0); return self.rfile.read(n)
def _page(self, items, q, maxlim):
lim = int(q.get("limit", [maxlim])[0]); start = int(q.get("cursor", ["0"])[0])
chunk = items[start:start + lim]; more = start + lim < len(items)
return {"data": chunk, "paging": {"next_cursor": str(start + lim) if more else None, "cur_cursor": str(start), "has_more": more, "page_size": len(chunk)}}
def do_POST(self):
u = urlparse(self.path); st.calls.append(("POST", u.path))
if u.path == "/token":
f = parse_qs(self._body().decode())
if f.get("client_secret") != ["good"]:
return self._send(404, {"status": 404, "error": "Not Found", "message": "Invalid client"})
t = uuid.uuid4().hex; st.tokens.add(t); return self._send(200, {"access_token": t, "token_type": "Bearer"})
if not self._auth(): return
if u.path == "/api/v2/scores":
rid = str(uuid.uuid4()); st.scores[rid] = {"polls": 0}
host = self.headers["Host"]
return self._send(200, {"score_request_id": rid, "file_upload_url": f"http://{host}/upload/{rid}?X-Amz-Signature=fake"})
self._send(404, {})
def do_PUT(self):
u = urlparse(self.path); st.calls.append(("PUT", u.path))
m = re.match(r"/upload/(.+)", u.path)
if self.headers.get("Authorization"): return self._send(400, {"error": "presigned URL 不应带 Authorization"})
st.uploads[m.group(1)] = self._body(); self._send(200)
def do_GET(self):
u = urlparse(self.path); q = parse_qs(u.query); st.calls.append(("GET", u.path))
if not self._auth(): return
m = re.match(r"/api/v2/scores/(.+)", u.path)
if m:
rid = m.group(1); s = st.scores[rid]; s["polls"] += 1
if rid not in st.uploads: return self._send(200, {"score_request_id": rid, "status": "FAILED", "updated_at": "2026-10-21T11:00:00Z", "score": None})
if s["polls"] < st.polls_needed: return self._send(200, {"score_request_id": rid, "status": "PENDING", "updated_at": "2026-10-21T11:00:00Z", "score": None})
return self._send(200, {"score_request_id": rid, "status": "COMPLETED", "updated_at": "2026-10-21T11:00:05Z",
"score": {"general_score": 0.86, "engagement": 0.86, "correctness": 0.89, "delivery": 0.86, "clarity": 0.83}})
if u.path == "/api/v2/analytics/users": return self._send(200, self._page(st.analytics, q, 400))
if u.path == "/api/v1/users": return self._send(200, self._page(st.users, q, 250))
if u.path == "/api/v1/invitees": return self._send(200, self._page(st.invitees, q, 250))
self._send(404, {})
def do_DELETE(self):
u = urlparse(self.path); st.calls.append(("DELETE", u.path))
if not self._auth(): return
m = re.match(r"/api/v1/(users|invitees)/(\d+)", u.path)
if not m: return self._send(404, {})
kind, i = m.group(1), int(m.group(2))
if i in st.conflict_ids: return self._send(409, {"status": 409})
lst = st.users if kind == "users" else st.invitees; key = "user_id" if kind == "users" else "id"
hit = [x for x in lst if x[key] == i]
if not hit: return self._send(404, {"status": 404})
if kind == "users" and hit[0].get("is_admin"): return self._send(403, {"status": 403})
lst.remove(hit[0]); self._send(204)
srv = ThreadingHTTPServer(("127.0.0.1", 0), H)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv
def env_for(srv):
base = f"http://127.0.0.1:{srv.server_address[1]}"
return {"GRAMMARLY_TOKEN_URL": base + "/token", "GRAMMARLY_API": base + "/api"}
第 2 步：测试
import os, pytest
import fake_grammarly as F, gclient, score_doc as S
@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")
yield st; srv.shutdown()
def C(**kw): return gclient.Client(["scores-api:read", "scores-api:write"], sleep=lambda s: None, **kw)
def test_happy_path(env, tmp_path):
r = S.score(C(), "samples/memo.txt", sleep=lambda s: None)
assert r["status"] == "COMPLETED" and r["score"]["general_score"] == 0.86
assert [c[0] for c in env.calls] == ["POST", "POST", "PUT", "GET", "GET"] # token, 创建, 上传, 轮询×2
def test_bad_secret_is_404(env, monkeypatch):
monkeypatch.setenv("GRAMMARLY_CLIENT_SECRET", "wrong")
with pytest.raises(gclient.GrammarlyError) as e: S.score(C(), "samples/memo.txt")
assert e.value.status == 404 and "Invalid client" in e.value.body
def test_401_refreshes_token_once(env):
c = C(); c._get_token(); env.expire_token_once = True
assert S.score(c, "samples/memo.txt", sleep=lambda s: None)["status"] == "COMPLETED"
assert sum(1 for c_ in env.calls if c_[1] == "/token") == 2
def test_429_backoff(env):
waits = []; env.fail_429 = 3
c = gclient.Client(["scores-api:read"], sleep=waits.append)
S.score(c, "samples/memo.txt", sleep=lambda s: None)
assert waits == [2, 4, 8]
def test_429_gives_up(env):
env.fail_429 = 99
with pytest.raises(gclient.GrammarlyError) as e: S.score(C(max_429=2), "samples/memo.txt")
assert e.value.status == 429
def test_preflight(tmp_path):
big = tmp_path / "big.txt"; big.write_text("word " * 21000) # 105,000 字符
assert any("100,000" in x for x in S.preflight(str(big))[0])
huge = tmp_path / "huge.rtf"; huge.write_bytes(b"x" * (4_194_304 + 1))
assert any("4 MB" in x for x in S.preflight(str(huge))[0])
assert S.preflight("samples/short.txt")[1] and not S.preflight("samples/short.txt")[0]
assert S.preflight("samples/memo.docx")[2]["words"] == 51
assert S.preflight("samples/empty.txt")[0] and S.preflight("samples/notes.md")[0]
def test_cli_csv(env, tmp_path):
out = tmp_path / "s.csv"
S.main(["samples/memo.txt", "samples/empty.txt", "--csv", str(out)])
t = out.read_text(encoding="utf-8")
assert "COMPLETED,86,86,89,86,83" in t and "SKIPPED" in t
第 3 步：运行
$ python score_doc.py --check-only samples/*
[SKIP] samples/empty.txt: 空文件：服务器会返回 file_upload_failed
[OK] samples/memo.docx: {'bytes': 662, 'chars': 302, 'words': 51}
[OK] samples/memo.txt: {'bytes': 302, 'chars': 302, 'words': 51}
[SKIP] samples/notes.md: 不支持的格式 .md（只支持 doc、docx、odt、txt、rtf）
[WARN] samples/short.txt: 只有 7 个词（少于 30），分数不可靠或为 null
[OK] samples/short.txt: {'bytes': 39, 'chars': 39, 'words': 7}
$ GRAMMARLY_CLIENT_ID=invalid-demo GRAMMARLY_CLIENT_SECRET=invalid python score_doc.py samples/memo.txt
API 错误： HTTP 404: {"timestamp":...,"status":404,"error":"Not Found","message":"Invalid client"} — 检查 Client ID、Secret，以及管理后台给这个凭据勾选的 API 和 scope
$ python -m pytest -q
7 passed
替身环境中，CSV 的内容是 samples/memo.txt,COMPLETED,86,86,89,86,83,。
怎么检查结果？
- 有真实凭据时，先拿一份 30 个词以上的 txt 试一次；状态应该从 PENDING 变为 COMPLETED。
- CSV 中的分数是把 0～1 的小数乘以 100 后四舍五入；原值以 API 返回为准。
- 被 SKIPPED 的文件不会占用 API 调用次数。
调试示例
1. 错误凭据返回 404，而不是 401。 我原本按 OAuth 的惯例，只把 401 当作“凭据错误”。用无效凭据实际请求 token 接口后，返回的是 HTTP 404 Invalid client。如果只看状态码，很容易误以为接口地址写错了。所以 gclient._get_token 把 400、401、404 都当作凭据问题，并附上排查提示；测试 test_bad_secret_is_404 覆盖了这种情况。
2. 替身服务器换不进去。 第一版 gclient.py 在模块导入时就读取 GRAMMARLY_API 环境变量，测试里用 monkeypatch 设置环境变量已经晚了。改为在创建 Client 时读取。
3. 429 退避的验证。 让替身连续返回 3 次 429，记录到的等待时间正好是 [2, 4, 8] 秒，和文档“基数 2 秒”一致；超过 max_429 次后抛出 429 错误，不会无限重试。
常见问题
| 现象 | 原因 | 处理 |
|---|---|---|
| token 接口返回 404 Invalid client | Client ID 或 Secret 错误 | 在管理后台重新复制 |
| 401 Invalid JWT token | token 无效或已过期 | 脚本会自动重新获取一次 token |
status FAILED，document_text_not_found |
120 秒内没有上传文件 | 创建请求后立即上传 |
| COMPLETED 但 score 为 null | 文字太少 | 至少 30 个词 |
document_size_exceeds_limit |
文件超过 4 MB | 拆分文档或去掉图片 |
| 上传返回 403 | 预签名地址过期，或上传时带了 Authorization | 重新创建请求；上传时不要带 Authorization |
模板规则
- 先做本地预检，再调用 API。
- 预签名上传不带 Authorization。
- 分数只作为参考，不用于考核个人。
- 失败时原样保存响应，不猜测字段。
项目资源
| 用途 | 地址 |
|---|---|
| API 文档首页 | https://developer.grammarly.com/ |
| OAuth 2.0 凭据和 scope | https://developer.grammarly.com/oauth-credentials.html |
| 第一个 API 请求 | https://developer.grammarly.com/your-first-api-request.html |
| Writing Score API | https://developer.grammarly.com/writing-score-api.html |
先预检、再上传、再轮询，批量打分既不浪费额度，失败时也能查到原因。