WorkBuddy使用飞书的这个坑,太可怕了
一、拾枝杂谈
WorkBuddy 最近在网上被讨论地沸沸扬扬,有关它的评测如雨后春笋般涌现出来。
于是乎,我也摩拳擦掌,想利用 WorkBuddy 做点东西出来,一是想试试 WorkBuddy 的实际工作能力怎么样,二是做出来东西还能发文章吹一波水,可谓是一举两得。
正好我最近在做 GEO 相关的内容,GEO 领域中一个很重要的事情就是流量统计,我在7个平台发表自己的原创文章,所以我想着能不能用 WorkBuddy 帮我去“统计”这七个平台的文章数据,然后再统一写入到飞书表格中,方便我统一查看和管理。
就像下面这样:

没想到 WorkBuddy 还真给做出来了。七个平台的数据流收集全部打通,而且我还实现了“每个平台统计各自的数据字段”。
我在 https://blog.csdn.net/fastopAI/article/details/163295016 这篇博文中,总览性地介绍了我的工作,包括每个平台的数据分析方法,以及每个平台各自遇到的个性化的问题。
随后,我就聚焦于每个平台遇到的个性化问题以及对应的解决方案,分别写文章去详细说明,算是一种形式的分享和记录。目前我已经针对这个系列写了5篇文章,对应着5个平台:
一是CSDN,https://blog.csdn.net/fastopAI/article/details/163334543这篇博文记录了我是如何以 CSDN 作为突破口,确定了总的数据爬取方式,包括在此过程中踩过的“坑”,以及CSDN 特有的问题,例如“展现量”的获取方式。
二是微信公众号,https://blog.csdn.net/fastopAI/article/details/163358386,在这篇文章中,我成功解决了“多平台统计数据字段不一致” 和 “新旧数据字段不一致” 这两个主要问题,并且让 WorkBuddy 帮我沉淀成 Skill,后面每个平台的数据爬取,以及自动化任务的设定,都可以复用 Skill。
三是知乎:https://blog.csdn.net/fastopAI/article/details/163373246,有了前面两个平台的经验,WorkBuddy 写得脚本一次运行就成功收集了知乎的文章数据,并写入到了飞书表格,但是一句“MEMORY.md” 超限并截断引起了我的警觉,于是我在这篇文章中重点介绍了 WorkBuddy 的三层 Memory 机制。
四是今日头条,我在https://blog.csdn.net/fastopAI/article/details/163448249 这篇博文中,系统分析了实现这个需求的三大难点,还分享了一个很有意思的 Bug——WorkBuddy最初写的脚本有“多动症”。
这篇文章是第五篇文章,对应的平台是小红书。
二、写入飞书时的"破坏性清空"
首先解决一个历史遗留问题,这个问题最初在进行微信公众号这个平台的数据爬取的时候就出现了,就是 WorkBuddy 在写入飞书时,最初采用的策略是“破坏性清空”。
这是因为 WorkBuddy 理解错了我的需求,我之前自己手动统计了两篇文章,而这两篇文章的统计字段并不是按照不同平台个性化统计的,而我又希望能够保留这些旧数据,并在保留旧数据的基础上统计新数据,所以当我通过 WorkBuddy 实现自动化爬取的时候,就出现了“新旧数据不一致”的问题,我当时的需求是“把不匹配的字段删了”,但是它给我理解成了“把整个表清空重写”。
所以它最初的策略是这样的:
--rewrite # 或者 cells-clear --scope all
这就是破坏性清空,前两篇文章对应的子表全被删了,后来还是我通过飞鼠版本历史回溯,重新做安全迁移才解决,我还做了一个 Skill,专门解决多平台字段不匹配的问题。

但是在 知乎 和 今日头条 这两个平台的数据爬取过程中,这个问题又出现了。但这次不是因为 WorkBuddy 误解了我的意思,而是 WorkBuddy 偷懒了。
它在写入飞书时,没有走之前 CSDN 和 微信公众号建立好的标准路径,而是在各自的 zhihu.py 和 toutiao.py 里面写了自定义的 write_to_feishu 逻辑,里面直接调用了 _clear_with_retry(obj_token)。
_clear_with_retry 不管表里有什么,先把表清空,再写新数据。所以每次运行都会把你之前累积的历史行全部删掉。它的代码如下:
def _clear_with_retry(self, obj_token: str, max_retries: int = 3) -> bool:
for attempt in range(max_retries):
self.clear_sheet(obj_token) # cells-clear --scope all
time.sleep(5)
# 验证是否清空
...
return False
这个代码逻辑做了三件事:1)调用飞书 cells-clear --scope all,清空 A1:Z200 的所有内容和格式;2)等待 5 秒,再读一遍确认真的空了;3)写入新的表头 + 当天数据。
结果就是表里只剩今天这一行,以前所有的历史数据全部消失,所以这种策略只有在一种情况下合理,就是我明确说了我不要历史数据了,但显然不是。
三、安全迁移

解决这个问题就是走“安全迁移”,对应migrate_schema_preserve_data(obj_token, new_headers, new_data_rows)方法。
这是一个公用脚本 feishu.writer.py 里面定义的一个方法,专门用于实现安全迁移逻辑。
它的相关代码如下:
def migrate_schema_preserve_data(self, obj_token, new_headers, new_data_rows):
old_header, old_rows = self._read_sheet(obj_token)
remapped = []
for old_row in old_rows:
old_dict = dict(zip(old_header, old_row))
new_row = []
for h in new_headers:
val = old_dict.get(h, "") # 匹配到的字段保留
if not val and h in rename_map:
val = old_dict.get(rename_map[h], "") # 重命名字段兼容
new_row.append(val)
remapped.append(new_row)
all_rows = remapped + new_data_rows # 旧行 + 新行
# 用 csv-put 整体写回
...
这个代码逻辑做了五件事:1)用 csv-get 把旧表的表头和所有数据行读出来;2)把旧行按照新的字段表头重新映射;3)匹配到的字段保留原值;新表有但旧表没有的字段留空;旧表有但新表没有的字段丢弃;4)去掉今天已经存在的旧行(避免和当天新数据重复);5)把迁移后的旧行 + 当天新行一起写回表格。
结果就是历史记录全部保留,只是列结构按新平台字段调整。
而在各个平台的爬虫脚本中,是这么调用公用的飞书写入脚本的:
from feishu_writer import FeishuSheetWriter
writer = FeishuSheetWriter(
space_id=FEISHU_SPACE_ID,
parent_node_token=FEISHU_WIKI_NODE_TOKEN,
platform_name="CSDN"
)
writer.write_scraped_data(scraped_articles)
注意看 write_scraped_data 这个方法!它的决策逻辑是这样的:
for each article:
1. 找有没有匹配的子表(按标题匹配)
if 没匹配到:
→ 新建子表
if 匹配到:
if rewrite=True:
→ 破坏性清空重写(必须用户明确指定)
elif 现有表头 != 新平台字段:
→ 安全迁移 migrate_schema_preserve_data()
elif 今天日期已经存在:
→ 跳过,不重复写入
else:
→ 正常追加一行
也就是说,各个平台对应的爬虫脚本先是引入了 feishu_writer.py 这个公用的飞书写入脚本,然后通过它又去调用了 write_scraped_data 这个入口方法,这个入口方法里面会先进行判断,如果发现了“多平台字段不一致”的问题,就会转去调用 migrate_schema_preserve_data 这个方法,这个方法内部实现了“安全迁移”的逻辑。
用大白话讲就是,write_scraped_data() 拿到数据后会自己判断,如果统计的文章对应的子表已经存在了,但是又和当前平台的统计字段不匹配,这时候就不能直接覆盖,不然会丢失历史数据,所以要调用 migrate_schema_preserve_data 这个方法来安全迁移一下。
但是对于不同平台的爬虫脚本来说,它们不需要知道 migrate_schema_preserve_data 这个内部方法的存在,这个内部方法只是一个干活儿的,它们真正需要知道其实是 write_scraped_data() 这个司令官,这个司令官才是真正决定谁去干活儿的人。
所以说各个平台的爬虫脚本中,直接调用的是 write_scraped_data() 方法,而不是自己去实现代码逻辑。
四、小红书特有的问题
第一个问题是,我之前统计的旧数据里面,用的是“转发”字段,而小红书后台统计的是“分享”。
这时候 WorkBuddy 用了一个 Python 字典 rename_map 作为字段重命名映射表,顾名思义,它的作用就是:当飞书表格的旧列名和新列名不一致但实际是同一个意思时,把旧列的数据自动搬运到新列,避免数据丢失。
例如:
rename_map = {"旧字段名": "新字段名"}
rename_map = {"转发": "分享"}
第二个问题是“空格误判”,就是我在小红书发文时,标题里面不小心漏了一个空格,但是原先的模糊匹配只处理了中文和中文之间的空格,没有处理ASCII字母和中文之间的空格。
新逻辑改成了“在比较标题之前,先把所有无意义空格统计删掉”,包括:1)中文之间的空格、中文和英文/数字之间的空格、多个连续空格合并为1一个。这样“AI 一直在”和“AI一直在”,在经过归一化后都能化成同一个字符串。
五、多次爬取产生了重复的日期记录
同一天连续跑两次某个平台的数据爬取,飞书表格里会出现两行相同的日期。这个是 feishu_writer.py 追加路径的通用Bug,所有平台都会受到影响。
修复前的逻辑大概长下面这样:
# 1. 找到表格最后一行
last_row = self.find_last_data_row(obj_token)
# 2. 从下一行开始,直接追加新数据
start_row = last_row + 1
self.append_data_rows(obj_token, data_rows, start_row=start_row)
大白话解释就是:不管今天有没有爬过,找到最后一行,在它下面再加一行。
修复Bug后,在 find_last_data_row() 之前加了一步去重检测,如下所示:
# Append mode: headers already match — add below existing data
# First, check if today's date row already exists (dedup)
_, existing_rows = self._read_sheet(obj_token)
today = datetime.now().strftime("%Y-%m-%d")
already_has_today = any(
row and row[0] == today for row in existing_rows
if row and len(row) > 0
)
if already_has_today:
print(f"[{self.platform_name}] Today's data already exists "
f"for '{matched_node['title']}' — skipping")
results.append({
"article": article_title,
"child_title": matched_node["title"],
"obj_token": obj_token,
"action": "skip-duplicate",
"ok": True,
})
continue
修复后的逻辑变成了:在追加之前,先把表格里现有的数据读出来。检查第一列"日期"里,是否已经存在今天的日期。如果存在,就跳过,不写入;如果不存在,再往下追加。
不过有意思的是,上文提到的 “安全迁移” 代码中本身已经做了日期去重。也就是说,当字段变化触发迁移时,旧数据里如果已经有今天的日期,会被新数据覆盖,不会重复。但是常规的“追加路径”之前没有这条防线,所以才出现了这个 Bug。
Δ总结
- OK,以上就是本篇文章的全部内容了,感谢阅读!。
- 这篇文章虽然目的是记录 小红书 平台数据爬取过程中出现的问题,但是主要讲解了 WorkBuddy 操作 飞书 时的“全表清空”的一个大坑,以及追加逻辑无去重的一个通用 Bug。下一篇文章会和大家分享一下 Agent 的“幻觉”。
- 关注迅高AI实验室,学习AI不迷路!
更多推荐
所有评论(0)