ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

用Python脚本批量更新多维表格:查询、过滤与安全修改实战

用Python脚本批量更新多维表格:查询、过滤与安全修改实战

上个月,运营同事又抱着电脑来找我,说一张多维表格里三百多条记录需要按条件批量改状态,她在网页端手动点了一个上午,眼睛都快花了。那张表我瞄了一眼:客户名称、负责人、跟进状态、最近跟进时间,需求很明确——把"待跟进"超过三十天的记录统一改成"已流失",再把负责人换成当前轮值的人。手工做至少两个小时,还容易点错行;我花了二十分钟写了一段 Python 脚本,查询、过滤、批量更新一次跑完,还顺手导出了一份改动日志。

这篇文章就是把这件事完整拆开讲:怎么用 Python 查询多维表格里的内容,怎么安全地批量修改,以及我在实际接口对接中总结出来的各种坑。适合刚接触多维表格开放接口、想在网页端工具之外用脚本批量处理数据的同学,也适合已经在上手、但被过滤语法、字段格式、分页这些问题反复折腾的人。下文会用"某平台"通指这类多维表格产品,代码里的域名和 ID 都是占位符,你们替换成实际环境就能用。

1. 多维表格没那么玄:动手前先把数据模型对齐

1.1 记录、字段和 record_id

多维表格这东西,界面上一看就是表格,但底层已经是个轻量数据库。每一行是一条记录,每一列是一个字段,而且每条记录都有一个稳定唯一的标识,接口里通常叫record_id。这个 ID 特别重要,做修改操作的时候,你必须靠它告诉平台"我要改的是哪一行",而不是靠行号。

我见过不少第一次写这类脚本的同学,习惯用行号去定位数据,结果中间插入、删除几条记录之后行号全变了,更新直接错位,数据一对不上就抓瞎。记住一条核心原则:查询靠条件匹配,修改一律按 record_id,永远不要依赖行号。行号只是界面展示层面的东西,接口层不认这个。

1.2 字段类型决定你能写什么值

多维表格和普通 Excel 最大的差异,是每一列都有明确的字段类型。文本、数字、布尔值、日期、单选、多选、人员、附件、关联记录等等。界面上你随便输入没问题,接口层面每个类型对传入的数据格式都有硬性约束,传错格式平台直接报参数错误。

字段类型界面上看到的样子接口里实际要传的值
文本一段文字字符串
数字123.45数值类型,不能用引号包起来
布尔是/否true / false
日期2024-03-01 10:30毫秒级时间戳(数字)
单选一个选项字符串,必须和选项名完全一致
多选多个选项字符串数组
人员某位同事的名字该同事的用户 ID 数组
附件文件名文件凭证 token 数组

最坑的是日期。界面上显示的是 "2024-03-01",看着就是个普通字符串,接口里要的却是毫秒时间戳。我第一次对接时就栽在这里:直接把 "2024-03-01" 传过去,平台返回参数错误,翻文档翻了半天才反应过来要转成时间戳。这张字段类型对照表建议收藏,写代码之前先把要改的字段从上到下过一遍,每个字段属于什么类型、要传什么格式,列清楚再动手。

1.3 响应信封:code 才是真正的裁判

多维表格平台的开放接口,响应格式高度统一:几乎都是code + msg + data这种信封结构。code为 0 表示成功,非 0 就是各种业务错误码。写代码的第一步永远是检查code,不要只检查 HTTP 状态码。HTTP 200 但业务 code 报错的情况我遇到太多次了,只依赖 HTTP 状态码的脚本在线上跑起来等于盲人开车,看起来一切正常,实际数据根本没写进去。

接口路径一般也有一层业务前缀。以我常用的某平台接口为例,查询记录大致是GET /tables/{table_id}/records,批量更新是PUT /tables/{table_id}/records/batch_update。不同平台的域名和路径前缀会有差异,但整体套路完全平移,核心是搞懂"请求什么资源、用什么方法、传什么参数"这三件事。

2. 准备阶段:密钥、环境和第一个打通请求

2.1 申请应用与开通权限

绝大部分多维表格平台都要求你先在管理后台创建一个"应用"或者"机器人",拿到一对凭证:App ID 和 App Secret,然后在后台给这个应用开通对应数据表的读写权限。这一步没什么技术含量,但有一个很多人容易忽略的细节:权限永远是按应用维度开的,不是按人开的。

同一个账号下可能有多个应用,每个应用能访问哪些空间、哪些数据表,都是后台单独配置的。申请权限的时候,我强烈建议只开最少够用的范围,别一上来就把所有权限勾满。这不是说不信任谁,而是权限开得太大,脚本一旦写错、或者密钥不小心泄露,影响面会大得多——本来只是误改一张表的问题,可能变成误改整个空间的问题。

2.2 Python 环境只需要 requests

Python 环境要求不高,3.8 以上就够。第三方库其实只需要一个requests,一条命令装完:

pip install requests

就这一行,别搞复杂。我见过有人一上来就上 pandas,虽然 pandas 处理表格数据确实香,但多维表格的接口返回的是嵌套 JSON,直接用requests解析、过滤、构造更新负载,反而更轻量,出问题也好排查。等数据量到了几万条以上,再引入 pandas 做复杂统计分析也不迟。脚本自动化的第一要务是简单可靠,不是技术栈华丽。

2.3 换取 token 并验证链路连通

鉴权方式各家平台虽然端点和参数名略有差异,但核心套路基本一致:拿着 App ID 和 App Secret 换一个临时访问令牌(token),后续每个请求的请求头里带上Authorization: Bearer <token>。下面我演示一种很常见的格式:

import requests BASE_URL = "https://open-api.example-multitable.com/v1" # 换成你们平台的实际网关地址 APP_ID = "cli_xxxxxxxxxxxxxxxx" APP_SECRET = "xxxxxxxxxxxxxxxxxxxxxxxx" session = requests.Session() def get_access_token(): resp = session.post( f"{BASE_URL}/auth/token", json={"app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"获取token失败: {payload}") session.headers.update({ "Authorization": f"Bearer {payload['data']['access_token']}" }) return payload["data"]["access_token"]

有人会问,为什么不每次请求前都现取一次 token?因为这类接口对 token 获取本身也有限流,频繁调用会被限制,而且 token 通常有两小时左右的有效期。正确做法是启动时取一次,存到变量里复用它,等请求返回 401 的时候再刷新。我用requests.Session()而不是每次requests.get(),也是为了让连接复用,减少 TCP 握手开销,批量拉取数据时性能差距很明显。

第一次打通,先拉一下自己应用下的数据表列表,看看权限和数据路径对不对:

def list_tables(app_id: str): resp = session.get( f"{BASE_URL}/apps/{app_id}/tables", timeout=10, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"获取数据表列表失败: {payload}") return payload["data"]["items"]

这一步能跑通,说明密钥、网络、权限链路都没问题。别急着写业务逻辑,先把这个最小验证跑通,后面所有问题都会好定位得多。凡是密钥配错、域名配错、权限没开这一类基础问题,都会在这一步暴露出来,而不是等到你批量改数据改到一半的时候才爆雷。

3. 查询记录:分页是唯一靠谱的全量获取方式

3.1 过滤条件:字段名、操作符和值

查数据是后续一切修改的前提,也是最容易出幺蛾子的环节。多维表格开放接口的查询参数一般支持过滤、排序、指定字段、分页。以常见的过滤结构为例,通常是"条件组合"的方式,类似下面这样:

filter_ = { "conjunction": "and", "conditions": [ { "field_name": "跟进状态", "operator": "is", "value": ["待跟进"], }, { "field_name": "最近跟进时间", "operator": "lessThan", "value": [ten_days_ago_ms], }, ], }

这里面有几个细节值得说清楚。第一,value在很多平台是数组,哪怕只有一个值也要写成["待跟进"],少一层括号平台就解析不出来。第二,操作符一般有is、isNot、contains、startsWith、in、greaterThan、lessThan这些,具体以文档为准。我个人的建议是:能用is精准匹配就不要用contains模糊匹配,模糊匹配会把不想要的记录也捞进来,修改的时候容易误伤一大片。第三,日期字段做过滤的时候,值同样要转成毫秒时间戳,你想着传"2024-01-01"这种字符串是行不通的。

3.2 分页循环的正确姿势

接口单次返回的记录条数通常有上限,常见的是 100 条或者 500 条。刚上手的人特别容易犯一个错误:只调一次查询接口,把返回的items当成全部数据。多维表格动辄几百上千条记录,一次根本拿不全,后续的批量修改就会静默漏数据——你以为全改完了,实际只改了一半,这种问题不复查很难发现。

分页的标准姿势是根据响应里的has_more和page_token循环拉取:

import json def fetch_all_records(table_id: str, filter_=None, field_names=None): all_records = [] page_token = None while True: params = {"page_size": 100} if page_token: params["page_token"] = page_token if filter_: params["filter"] = json.dumps(filter_, ensure_ascii=False) if field_names: params["field_names"] = json.dumps(field_names, ensure_ascii=False) resp = session.get( f"{BASE_URL}/tables/{table_id}/records", params=params, timeout=15, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"查询失败: {payload}") items = payload["data"]["items"] all_records.extend(items) if not payload["data"].get("has_more", False): break page_token = payload["data"]["page_token"] return all_records

这里解释了为什么filter参数是json.dumps之后传给requests的:以 JSON 字符串的形式放进查询参数,requests会自动帮你做 URL 编码,中文字段名和特殊字符都不用自己操心,直接传给params就行。而page_size我习惯设成 100,不追求一次拉满——单次请求太久容易超时,分批拉反而稳定,网络抖动时重试的代价也小。

分页必须用接口返回的page_token接着拉,不要自己用 offset 计算偏移量。这是我从教训里学到的。因为底层数据可能在你拉取的过程中被别人新增或删除,offset 一旦错位,你拿到的就不是连续一致的快照;page_token 是平台维护的游标,天然不会错位。

3.3 用 field_names 裁剪返回字段

很多平台的查询接口支持field_names参数,用来指定只返回哪些字段。这个参数别忽略。一张表字段可能有三四十列,但你的业务逻辑只需要其中两三列,全量返回不仅浪费带宽,还会让脚本变慢,尤其是在循环里反复拉取的时候。

上面代码里传field_names=["跟进状态", "最近跟进时间", "负责人"],返回的记录里就只剩这三列和record_id。数据量小的时候感觉不出来,到了几千条记录、几十个字段全量拉一遍的规模,响应体动不动好几兆,性能差距非常明显。而且字段少了,打印日志的时候眼睛也不累,排查问题能快不少。

4. 修改记录:新增、更新与反直觉的字段格式

4.1 新增:批量创建先切分

新增记录相对简单,构造一个fields字典,键是字段名,值按字段类型来,然后调用批量创建接口:

def create_records(table_id: str, record_fields_list: list): resp = session.post( f"{BASE_URL}/tables/{table_id}/records/batch_create", json={"records": [{"fields": fields} for fields in record_fields_list]}, timeout=30, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"新增失败: {payload}") return payload["data"]["records"]

批量创建接口通常也有单次条数上限,比如 100 条。所以新增前先按上限切分:

def chunked(seq, size): for i in range(0, len(seq), size): yield seq[i:i + size] for batch in chunked(record_fields_list, 100): create_records(table_id, batch)

这个切分的习惯一定要养成。平台对单个请求的 body 大小和记录条数都有限制,一次传 300 条被拒是常有的事。切分之后,每一批独立成功、独立报错,出问题也容易定位。

4.2 更新:按 record_id 精准修改

更新就不一样了,每条要改的记录都必须带上record_id。批量更新接口的负载大致是:

def batch_update_records(table_id: str, updates: list): resp = session.put( f"{BASE_URL}/tables/{table_id}/records/batch_update", json={"records": updates}, timeout=30, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"更新失败: {payload}") return payload["data"]["records"]

其中updates长这样:

updates = [ { "record_id": "rec_xxxxxx", "fields": { "跟进状态": "已流失", "负责人": ["ou_yyyyyy"], }, }, # ... 更多记录 ]

更新接口的语义通常是"部分更新":你只传要改的字段,没传的字段保持不变。这个设计本身很贴心,但也带来了一个隐患——如果你从查询结果里拿到整行数据,只改了一个字段就把整行全量传回去,有可能把别人并发修改的内容覆盖掉。所以构造更新负载的时候,只放你真正要动的字段,其余一律不碰。

4.3 界面与接口的字段格式差异清单

这里是实战里最容易被卡住的地方,我单独列一下容易出错的字段格式。

第一,日期。接口要毫秒时间戳,转换方式如下:

import datetime def to_ms_timestamp(date_str: str) -> int: dt = datetime.datetime.strptime(date_str, "%Y-%m-%d") return int(dt.timestamp() * 1000)

第二,多选和人员。多选字段传字符串数组,人员字段传用户 ID 数组。你要是从界面上复制一个同事的中文名字直接传进去,大概率报错——接口要的是平台内部的用户 ID,不是显示名。必须先建一张"姓名到用户 ID"的映射表,再构造负载。

第三,单选字段的值必须是选项的精确名称。界面上选项显示是"已完成",你手一抖传成"已完成 "(后面带了个空格),平台会认为这是个不存在的选项,直接报参数错误或者静默失败。这种肉眼几乎看不出来的问题最耗时间。

第四,布尔字段。接口要的是true/false,JSON 序列化后会自动变成小写,但你千万别在 Python 里传"是"、"否"这类字符串。

4.4 先查后改加一个干跑开关

批量修改最稳的模式是"读-改-写"循环:先查询出符合条件的记录,逐条检查字段当前值,决定哪些要改、改成什么,再构造更新负载。这个流程看起来多了一步,但它能让你在真正写数据之前看到这次操作的影响面。

我还习惯在脚本里加一个干跑开关:

DRY_RUN = True # 改成 False 才会真正执行修改

干跑模式下只打印将要做的修改,不实际调用更新接口。跑一次,肉眼扫一遍日志,确认没有误伤任何不该改的记录,再把开关关掉正式执行。这个习惯帮我避免过至少三次大规模误改——比如过滤条件写宽了,把已经完成的记录也捞进来改了状态。干跑日志一打印,这种问题当场就能看出来。

5. 实测半年里反复出现的翻车现场

5.1 401 和 token 过期引发的连环报错

脚本跑了一半突然全是 401,这是最常见的一类翻车。原因基本只有一个:token 过期。token 一般两小时有效期,脚本处理的数据量大、跑得久,启动时取的 token 到后半段就成了废纸。而我一开始只在启动时取了一次,于是每次遇到都要手动重启脚本,非常蠢。

修复也不复杂,给请求加一层"遇到 401 就重新鉴权再试一次"的包装。核心思路是这样:

def request_with_auth(method, url, **kwargs): resp = session.request(method, url, **kwargs) if resp.status_code == 401: get_access_token() resp = session.request(method, url, **kwargs) return resp

注意避免死循环:get_access_token()本身走的是不需要鉴权的接口,不会被这个包装递归调用。加上这一层之后,长时间跑批基本不会再中断在 401 上。

5.2 空结果排查:字段名里藏着隐形空格

有次查"跟进状态等于待跟进",返回结果死活是空的。我反复检查过滤条件,怎么看都没问题,字段名对、操作符对、值也对。最后实在没招了,把字段名逐个字符打印出来才发现了猫腻:字段名是"跟进状态 ",尾巴上带着一个看不见的空格。从界面复制字段名的时候,鼠标多选了一个空格进来。

平台对字段名是完全精确匹配的,多一个空格就找不到这个字段,查询直接给空结果。从此我学乖了:字段名一律从字段列表接口拉取,用代码比对,不靠肉眼从界面复制。很多平台都有获取字段列表的接口,比如GET /tables/{table_id}/fields,返回所有字段的精确名称和类型。写代码前先把这个列表打印出来,既能拿到准确的字段名,也能顺便确认每个字段的类型,一举两得。

5.3 批量上限、请求超时与重试策略

不同平台的批量接口单次条数上限不一样,有的是 100,有的是 500。我遇到过两次把 300 条一锅端传上去、被平台直接拒绝的情况,报错信息也很直接,说批量更新条数超限。解法就是前面写的chunked切分,按平台上限切成多批。

超时则更多出现在查询大表时。我建议所有请求都显式设置timeout,不要用 Python requests 的默认行为。一旦请求超时,异常会直接抛出来,脚本原地崩溃。更稳的做法是给请求加一个重试机制,配合指数退避:

import time def request_with_retry(func, *args, retries=3, **kwargs): for i in range(retries): try: return func(*args, **kwargs) except requests.exceptions.Timeout: time.sleep(min(2 ** i, 8)) raise RuntimeError("重试多次仍然超时")

5.4 并发写导致的更新丢失

还有一次翻车比较隐蔽。小组里两个同事各自跑自己的脚本,对同一张表做更新。A 先查询到了某条记录的旧值,B 后查询到了同一条记录的旧值,然后 A 把"跟进状态"改成"已联系",B 把"负责人"改成另一个人。因为两个脚本都是拿着旧值去写整行,最后写入的那个人把另一个人的字段覆盖回了旧值——两个人改的字段明明不一样,结果有一个人的改动丢了。

这种问题的根治思路是加乐观锁:查询时把记录的last_modified_time或版本号一并取出来,更新前比对当前值是否还是原来那个版本,如果不是,说明期间有人动过,就要跳过或者二次确认。当然,如果整个团队就你一个脚本在跑,这个问题基本不会出现;但接口一旦对外开放、多人同时维护脚本,这个意识一定要有。

6. 一个可以直接改来用的完整脚本骨架

6.1 脚本结构设计

前面的能力拆开讲了,这一节把它们组合成一个能直接落地的脚本。我建议按下面这种结构组织脚本,逻辑清晰,也好维护:

  1. 配置区:网关地址、App ID、App Secret、完整应用 ID、表 ID、干跑开关。
  2. 鉴权模块:获取 token、401 自动重试。
  3. 查询模块:分页拉取、过滤条件、字段裁剪。
  4. 业务逻辑:决定哪些记录需要改、改成什么值。
  5. 执行模块:批量更新、切分批次、记录日志。
  6. 输出模块:把每条记录改前改后的值写到 JSON 日志文件。

6.2 完整代码骨架

import datetime import json import requests # ---------- 配置区 ---------- BASE_URL = "https://open-api.example-multitable.com/v1" APP_ID = "cli_xxxxxxxxxxxxxxxx" APP_SECRET = "xxxxxxxxxxxxxxxxxxxxxxxx" APP_ID_FULL = "app_xxxxxxxxxxxxxxxx" TABLE_ID = "tbl_xxxxxxxxxxxxxxxx" DRY_RUN = True # 改成 False 才会真正写入 # ---------------------------- session = requests.Session() def get_access_token(): resp = session.post( f"{BASE_URL}/auth/token", json={"app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"获取token失败: {payload}") session.headers.update({ "Authorization": f"Bearer {payload['data']['access_token']}" }) def api_get(path: str, params: dict): resp = session.get(f"{BASE_URL}{path}", params=params, timeout=15) if resp.status_code == 401: get_access_token() resp = session.get(f"{BASE_URL}{path}", params=params, timeout=15) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"接口错误 {path}: {payload}") return payload["data"] def fetch_all_records(table_id: str, filter_=None, field_names=None): records = [] page_token = None while True: data = api_get(f"/tables/{table_id}/records", { "page_size": 100, "page_token": page_token, "filter": json.dumps(filter_, ensure_ascii=False) if filter_ else None, "field_names": json.dumps(field_names, ensure_ascii=False) if field_names else None, }) items = data.get("items", []) records.extend(items) if not data.get("has_more", False): break page_token = data["page_token"] return records def batch_update_records(table_id: str, updates: list): resp = session.put( f"{BASE_URL}/tables/{table_id}/records/batch_update", json={"records": updates}, timeout=30, ) payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"更新失败: {payload}") def chunked(seq, size): for i in range(0, len(seq), size): yield seq[i:i + size] def build_filter(threshold_ms: int): return { "conjunction": "and", "conditions": [ {"field_name": "跟进状态", "operator": "is", "value": ["待跟进"]}, {"field_name": "最近跟进时间", "operator": "lessThan", "value": [threshold_ms]}, ], } def main(): get_access_token() threshold = datetime.datetime.now() - datetime.timedelta(days=30) threshold_ms = int(threshold.timestamp() * 1000) records = fetch_all_records( TABLE_ID, filter_=build_filter(threshold_ms), field_names=["跟进状态", "负责人", "最近跟进时间"], ) print(f"命中记录数: {len(records)}") updates = [] for rec in records: updates.append({ "record_id": rec["record_id"], "fields": { "跟进状态": "已流失", "负责人": ["ou_yyyyyy"], }, }) if DRY_RUN: for u in updates[:10]: print(f"[干跑] {u['record_id']} -> {u['fields']}") return for batch in chunked(updates, 100): batch_update_records(TABLE_ID, batch) print(f"已更新一批,{len(batch)} 条") with open("change_log.json", "w", encoding="utf-8") as f: json.dump(updates, f, ensure_ascii=False, indent=2) if __name__ == "__main__": main()

这个骨架里有几个细节值得留意。filter_和field_names序列化时都用了ensure_ascii=False,这样打印出来的日志是中文而不是一串\u转义,排查问题时人眼可读性高很多。查询接口和更新接口统一走api_get里的 401 重试逻辑,避免各处重复写鉴权代码。切分批次后每一批独立打印进度,跑大数据量的时候你能清楚知道卡在哪一批。

6.3 从脚本到团队工具的扩展方向

这个脚本跑通之后,再往后基本就两条路。一条是挂定时任务,比如每天早上跑一遍,自动把超过三十天未跟进的记录状态刷掉,运营同事再也不用手动盯。另一条是包装成简单的命令行工具,接收日期范围和负责人参数,谁要用直接在参数里指定,不用碰代码。这些扩展都不难,核心还是把查询和修改这两步做扎实。

我个人的体会是:先别急着做大而全的管理平台,把当前最痛的那个场景用脚本先固定下来,跑一两个月,确认稳定了,再逐步加参数、加功能。多维表格自动化的价值是滚雪球式的,第一个脚本带来的效率提升,会促使你发现更多值得自动化的场景;等踩过几轮坑、把通用模块沉淀下来,后面的脚本基本就是拼积木。希望上面的代码和踩坑记录能帮你少走一段弯路,把那些本该用脚本解决的重复劳动,真正交给脚本去解决。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进