☰
影刀RPA新手教程:Notion数据库对接实战——自动写入记录与任务管理
2026/10/3 6:54:08 网站建设 项目流程

影刀RPA新手教程:Notion数据库对接实战——自动写入记录与任务管理

前段时间团队用Notion做任务看板,每天要把采集到的项目进度手动更新到Notion里。20个任务还好,后来涨到200多个,手动更新直接成了噩梦。研究了Notion的API后发现,这玩意儿其实就是一套RESTful接口,影刀发HTTP请求就能操作Notion数据库。这篇文章从零开始带你打通影刀和Notion的数据通道。

认识影刀——准备工作先做好

影刀社区版支持Python指令,这一点很关键——因为Notion API的请求头构建和JSON处理用Python比影刀内置指令灵活得多。安装后确认Python环境正常:import requests; print(requests.__version__),如果报错ModuleNotFoundError说明requests库没装。去影刀设置的Python环境那里手动pip install。

影刀界面里最常驻的区域是"调试变量"面板——运行流程时在这里可以实时看到每个变量的值和类型。An HTTP请求返回的JSON你不需要打印到文档里看,直接在调试变量面板点开看结构就行。Alt+4打开,没有的话在顶部菜单→视图→调试变量。

Chrome浏览器插件是必装的,Notion本身是Web应用,但不是我们操作的对象。我们是通过API直接跟Notion服务器通信,不需要打开Notion网页——影刀的HTTP指令替代了浏览器操作。

社区版vs创业版:社区版完全能做HTTP请求,Notion API对接不依赖高级功能。创业版多了数据库连接和API调度,等你需要把数据先存本地数据库再推Notion的时候再升级不迟。

元素定位——HTTP请求的参数定位逻辑

API对接不需要XPath和CSS选择器,但思维是一样的:你不再定位网页的DOM元素,而是定位JSON里的字段路径。原来用//div[@class='title']找标题,现在用data['properties']['Title']['title'][0]['text']['content']找标题字段。本质都是定位,只是目标从HTML变成了JSON。

CSS选择器的思路在这里依然有用:data.results[0].properties.Status.select.name这种链式取值,和CSS里.results .properties .status是一样的逻辑——逐级往下。

XPath的contains语法对应到JSON里就是字典的get和in判断:if 'Status' in props and 'select' in props['Status'],等价于XPath的contains(@class,'status')。

正则的最常用场景:Notion页面标题里提取编号——“任务_20240630_采集百度地图数据”,用re.search(r'任务_(\d+)', title)提取日期。或者检查数据库名字是否符合规范:re.match(r'^Project_.*_DB$', db_name)。

全局变量的场景:Notion API Token用全局变量NOTION_API_TOKEN存储,不要写死在流程里。数据库ID也是全局变量TASK_DATABASE_ID,切换工作区只需要改变量。

变量与数据类型——Notion数据格式的核心坑

先说结论:Notion的属性类型有十几种,每种的结构都不一样。理解这个结构是写查询和创建请求的基础。

字符串类型的字段(title/rich_text)的结构:

"Title":{"title":[{"text":{"content":"任务名称"},"type":"text"}]}

数字类型(number)的结构:

"Score":{"number":85}

选择类型(select)的结构:

"Status":{"select":{"name":"进行中"}}

日期类型(date)的结构:

"Deadline":{"date":{"start":"2024-06-30","end":None}}

我第一次写创建记录的代码时,直接把字符串当值传入,接口返回400 Bad Request。排查了一下午才发现是properties结构不对——Notion要求每个字段都包一层类型名称的外壳。后面我会给一个通用的构建函数,套用就行。

JSON在Notion API中贯穿全程:请求体是JSON,响应也是JSON。影刀HTTP指令发请求后,返回的response.text转成字典:

response=requests.post(url,headers=headers,json=body)data=response.json()# 查看结构ifdata.get('object')=='list':forpageindata.get('results',[]):props=page['properties']

列表操作:影刀ForEach循环遍历data['results']列表,每个元素是一个Notion页面对象。字典取值:连续get避免KeyError:

title=page.get('properties',{}).get('Title',{}).get('title',[{}])content=title[0].get('text',{}).get('content','')iftitleelse''

流程控制——Notion操作的标准流程

获取Notion API Token -> 查询数据库现有任务 -> ForEach循环处理新采集的数据 -> 判断任务是否已存在 -> 存在就更新(PATCH),不存在就创建(POST) -> 记录日志。

具体代码逻辑:

# 步骤1: 查询数据库,获取所有已有任务query_url=f"https://api.notion.com/v1/databases/{DB_ID}/query"resp=requests.post(query_url,headers=HEADERS,json={"page_size":100})existing_tasks={}forpageinresp.json().get('results',[]):task_name=get_title(page)existing_tasks[task_name]=page['id']# 步骤2: 遍历新数据fornew_taskincollected_data:task_name=new_task['name']iftask_nameinexisting_tasks:# 更新已有任务update_url=f"https://api.notion.com/v1/pages/{existing_tasks[task_name]}"requests.patch(update_url,headers=HEADERS,json=build_update_body(new_task))else:# 创建新任务create_url="https://api.notion.com/v1/pages"requests.post(create_url,headers=HEADERS,json=build_create_body(new_task,DB_ID))

While循环的应用:Notion数据库查询有分页限制(每页最多100条),用While循环翻页:

has_more=Truestart_cursor=Noneall_pages=[]whilehas_more:body={"page_size":100}ifstart_cursor:body["start_cursor"]=start_cursor resp=requests.post(query_url,headers=HEADERS,json=body)data=resp.json()all_pages.extend(data['results'])has_more=data.get('has_more',False)start_cursor=data.get('next_cursor')

Try-Catch在API调用里是基本素养:网络超时、Token过期、请求频率限制——都要包在Try里:

try:resp=requests.post(url,headers=HEADERS,json=body,timeout=30)ifresp.status_code==429:# 限流time.sleep(3)resp=requests.post(url,headers=HEADERS,json=body,timeout=30)exceptrequests.exceptions.Timeout:print(f"请求超时,跳过:{url}")continue

Notion API的速率限制是每秒3次请求。大量操作时加time.sleep(0.5)每个请求之间,我踩过被429限流后连续失败100多条记录的坑。

网页自动化——虽然不用浏览器,监听依然重要

Notion对接不是网页操作,是API操作。但调试时网页自动化技能依然有用——Notion的开发者页面(notion.so/my-integrations)需要在浏览器中操作创建Integration。

网页监听的思路用到JSON调试上:不确定Notion返回的数据结构时,先把响应结果json.dumps(data, indent=2, ensure_ascii=False)写入文本文件,用VS Code打开格式化查看。这和影刀网页监听后"写入文本文件→Ctrl+F查找"的思路一模一样。

影刀网页监听的典型流程回顾:开始监听→触发请求→获取结果→停止监听→写入文件。换成API调试就是:发请求→获取响应→打印structure→找目标字段路径。

窗口切换的场景:可能有人在Notion网页里手动操作同时跑API流程。两者的窗口不冲突——API不依赖网页,但如果你需要验证API写入结果,在浏览器里刷新Notion页面查看。

数据处理——构建Notion属性值的通用函数

前面说了Notion属性结构复杂,这里给一个通用构建函数:

defbuild_property(field_type,value):"""根据字段类型构建Notion属性值"""iffield_type=="title":return{"title":[{"text":{"content":str(value)},"type":"text"}]}eliffield_type=="rich_text":return{"rich_text":[{"text":{"content":str(value)},"type":"text"}]}eliffield_type=="number":return{"number":float(value)ifvalueelseNone}eliffield_type=="select":return{"select":{"name":str(value)}}eliffield_type=="multi_select":return{"multi_select":[{"name":v.strip()}forvinvalue.split(',')]}eliffield_type=="date":return{"date":{"start":str(value)}}eliffield_type=="checkbox":return{"checkbox":bool(value)}eliffield_type=="url":return{"url":str(value)}

然后批量构建创建请求的body:

defbuild_create_body(data,db_id):properties={}field_mapping={'任务名称':('title','name'),'状态':('select','status'),'优先级':('select','priority'),'截止日期':('date','deadline'),'完成度':('number','progress'),'描述':('rich_text','description'),}fornotion_field,(field_type,data_key)infield_mapping.items():ifdata_keyindataanddata[data_key]isnotNone:properties[notion_field]=build_property(field_type,data[data_key])return{"parent":{"database_id":db_id},"properties":properties}

Excel数据导入Notion:影刀读取Excel→每行转字典→调用build_create_body→POST到Notion。批量导入时注意:每500条报告一次进度;每请求间sleep 0.5秒防限流;出错行记录到错误日志方便后续补录。

数据库连接进行批量插入的经验:如果你有MySQL存储原始数据,可以先把数据写入MySQL→再从MySQL读取→批量同步到Notion。这样做的好处是原始数据有备份,Notion只是一个展示层。

鼠标键盘与图像自动化——Notion对接不需要,但有个特殊场景

大部分情况Notion API用不到图像操作。但有一个例外:Notion API不能直接上传文件附件,你需要先在Notion页面手动拖入模板文件,然后让影刀通过图像识别点击页面上的按钮。

另一个场景:当API Token绑定的Integration权限不够时(比如某些Notion工作区设置不允许API操作),退而求其次用网页自动化——打开Notion网页→元素捕获定位表格行→逐行填写数据→点击"添加"按钮。效率低但能跑。这种场景下元素定位(找表格行)、键盘输入(填文字)、图像识别(找按钮)全部用上。

快捷键操作:当页面元素被遮挡无法点击时,Tab键切换焦点→Enter确认→空格勾选复选框。影刀的虚拟键盘驱动支持这种操作——比模拟模式更快,但注意先切换到正确的窗口。

进阶技能——HTTP请求 + Python协同

这是本篇文章的核心。Notion API全部走HTTP,请求头固定格式:

HEADERS={"Authorization":f"Bearer{NOTION_API_TOKEN}","Content-Type":"application/json","Notion-Version":"2022-06-28"}

三个关键点:①Token创建在notion.so/my-integrations→新建Integration→获取Internal Integration Token。②数据库ID从Notion页面URL中提取——URL是notion.so/workspace/xxx?v=yyy,xxx部分就是Database ID。③数据库要共享给Integration——在Notion数据库右上角→连接→添加你创建的Integration,不然API访问会返回404。

查询数据库加筛选条件:

body={"filter":{"and":[{"property":"Status","select":{"equals":"进行中"}},{"property":"Priority","select":{"equals":"高"}}]},"sorts":[{"property":"Deadline","direction":"ascending"}]}

更新记录的PATCH操作——只用传要修改的字段,不改的字段不用传。这个和数据库的UPDATE有本质区别:requests.patch(url, json={"properties": {"Status": build_property('select','已完成')}})只更新Status,其他不变。

Python协同:把Notion操作封装成函数模块——查询函数、创建函数、更新函数、删除函数各一个py文件。主流程调用这些函数。这样做的好处是API地址变了只改一个地方,主流程完全不用动。

OCR在Notion场景的应用:如果你需要采集纸质任务单上的手写内容,先拍照→影刀OCR识别文字→结构化后写入Notion。影刀的文字识别提供了通用标准版和高精度版,一般使用标准版就够了,高精度版用于手写和模糊文字。

ADB手机自动化:如果在手机上用Notion App操作不方便自动化,可以考虑用ADB连接手机→影刀控制手机屏幕→模拟点击操作Notion。这个方案太折腾不推荐,还是老老实实用API吧。

平台实战——Notion + 飞书 + 微信通知三联动

Notion只是任务管理系统的一环,真正的生产力是多个平台联动。最常见的组合:爬虫采集任务数据→存入Notion→飞书多维表格同步展示→微信/钉钉消息通知。

飞书多维表格同步:影刀飞书指令→读取Notion最新数据→写入飞书多维表格。两边的字段映射要对齐——Notion的Select->name对应飞书多维表格的fld_select字段。

微信通知:通过企业微信机器人的Webhook发消息:

webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"msg={"msgtype":"markdown","markdown":{"content":f"任务更新通知\n> 今日新增任务:{new_count}个\n> 已完成:{done_count}个"}}requests.post(webhook_url,json=msg)

邮件通知同上,影刀内置邮件指令或者Python的smtplib都行。

定时任务:影刀计划任务→每天早8点跑一次→采集任务进度→更新Notion→推送通知。如果某个环节失败,记录到日志文件→第二天人工检查。

系统联动——Notion作为数据中心

Notion在这里的角色是"中心数据库"——所有自动化流程的输出最终都汇集到Notion。影刀采集的数据、Excel报表、飞书沟通记录……全部通过API写入Notion,人在Notion上查看、编辑、管理。

飞书消息通知的Webhook配置:飞书群→群设置→群机器人→添加机器人→复制Webhook地址。影刀发送消息时只需要HTTP POST到这个地址,消息格式支持纯文本和富文本卡片。

定时任务配置进阶:多个定时任务注意错开时间,比如8:00跑数据采集→8:15跑Notion同步→8:30跑报告生成。它们之间有依赖关系,用计划任务的"前一任务成功后执行"选项串联。

企业版和创业版支持API调度——外部系统可以通过影刀的调度API触发流程执行。比如有人改了Notion里的某个字段→Webhook触发影刀流程→影刀根据改动做后续处理。这个属于进阶操作了,知道有这个能力就行。

工程化与规范——Notion对接的代码组织

Token和ID集中管理:所有API Token、Database ID、Webhook地址定义在一个全局配置文件里:

# config.pyNOTION_TOKEN="secret_xxx"TASK_DB_ID="xxx-yyy-zzz"PROJECT_DB_ID="aaa-bbb-ccc"FEISHU_WEBHOOK="https://open.feishu.cn/xxx"

流程中只import config,不暴露敏感信息。

子流程封装:notion_query.py只负责查询,notion_create.py只负责创建,notion_update.py只负责更新。每个子流程输入参数是数据库ID和数据,返回操作结果。

调试技巧:影刀打断点→在HTTP请求前打断点→查看请求的url、headers、body是否正确。常见错误:Token少个字符、Headers拼错(Notion-Version写成NotionVersion差了横线)。有一回我的Token末尾多了个空格,API返回401 Unauthorized,查了半个多小时才发现。

命名规范:流程任务同步_Notion_v1.2,变量task_name(蛇形),函数build_create_body()(动词开头),常量NOTION_API_TOKEN(全大写)。

速查表/常见报错

401 Unauthorized:Token无效。检查Token是否过期、是否多空格、Integration是否已创建。

404 Not Found:数据库找不到。检查Database ID是否正确提取(URL中的那串字符)、数据库是否已共享给Integration。

400 Bad Request:请求体格式错误。检查properties结构是否符合Notion规范,每个字段有没有包type外壳。

429 Too Many Requests:请求频率超限。加sleep,降低并发。Notion限制每秒3次请求。

504 Gateway Timeout:Notion服务器超时。重试机制,用Try-Catch包起来自动retry 3次。

字段更新后没变化:PATCH请求只传了要改的属性,但用错了PUT(应该用PATCH)。PATCH是部分更新,PUT是全量替换。

created_time/last_edited_time显示不对:这两个字段是Notion自动生成的,无法通过API修改,创建记录时会自动填充当前时间。

Notion API的学习成本主要在理解它的数据结构。你只要把这个结构搞懂一遍,后面所有的Notion操作都是套路。更多Notion API的实战案例在home.linyan.cloud的专栏里有整理。

#影刀RPA #NotionAPI #自动化办公 #任务管理 #新手教程
作者:林焱

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询