企业微信二次开发：异步导出接口怎么用，通讯录/客户大数据量不超时？

getSyncList 游标增量拉通讯录很方便，但只能拉实时变化的那几条。如果需要一次性导出全部客户（比如做年度盘点、对接第三方数据平台），几百上千个客户能正常返回，上万个客户大概率会超时或丢失。

企微 API 有专门的异步导出模式，就是为大数据量准备的。submit → poll → download 三步，不会超时。

同步 vs 异步，什么时候用哪个

场景

推荐方式

原因

第一步：submit 提交导出请求

异步导出的入口接口按数据类型分几个：

数据类型

submit 路径

resp = requests.post(
f"{BASE}/wx-api/api/contact/submitExportMembers",
headers=HEADERS,
json={
"appid": APPID,
"offset": 0,
"limit": 100000, # 一次最多导 10 万条
"fields": ["basic", "deptIds", "phone", "external"]
}
).json()

if resp["code"] == 0:
export_id = resp["data"]["exportId"]

submit 返回的 exportId 是异步任务的唯一标识，后续轮询和下载都要用它。

第二步：poll 轮询任务状态

异步任务不会立即完成，需要每隔几秒查一次状态：

def poll_export_status(export_id, max_wait=300, interval=5):
"""轮询异步导出任务，最多等 5 分钟"""
start = time.time()

while time.time() – start < max_wait:
resp = requests.post(
f"{BASE}/wx-api/api/contact/getExportStatus",
headers=HEADERS,
json={"appid": APPID, "exportId": export_id}
).json()

if resp["code"] != 0:
raise Exception(f"状态查询失败: {resp}")

data = resp["data"]
status = data["status"] # 0=处理中, 1=已完成, 2=失败

if status == 1:
return data # 包含 downloadUrl / fileId
elif status == 2:
raise Exception(f"导出失败: {data.get('errMsg')}")

time.sleep(interval)

raise TimeoutError(f"导出超时，exportId={export_id}")

轮询间隔建议 5 秒起，任务量大可以调到 10 秒。任务完成时返回的数据里有两种下载方式：

返回字段

下载方式

第三步：download 下载文件

小文件直接 GET downloadUrl 就行。大文件要走 CDN 凭证机制（和发送文件时的凭证是同一套）：

def download_large_file(file_id, file_size):
# 1. 拿 CDN 下载凭证
ticket = requests.post(
f"{BASE}/wx-api/api/cdn/download",
headers=HEADERS,
json={
"appid": APPID,
"id": file_id, # 注意是 id 不是 fileId
"size": file_size, # size 字段
"fileType": 5 # 5=文件
}
).json()

if ticket["code"] != 0:
raise Exception(f"CDN 凭证失败: {ticket}")

cdn_url = ticket["data"]["url"] # 带签名的临时下载 URL

# 2. 直接下载
resp = requests.get(cdn_url, stream=True)
with open("export.csv", "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)

注意 CDN 凭证里的字段名是 id、size、md5，不是 fileId、fileSize、fileMd5。和发送文件时一样，要改名。

第四步：处理导出文件

导出的文件默认是 CSV 格式，包含完整字段：

import csv

def parse_export_csv(filepath):
with open(filepath, encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
# 每条 row 是一个成员/客户
yield {
"uin": int(row["uin"]), # int64 显式转换
"name": row["nickname"],
"phone": row.get("phone", ""),
"dept_ids": json.loads(row.get("deptIds", "[]")),
…
}

CSV 里的数字列要显式转类型（uin、corpId 是 int64，不能留字符串）。

异步导出 vs 游标增量，怎么选

维度

异步导出

游标增量

最佳实践：游标增量做主，异步导出做校验。

# 日常：每 5 分钟游标增量
schedule.every(5).minutes.do(sync_by_cursor)

# 每日凌晨：异步导出全量，和增量数据对比差异
schedule.every().day.at("03:00").do(validate_by_export)

异步导出负责"兜底"，游标增量负责"实时"。两层结合，既不漏数据也不超时。

几个坑

任务不是立即完成的。submit 后不要紧接着调 download，一定会失败。必须轮询 status=1 才往下走。

exportId 有有效期。平台只保留导出文件 24 小时，过了要重新 submit。

limit 有上限。单次 submit 的 limit 最多 10 万条，不是无限大。超过 10 万要分批 submit。

CDN 凭证的字段名。download 接口传 id/size/md5，不是 fileId/fileSize/fileMd5。这个坑之前在发送文件时就踩过一次，下载时一样。

最简骨架

def export_all_members():
export_id = submit_export_members()
status_data = poll_export_status(export_id)

if "downloadUrl" in status_data:
download_file(status_data["downloadUrl"], "members.csv")
else:
download_large_file(status_data["fileId"], status_data["fileSize"])

return parse_export_csv("members.csv")

# 每日凌晨 3 点全量校验
schedule.every().day.at("03:00").do(export_all_members)

异步导出接口的参数说明和 CDN 凭证机制在开发文档的 contact 和 cdn 模块里，控制台开通实例后就能用。