用 Otter Public API 导出近期会议纪要：分页、限速与错误处理

Otter 的 Public API 可以列出工作区里的会议，并取回摘要、大纲和待办事项。 这篇做 otter_minutes.py：把最近 N 天的会议各生成一份 Markdown 纪要（时间换算成 UTC+8），遵守官方限速，正确处理 429 和 401。

先看清成品

• 用无效 key 真实调用官方接口，得到 401 {"error": "unauthorized"}，脚本给出可读的提示，退出码 1。
• 本地替身上完整跑通：第一次请求 429 → 退避后成功，分页在超出时间范围时停止，生成 2 份纪要。
• 修正了一个误报：官方示例中 process_status 为 null 但数据已存在。pytest 6 项通过。

合规提醒：录音和转写前要告知所有参会者并取得同意（不同地区对录音同意的法律要求不同）；分享文字稿前确认里面没有不该外传的信息。本文只使用 Otter 官方功能：网页/客户端的导出，以及官方 Public API（仅 Enterprise 工作区）。不使用非官方接口或抓包，不共享账号。功能和套餐以 Otter 官方说明为准（本文不列价格）。

运行说明：box 上没有 Otter 账号，也没有 Enterprise API key。我只用一个无效 key 真实调用过 GET https://api.otter.ai/v1/conversations，得到 HTTP 401 {"error": "unauthorized"}（2026-10-21 19:58，UTC+8）。其余接口响应来自我自己写的本地替身（字段结构按官方帮助文章中的示例）。示例文字稿、姓名、联系方式都是虚构的。所有脚本和测试都在 box 上真实运行过（Python 3.13，requests）。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

help.otter.ai 对 curl 返回 403（站点限制），用浏览器可以正常打开。接口字段如有变化，以官方文章为准。

文件结构

70-otter-ai-A-src/
├── otter_minutes.py
├── fake_otter.py
├── demo_local.py
└── test_minutes.py

第 1 步：创建 API key（手动）

Otter 网页版 → Integrations → Developer → 创建 key。把 key 放进环境变量 OTTER_API_KEY，不要写进代码或提交到仓库。（本文没有 Enterprise 账号，没有实际操作这一步。）

第 2 步：主脚本

"""用 Otter.ai Public API（Enterprise 工作区）把最近的会议整理成 Markdown 纪要：摘要、大纲、待办事项。
- 鉴权：Authorization: Bearer <API key>（Integrations → Developer → Create key；每个用户最多 2 个 key）
- GET /v1/conversations：按时间倒序、游标分页，limit 1～100
- GET /v1/conversations/{id}?include=action_items,outline：官方建议只请求需要的字段，不要用 all
- 速率限制：每秒 10 次，超过返回 429；按官方建议使用指数退避
只读取数据，不修改 Otter 中的任何内容。"""
import argparse, datetime as dt, os, re, sys, time
import requests

API = os.environ.get("OTTER_API", "https://api.otter.ai/v1")

class OtterError(Exception): pass

class Client:
    def __init__(self, key=None, sleep=time.sleep, min_interval=0.11, max_retry=5):
        self.s = requests.Session(); self.s.headers.update({"Authorization": f"Bearer {key or os.environ['OTTER_API_KEY']}", "User-Agent": "binbho-tutorial/1.0"})
        self.sleep, self.min_interval, self.max_retry, self.last = sleep, min_interval, max_retry, 0.0
        self.api = os.environ.get("OTTER_API", API)

    def get(self, path, **params):
        for attempt in range(self.max_retry + 1):
            wait = self.min_interval - (time.monotonic() - self.last)      # 主动限速，保持在每秒 10 次以内
            if wait > 0: self.sleep(wait)
            self.last = time.monotonic()
            r = self.s.get(self.api + path, params=params, timeout=30)
            if r.status_code == 429 and attempt < self.max_retry:
                self.sleep(min(2 ** attempt, 30)); continue
            if r.status_code == 401: raise OtterError("401 unauthorized：API key 无效，或工作区不是 Enterprise")
            if r.status_code >= 400: raise OtterError(f"HTTP {r.status_code}: {r.text[:200]}")
            return r.json()
        raise OtterError("连续收到 429，已放弃")

    def conversations(self, since=None, limit=50):
        """倒序遍历；遇到早于 since 的会议就停止翻页。"""
        cursor = None
        while True:
            j = self.get("/conversations", limit=limit, **({"cursor": cursor} if cursor else {}))
            for c in j["data"]:
                if since and parse_ts(c["created_at"]) < since: return
                yield c
            meta = j.get("meta", {})
            if not meta.get("has_more"): return
            cursor = meta["next_cursor"]

def parse_ts(s): return dt.datetime.fromisoformat(s.replace("Z", "+00:00").replace(" ", "T"))

def to_local(s): return parse_ts(s).astimezone(dt.timezone(dt.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")

def minutes_md(conv):
    rel = conv.get("relationships") or {}
    out = [f"# {conv['title']}", "", f"- 时间：{to_local(conv['created_at'])}（UTC+8）", f"- 负责人：{conv['owner']['name']}", f"- Otter 链接：{conv['url']}", ""]
    ps = conv.get("process_status") or {}
    # 官方示例中，process_status 全为 null 时摘要和待办仍然存在，所以 null 不等于“没处理完”：
    # 只有状态明确不是 finished，或者为 null 且对应的数据确实缺失时，才提示
    have = {"abstract_summary": conv.get("abstract_summary"), "action_item": rel.get("action_items"), "outline": rel.get("outline")}
    pending = [k for k, v in ps.items() if (v is not None and v != "finished") or (v is None and not have.get(k))]
    if pending:
        out += [f"> 注意：以下内容 Otter 可能还没有生成：{', '.join(pending)}", ""]
    out += ["## 摘要", "", conv.get("abstract_summary") or "（无）", ""]
    if rel.get("outline"):
        out += ["## 大纲", ""] + [f"### {s['section']}\n" + "\n".join(f"- {t}" for t in s["text"]) + "\n" for s in rel["outline"]]
    out += ["## 待办事项", ""]
    items = rel.get("action_items") or []
    for a in items:
        done = (a.get("status") or {}).get("completed")
        who = (a.get("assignee") or {}).get("name") or "未指定"
        out.append(f"- [{'x' if done else ' '}] {a['text']}（{who}）")
    if not items: out.append("（无）")
    return "\n".join(out) + "\n"

def slug(s): return re.sub(r"[^\w\u4e00-\u9fff-]+", "-", s).strip("-")[:60] or "meeting"

def main(argv=None, client=None):
    ap = argparse.ArgumentParser(); ap.add_argument("--days", type=int, default=7); ap.add_argument("--out", default="minutes")
    a = ap.parse_args(argv); c = client or Client()
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=a.days); os.makedirs(a.out, exist_ok=True); n = 0
    for conv in c.conversations(since):
        full = c.get(f"/conversations/{conv['id']}", include="action_items,outline")["data"]
        path = os.path.join(a.out, f"{parse_ts(full['created_at']).astimezone(dt.timezone(dt.timedelta(hours=8))):%Y%m%d}-{slug(full['title'])}.md")
        open(path, "w", encoding="utf-8").write(minutes_md(full)); n += 1; print("写入", path)
    print(f"共 {n} 场会议（最近 {a.days} 天）")
    return 0

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

第 3 步：本地替身

"""本地替身（我自己写的，不是 Otter 官方的）。字段结构照抄官方 Public API 帮助文章中的示例；401 的响应体是对真实接口的实测结果。"""
import json, threading, datetime as dt
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

OWNER = {"id": "owner_id", "name": "Jane Doe", "first_name": "Jane", "last_name": "Doe", "email": "jane.doe@example.com"}

def conv(i, days_ago, title, finished=True):
    t = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=days_ago)).strftime("%Y-%m-%dT%H:%M:%SZ")
    st = "finished" if finished else None
    return {"id": f"c{i}", "title": title, "url": f"https://otter.ai/u/c{i}", "owner": OWNER, "created_at": t,
            "process_status": {"abstract_summary": st, "action_item": st, "outline": st}, "calendar_guests": None,
            "shared_emails": [], "shared_channels": [], "abstract_summary": f"{title} 的摘要" if finished else None, "conf_join_url": None}

class State:
    def __init__(self):
        self.convs = [conv(1, 1, "Customer Discovery Call – ACME Corp"), conv(2, 3, "周会 / 产品规划", finished=False), conv(3, 20, "旧会议")]
        self.fail_429 = 0; self.calls = []; self.times = []

def make(st):
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def send(self, code, obj):
            b = 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 do_GET(self):
            import time
            u = urlparse(self.path); q = parse_qs(u.query); st.calls.append((u.path, q)); st.times.append(time.monotonic())
            if self.headers.get("Authorization") != "Bearer good": return self.send(401, {"error": "unauthorized"})
            if st.fail_429 > 0: st.fail_429 -= 1; return self.send(429, {"error": "Too Many Requests"})
            meta = {"retrieved_at": "2026-10-21T12:00:00Z"}
            if u.path == "/v1/conversations":
                lim = int(q.get("limit", ["20"])[0])
                if not 1 <= lim <= 100: return self.send(400, {"error": "invalid limit"})
                start = int(q.get("cursor", ["0"])[0]); page = st.convs[start:start + lim]; more = start + lim < len(st.convs)
                meta.update(has_more=more, **({"next_cursor": str(start + lim)} if more else {}))
                return self.send(200, {"meta": meta, "data": page})
            if u.path.startswith("/v1/conversations/"):
                cid = u.path.rsplit("/", 1)[1]; c = next((x for x in st.convs if x["id"] == cid), None)
                if not c: return self.send(404, {"error": "Conversation not found"})
                inc = set(q.get("include", [""])[0].split(","))
                rel = {"action_items": None, "insights": None, "outline": None, "transcript": None, "custom_prompt": None}
                if "action_items" in inc:
                    rel["action_items"] = [{"id": "a1", "text": "Send follow-up email with pricing summary and implementation overview.", "assignee": OWNER, "status": None},
                                           {"id": "a2", "text": "ACME to confirm final user count and preferred contract term.",
                                            "assignee": {"id": "x", "name": "Alex Taylor", "first_name": "Alex", "last_name": "Taylor", "email": "alex.taylor@example.com"},
                                            "status": {"completed": True, "created_at": "2025-07-31T12:00:00Z", "last_modified_at": "2025-07-31T12:00:00Z", "completed_at": "2025-07-31T12:00:00Z"}}]
                if "outline" in inc:
                    rel["outline"] = [{"section": "Introductions and Context", "text": ["Participants introduced themselves and clarified roles."]},
                                      {"section": "Pricing and Rollout Options", "text": ["The team reviewed per-seat and flat-rate pricing options."]}]
                return self.send(200, {"meta": meta, "data": dict(c, relationships=rel)})
            self.send(404, {"error": "not found"})
    srv = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, f"http://127.0.0.1:{srv.server_address[1]}/v1"

第 4 步：演示脚本

"""用本地替身演示一次完整运行（不连接 Otter）。用法：python demo_local.py [key]"""
import os, sys, fake_otter as F
st = F.State(); st.fail_429 = 1                      # 第一次请求返回 429，演示退避
srv, url = F.make(st)
os.environ["OTTER_API"] = url; os.environ["OTTER_API_KEY"] = sys.argv[1] if len(sys.argv) > 1 else "good"
import otter_minutes as M
try: code = M.main(["--days", "7", "--out", "minutes"])
except M.OtterError as e: print("Otter API 错误：", e); code = 1
finally: srv.shutdown()
sys.exit(code)

第 5 步：测试

import os, pytest
import fake_otter as F, otter_minutes as O

@pytest.fixture
def env(monkeypatch):
    st = F.State(); srv, url = F.make(st); monkeypatch.setenv("OTTER_API", url); yield st; srv.shutdown()

def test_minutes(env, tmp_path):
    out = tmp_path / "m"
    O.main(["--days", "7", "--out", str(out)], client=O.Client("good", min_interval=0))
    files = sorted(os.listdir(out)); assert len(files) == 2                  # 20 天前的会议不导出
    acme = (out / [f for f in files if "ACME" in f][0]).read_text(encoding="utf-8")
    assert "- [ ] Send follow-up email" in acme and "- [x] ACME to confirm" in acme and "### Pricing and Rollout Options" in acme
    weekly = (out / [f for f in files if "ACME" not in f][0]).read_text(encoding="utf-8")
    assert "可能还没有生成" in weekly
    details = [c for c in env.calls if c[0].startswith("/v1/conversations/")]
    assert all(c[1]["include"] == ["action_items,outline"] for c in details)    # 不用 include=all

def test_pagination_stops_at_since(env):
    c = O.Client("good", min_interval=0)
    import datetime as dt
    got = [x["id"] for x in c.conversations(dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=7), limit=1)]
    assert got == ["c1", "c2"] and len([x for x in env.calls if x[0] == "/v1/conversations"]) == 3

def test_401(env):
    with pytest.raises(O.OtterError, match="401"): list(O.Client("bad", min_interval=0).conversations())

def test_429_backoff(env):
    waits = []; env.fail_429 = 2
    c = O.Client("good", sleep=waits.append, min_interval=0); c.get("/conversations", limit=5)
    assert waits == [1, 2]

def test_rate_limit_interval(env):
    c = O.Client("good")                                                         # 默认最小间隔 0.11 秒
    for _ in range(5): c.get("/conversations", limit=5)
    gaps = [b - a for a, b in zip(env.times, env.times[1:])]
    assert min(gaps) >= 0.1

def test_null_status_with_data_is_not_pending():
    # 调试时发现：官方示例的 process_status 全是 null，但摘要和待办都在，第一版仍然提示“还没有处理完”
    ex = {"relationships": {"action_items": [{"id": "a", "text": "Follow up with the sales team", "assignee": {"name": "Jane Doe"}, "status": {"completed": True}}],
                            "outline": None}, "id": "conversation_id", "title": "product launch meeting", "url": "https://otter.ai/u/conversation_id",
          "owner": {"name": "Jane Doe"}, "created_at": "2025-07-31T12:00:00Z",
          "process_status": {"abstract_summary": None, "action_item": None, "outline": None}, "abstract_summary": "This meeting discussed the next steps"}
    md = O.minutes_md(ex)
    assert "可能还没有生成：outline" in md and "abstract_summary" not in md.split("可能还没有生成")[1]

第 6 步：运行

$ python demo_local.py          # 本地替身，第一次请求故意返回 429
写入 minutes/20261020-Customer-Discovery-Call-ACME-Corp.md
写入 minutes/20261018-周会-产品规划.md
共 2 场会议（最近 7 天）
$ cat minutes/20261018-周会-产品规划.md
# 周会 / 产品规划

- 时间：2026-10-18 20:02（UTC+8）
- 负责人：Jane Doe
- Otter 链接：https://otter.ai/u/c2

> 注意：以下内容 Otter 可能还没有生成：abstract_summary

## 摘要

（无）

## 大纲

### Introductions and Context
- Participants introduced themselves and clarified roles.

### Pricing and Rollout Options
- The team reviewed per-seat and flat-rate pricing options.

## 待办事项

- [ ] Send follow-up email with pricing summary and implementation overview.（Jane Doe）
- [x] ACME to confirm final user count and preferred contract term.（Alex Taylor）

$ python demo_local.py bad
Otter API 错误： 401 unauthorized：API key 无效，或工作区不是 Enterprise
$ curl -s -w ' %{http_code}' -H 'Authorization: Bearer invalid' https://api.otter.ai/v1/conversations     # 真实接口
{"error": "unauthorized"} 401
$ python -m pytest -q
6 passed

怎么检查结果？

• 两次请求之间至少间隔 0.11 秒（每秒不超过 10 次）；测试会检查请求时间戳。
• 429 时按 1、2、4 秒退避重试。
• 纪要中的时间是 UTC+8；API 返回的是 UTC（以 Z 结尾）。
• 摘要等字段还没生成时，纪要顶部会给出提示。

调试示例

process_status 为 null 时误报“还没生成”。 第一版只要 process_status 中某一项不是 finished 就给出警告。用官方帮助文章中的示例响应测试时，各项状态都是 null，但摘要和待办事项其实已经存在，结果每份纪要都多了一条错误的提示。 修复：只有状态明确是未完成，或者状态为 null 且对应数据确实为空时，才提示。替身中“周会”的摘要为空，所以仍然提示 abstract_summary，这是正确的。

401。 用无效 key 调用真实接口得到 401。官方文章说明 API 只对 Enterprise 开放，所以提示同时写出“key 无效”和“工作区不是 Enterprise”两种可能。

常见问题

模板规则

• API key 只放在环境变量里。
• 遵守限速，处理 429。
• 只请求需要的字段。
• 生成的纪要和原会议一样，只分享给有权限的人。

项目资源

让 API 搬运数据，纪要的内容仍要由参会者确认。