用 Fireflies GraphQL 批量导出会议记录与逐字稿

Fireflies 会自动记录和转写会议。想把记录存到自己的笔记库里，可以用它的 GraphQL API 批量导出。 这篇做 ff_export.py：分页列出会议（每页最多 50 条），逐场取逐字稿、概要和待办，生成 Markdown（时间按 UTC+8）；已导出的跳过，重跑不浪费额度。

先看清成品

• 真实接口：无效 key 得到的是 HTTP 500 + auth_failed，脚本识别为认证失败，不当成服务器故障重试。
• 替身上跑通：57 场会议分 2 页列出；中途一次 429 按 Retry-After 等待后继续；导出 5 场。
• 重跑时“新导出 0 场”，只用 2 次请求。pytest 5 项通过。

合规提醒：会议录音和转写涉及所有参会者的个人信息。录音前必须告知并取得同意（不同地区法律要求不同）；只导出你有权访问的会议，导出的文件妥善保管，分享前确认参会者知情。发言时长报告只用于改进会议，不要用来考核或监控个人。本文只使用 Fireflies 官方 GraphQL API（https://api.fireflies.ai/graphql），key 在 Fireflies 的 Integrations → Fireflies API 中获取，放在 FIREFLIES_API_KEY。

运行说明：box 上没有 Fireflies 账号和 key。我用无效 key 真实调用了 GraphQL 接口（2026-10-21 20:35 和 20:53，UTC+8），得到 HTTP 500 和 {"errors":[{"friendly":true,"code":"auth_failed","message":"An error occurred while authenticating your request. Please recheck your API key and try again. …"}]}。其余响应来自我写的本地替身：字段名按官方文档，57 场会议的内容全是编的（每天 3 场，时间都是 08:00）。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

sentences 的 start_time 文档只写了 String 类型、没写单位；脚本按秒处理，用真实账号时先拿一场会议核对。

文件结构

79-fireflies-A-src/
├── ff_export.py
├── fake_fireflies.py
├── demo_local.py
└── test_export.py

第 1 步：主脚本

"""把 Fireflies 的会议记录导出成 Markdown（一场一个文件）：GraphQL 分页列出 → 逐场取全文和摘要。
官方文档：https://docs.fireflies.ai  key：Fireflies 的 Integrations → Fireflies API（或 Settings → Developer settings），放在 FIREFLIES_API_KEY。
只导出你有权访问的会议；分享前确认参会者知情同意。"""
import argparse, json, os, re, sys, time
from datetime import datetime, timedelta, timezone
from pathlib import Path
import requests

URL = os.environ.get("FIREFLIES_URL", "https://api.fireflies.ai/graphql")
TZ = timezone(timedelta(hours=8))
PAGE = 50  # transcripts 的 limit 最大 50

LIST_Q = """query Transcripts($limit: Int, $skip: Int, $fromDate: DateTime, $toDate: DateTime) {
  transcripts(limit: $limit, skip: $skip, fromDate: $fromDate, toDate: $toDate) { id title date duration } }"""
ONE_Q = """query Transcript($transcriptId: String!) {
  transcript(id: $transcriptId) { id title date duration participants transcript_url
    sentences { index speaker_name text start_time end_time }
    summary { overview action_items keywords } } }"""

class FFError(Exception): pass

def gql(key, query, variables, *, session=requests, sleep=None, tries=3):
    sleep = sleep or time.sleep
    for n in range(tries):
        r = session.post(URL, json={"query": query, "variables": variables}, timeout=60,
                         headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "binbho-tutorial/1.0"})
        try: body = r.json()
        except ValueError: body = {}
        errs = body.get("errors") or []
        code = (errs[0].get("code") or errs[0].get("extensions", {}).get("code")) if errs else None
        if code == "auth_failed":  # 注意：真实接口对无效 key 返回 HTTP 500，不能当成服务器故障去重试
            raise FFError("认证失败（auth_failed）：检查 FIREFLIES_API_KEY")
        if r.status_code == 429 or code == "too_many_requests":
            wait = float(r.headers.get("Retry-After") or 60)
            if n < tries - 1:
                print(f"  被限流，{wait:g}s 后重试（限流期间的请求也会计数，不要连续重试）", file=sys.stderr); sleep(wait); continue
            raise FFError("持续被限流：稍后再跑，或减少请求")
        if errs: raise FFError(f"{code}: {errs[0].get('message')}")
        if r.status_code >= 500 and n < tries - 1: sleep(2 ** n); continue
        if r.status_code >= 400: raise FFError(f"HTTP {r.status_code}: {r.text[:200]}")
        return body["data"]
    raise FFError("请求失败")

def list_all(key, since=None, until=None, **kw):
    out, skip = [], 0
    while True:
        v = {"limit": PAGE, "skip": skip}
        if since: v["fromDate"] = since
        if until: v["toDate"] = until
        page = gql(key, LIST_Q, v, **kw)["transcripts"]
        out += page
        if len(page) < PAGE: return out
        skip += PAGE

def local(ms): return datetime.fromtimestamp(ms / 1000, TZ).strftime("%Y-%m-%d %H:%M")

def mmss(t):
    s = int(float(t)); return f"{s // 60:02d}:{s % 60:02d}"  # 文档里 start_time 是 String、未写单位；按秒处理，先用一场真会议核对

def to_md(t):
    lines = [f"# {t['title']}", "", f"- 时间：{local(t['date'])}（UTC+8）", f"- 时长：{t['duration']:.0f} 分钟",
             f"- 参会者：{', '.join(t.get('participants') or [])}", f"- 原始记录：{t.get('transcript_url', '')}", ""]
    s = t.get("summary") or {}
    if s.get("overview"): lines += ["## 概要", "", s["overview"].strip(), ""]
    if s.get("action_items"): lines += ["## 待办", "", s["action_items"].strip(), ""]
    lines += ["## 逐字稿", ""]
    for x in t.get("sentences") or []:
        lines.append(f"**[{mmss(x['start_time'])}] {x['speaker_name']}**：{x['text']}")
    return "\n".join(lines) + "\n"

def safe(name): return re.sub(r'[\\/:*?"<>|\s]+', "_", name).strip("_")[:60]

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("--since", help="ISO 时间，例如 2026-10-01T00:00:00+08:00")
    ap.add_argument("--until"); ap.add_argument("-o", default="meetings"); ap.add_argument("--max", type=int, default=0)
    a = ap.parse_args(argv)
    key = os.environ.get("FIREFLIES_API_KEY") or sys.exit("请先 export FIREFLIES_API_KEY=…")
    out = Path(a.o); out.mkdir(exist_ok=True)
    try:
        items = list_all(key, a.since, a.until)
        print(f"共 {len(items)} 场会议", file=sys.stderr)
        done = 0
        for it in items[: a.max or None]:
            f = out / f"{local(it['date'])[:10]}_{safe(it['title'])}_{it['id']}.md"
            if f.exists(): continue  # 已导出的跳过：重跑时不浪费每日额度
            t = gql(key, ONE_Q, {"transcriptId": it["id"]})["transcript"]
            f.write_text(to_md(t), encoding="utf-8"); done += 1
        print(f"新导出 {done} 场 → {out}/（共调用约 {done + len(items) // PAGE + 1} 次 API）")
    except FFError as e:
        sys.exit(f"错误：{e}")

if __name__ == "__main__":
    main()

第 2 步：本地替身

"""本地替身（我写的）：只认官方文档里出现的字段名（transcripts / transcript / sentences / summary / analytics）。
会议内容全是编的。无效 key 的响应照抄真实接口：HTTP 500 + errors[0].code = auth_failed（2026-10-21 实测）。"""
import json, re, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

def _mk(i, day, title, lines, actions, speakers):
    return {"id": f"T{i:03d}", "title": title, "date": day * 1000, "dateString": "", "duration": 30.0 + i,
            "organizer_email": "lead@example.com", "participants": ["lead@example.com", "dev@example.com"],
            "transcript_url": f"https://app.fireflies.ai/view/T{i:03d}",
            "sentences": [{"index": k, "speaker_name": s, "text": t, "start_time": str(5.0 * k), "end_time": str(5.0 * k + 4.2)} for k, (s, t) in enumerate(lines)],
            "summary": {"overview": f"{title}：示例概要。", "action_items": actions, "keywords": "示例"},
            "analytics": {"speakers": [dict(sp, duration_pct=round(sp["duration"] / ((30.0 + i) * 60) * 100, 1)) for sp in speakers]}}

BASE_DAY = 1792540800  # 2026-10-21 08:00 UTC+8
DATA = []
for i in range(1, 58):  # 57 场会议：超过一页（50 条），用来测试分页
    DATA.append(_mk(i, BASE_DAY - 86400 * (i % 20), f"周会 #{i}",
                    [("王敏", "我们先看上周的待办。"), ("李强", "接口文档我周三前补完。"), ("王敏", "好，那测试环境谁负责？"), ("赵雪", "我来，周五之前。")],
                    "**李强**\n补完接口文档（周三前）\n\n**赵雪**\n搭好测试环境（周五前）\n",
                    [{"speaker_id": 1, "name": "王敏", "duration": 600 + i, "word_count": 900, "longest_monologue": 120, "monologues_count": 8, "filler_words": 12, "questions": 6, "duration_pct": 55.0, "words_per_minute": 150},
                     {"speaker_id": 2, "name": "李强", "duration": 300, "word_count": 420, "longest_monologue": 60, "monologues_count": 5, "filler_words": 3, "questions": 1, "duration_pct": 27.0, "words_per_minute": 140},
                     {"speaker_id": 3, "name": "赵雪", "duration": 200, "word_count": 300, "longest_monologue": 40, "monologues_count": 4, "filler_words": 2, "questions": 0, "duration_pct": 18.0, "words_per_minute": 145}]))
CALLS = {"n": 0, "throttle_after": None}

def pick(obj, sel):
    """极简字段选择：只返回查询里写到的字段名（嵌套对象原样返回）。"""
    return {k: v for k, v in obj.items() if re.search(rf"\b{k}\b", sel)}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, code, obj, headers=()):
        b = json.dumps(obj, ensure_ascii=False).encode(); self.send_response(code); self.send_header("Content-Type", "application/json")
        for k, v in headers: self.send_header(k, v)
        self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def do_POST(self):
        req = json.loads(self.rfile.read(int(self.headers["Content-Length"]))); q, v = req["query"], req.get("variables", {})
        if self.headers.get("Authorization") != "Bearer demo":
            return self._send(500, {"errors": [{"friendly": True, "code": "auth_failed", "message": "An error occurred while authenticating your request. Please recheck your API key and try again. If this persists, please contact Fireflies support.", "extensions": {"code": "auth_failed"}}]})
        CALLS["n"] += 1
        if CALLS["throttle_after"] is not None and CALLS["n"] > CALLS["throttle_after"]:
            CALLS["throttle_after"] = None
            return self._send(429, {"errors": [{"code": "too_many_requests", "message": "Too many requests", "extensions": {"code": "too_many_requests", "status": 429, "metadata": {"retryAfter": 1792586400000}}}]}, [("Retry-After", "2")])
        if "transcripts(" in q:
            lim = v.get("limit", 50)
            if lim > 50: return self._send(400, {"errors": [{"code": "args_required", "message": "limit must be <= 50"}]})
            rows = sorted(DATA, key=lambda d: -d["date"])
            if v.get("fromDate"): rows = [d for d in rows if d["date"] >= _ms(v["fromDate"])]
            if v.get("toDate"): rows = [d for d in rows if d["date"] <= _ms(v["toDate"])]
            sk = v.get("skip", 0)
            return self._send(200, {"data": {"transcripts": [pick(d, q) for d in rows[sk:sk + lim]]}})
        if "transcript(" in q:
            d = next((x for x in DATA if x["id"] == v["transcriptId"]), None)
            if not d: return self._send(200, {"errors": [{"code": "object_not_found", "message": "transcript not found"}], "data": None})
            return self._send(200, {"data": {"transcript": pick(d, q)}})
        self._send(400, {"errors": [{"message": "unknown query"}]})

def _ms(iso):
    from datetime import datetime
    return int(datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() * 1000)

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=s.serve_forever, daemon=True).start(); return s

第 3 步：演示脚本

import os, sys
import fake_fireflies
srv = fake_fireflies.start(); os.environ["FIREFLIES_URL"] = f"http://127.0.0.1:{srv.server_port}/graphql"
if "--throttle" in sys.argv: sys.argv.remove("--throttle"); fake_fireflies.CALLS["throttle_after"] = 3
import ff_export; ff_export.time.sleep = lambda s: print(f"  （演示中跳过等待 {s}s）", file=sys.stderr)
sys.argv = ["ff_export.py"] + sys.argv[1:]; ff_export.main()

第 4 步：测试

import pytest
import fake_fireflies

@pytest.fixture
def ff(monkeypatch):
    srv = fake_fireflies.start(); monkeypatch.setenv("FIREFLIES_URL", f"http://127.0.0.1:{srv.server_port}/graphql")
    import importlib, ff_export; importlib.reload(ff_export); fake_fireflies.CALLS.update(n=0, throttle_after=None)
    yield ff_export; srv.shutdown()

def test_pagination_gets_all(ff):
    items = ff.list_all("demo")
    assert len(items) == 57 and fake_fireflies.CALLS["n"] == 2  # 50 + 7

def test_auth_failed_500_not_retried(ff):
    with pytest.raises(ff.FFError, match="auth_failed"):
        ff.gql("bad", ff.LIST_Q, {"limit": 1}, sleep=lambda s: pytest.fail("不应重试"))

def test_429_uses_retry_after(ff):
    fake_fireflies.CALLS["throttle_after"] = 0; waits = []
    ff.gql("demo", ff.LIST_Q, {"limit": 1}, sleep=waits.append)
    assert waits == [2.0]

def test_date_filter_and_markdown(ff):
    items = ff.list_all("demo", since="2026-10-20T00:00:00+08:00")
    assert items and all(ff.local(i["date"]) >= "2026-10-20" for i in items)
    md = ff.to_md(ff.gql("demo", ff.ONE_Q, {"transcriptId": items[0]["id"]})["transcript"])
    assert "（UTC+8）" in md and "**[00:05] 李强**" in md

def test_skip_existing(ff, tmp_path, monkeypatch, capsys):
    monkeypatch.setenv("FIREFLIES_API_KEY", "demo"); monkeypatch.chdir(tmp_path)
    ff.main(["--max", "3"]); ff.main(["--max", "3"])
    assert "新导出 0 场" in capsys.readouterr().out

第 5 步：运行

$ FIREFLIES_API_KEY=demo python demo_local.py --max 5 --throttle     # 替身在第 4 次请求时返回 429
共 57 场会议
  被限流，2s 后重试（限流期间的请求也会计数，不要连续重试）
  （演示中跳过等待 2.0s）
新导出 5 场 → meetings/（共调用约 7 次 API）
$ FIREFLIES_API_KEY=demo python demo_local.py --max 5                # 再跑一次
共 57 场会议
新导出 0 场 → meetings/（共调用约 2 次 API）
$ head -24 meetings/2026-10-20_周会_#1_T001.md
# 周会 #1

- 时间：2026-10-20 08:00（UTC+8）
- 时长：31 分钟
- 参会者：lead@example.com, dev@example.com
- 原始记录：https://app.fireflies.ai/view/T001

## 概要

周会 #1：示例概要。

## 待办

**李强**
补完接口文档（周三前）

**赵雪**
搭好测试环境（周五前）

## 逐字稿

**[00:00] 王敏**：我们先看上周的待办。
**[00:05] 李强**：接口文档我周三前补完。
**[00:10] 王敏**：好，那测试环境谁负责？
…
$ FIREFLIES_API_KEY=invalid python ff_export.py --max 1      # 真实接口
错误：认证失败（auth_failed）：检查 FIREFLIES_API_KEY
$ python -m pytest -q
5 passed

怎么检查结果？

• 会议总数和 Fireflies 网页上一致。
• 文件名含日期（UTC+8）、标题和 id，同名会议不会互相覆盖。
• 每日额度按 UTC 零点重置：北京时间早上 8 点前后跑，用的是不同日期的额度。

调试示例

无效 key 返回 500。 一般的写法是 r.raise_for_status()，再对 5xx 做退避重试。我先用这种写法试了真实接口：

500 Internal Server Error
HTTPError 500 Server Error: Internal Server Error for url: https://api.fireflies.ai/graphql

看起来是“服务器内部错误”，按惯例会去重试，但其实是 key 错了，重试多少次都一样，还在消耗请求次数。响应体里的 errors[0].code 是 auth_failed。

处理：先解析 GraphQL 的 errors，auth_failed 直接报“检查 FIREFLIES_API_KEY”；too_many_requests 或 HTTP 429 按 Retry-After 等待；其他 5xx 才退避重试。测试 test_auth_failed_500_not_retried 会在发生重试时直接失败。

常见问题

模板规则

• 先看 GraphQL errors，再看 HTTP 状态码。
• auth_failed 不重试。
• 已导出的会议跳过。
• 只导出有权访问的会议，并取得参会者同意。

项目资源

导出之前先把错误分清楚，额度才不会白白浪费。