ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

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

影刀RPA新手教程:Notion数据库对接实战——自动写入记录与任务管理 影刀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你不需要打印到文档里看直接在调试变量面板点开看结构就行。Alt4打开没有的话在顶部菜单→视图→调试变量。Chrome浏览器插件是必装的Notion本身是Web应用但不是我们操作的对象。我们是通过API直接跟Notion服务器通信不需要打开Notion网页——影刀的HTTP指令替代了浏览器操作。社区版vs创业版社区版完全能做HTTP请求Notion API对接不依赖高级功能。创业版多了数据库连接和API调度等你需要把数据先存本地数据库再推Notion的时候再升级不迟。元素定位——HTTP请求的参数定位逻辑API对接不需要XPath和CSS选择器但思维是一样的你不再定位网页的DOM元素而是定位JSON里的字段路径。原来用//div[classtitle]找标题现在用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转成字典responserequests.post(url,headersheaders,jsonbody)dataresponse.json()# 查看结构ifdata.get(object)list:forpageindata.get(results,[]):propspage[properties]列表操作影刀ForEach循环遍历data[results]列表每个元素是一个Notion页面对象。字典取值连续get避免KeyErrortitlepage.get(properties,{}).get(Title,{}).get(title,[{}])contenttitle[0].get(text,{}).get(content,)iftitleelse流程控制——Notion操作的标准流程获取Notion API Token - 查询数据库现有任务 - ForEach循环处理新采集的数据 - 判断任务是否已存在 - 存在就更新(PATCH),不存在就创建(POST) - 记录日志。具体代码逻辑# 步骤1: 查询数据库,获取所有已有任务query_urlfhttps://api.notion.com/v1/databases/{DB_ID}/queryresprequests.post(query_url,headersHEADERS,json{page_size:100})existing_tasks{}forpageinresp.json().get(results,[]):task_nameget_title(page)existing_tasks[task_name]page[id]# 步骤2: 遍历新数据fornew_taskincollected_data:task_namenew_task[name]iftask_nameinexisting_tasks:# 更新已有任务update_urlfhttps://api.notion.com/v1/pages/{existing_tasks[task_name]}requests.patch(update_url,headersHEADERS,jsonbuild_update_body(new_task))else:# 创建新任务create_urlhttps://api.notion.com/v1/pagesrequests.post(create_url,headersHEADERS,jsonbuild_create_body(new_task,DB_ID))While循环的应用Notion数据库查询有分页限制每页最多100条用While循环翻页has_moreTruestart_cursorNoneall_pages[]whilehas_more:body{page_size:100}ifstart_cursor:body[start_cursor]start_cursor resprequests.post(query_url,headersHEADERS,jsonbody)dataresp.json()all_pages.extend(data[results])has_moredata.get(has_more,False)start_cursordata.get(next_cursor)Try-Catch在API调用里是基本素养网络超时、Token过期、请求频率限制——都要包在Try里try:resprequests.post(url,headersHEADERS,jsonbody,timeout30)ifresp.status_code429:# 限流time.sleep(3)resprequests.post(url,headersHEADERS,jsonbody,timeout30)exceptrequests.exceptions.Timeout:print(f请求超时,跳过:{url})continueNotion API的速率限制是每秒3次请求。大量操作时加time.sleep(0.5)每个请求之间我踩过被429限流后连续失败100多条记录的坑。网页自动化——虽然不用浏览器,监听依然重要Notion对接不是网页操作,是API操作。但调试时网页自动化技能依然有用——Notion的开发者页面notion.so/my-integrations需要在浏览器中操作创建Integration。网页监听的思路用到JSON调试上不确定Notion返回的数据结构时先把响应结果json.dumps(data, indent2, ensure_asciiFalse)写入文本文件用VS Code打开格式化查看。这和影刀网页监听后写入文本文件→CtrlF查找的思路一模一样。影刀网页监听的典型流程回顾开始监听→触发请求→获取结果→停止监听→写入文件。换成API调试就是发请求→获取响应→打印structure→找目标字段路径。窗口切换的场景可能有人在Notion网页里手动操作同时跑API流程。两者的窗口不冲突——API不依赖网页但如果你需要验证API写入结果在浏览器里刷新Notion页面查看。数据处理——构建Notion属性值的通用函数前面说了Notion属性结构复杂这里给一个通用构建函数defbuild_property(field_type,value):根据字段类型构建Notion属性值iffield_typetitle:return{title:[{text:{content:str(value)},type:text}]}eliffield_typerich_text:return{rich_text:[{text:{content:str(value)},type:text}]}eliffield_typenumber:return{number:float(value)ifvalueelseNone}eliffield_typeselect:return{select:{name:str(value)}}eliffield_typemulti_select:return{multi_select:[{name:v.strip()}forvinvalue.split(,)]}eliffield_typedate:return{date:{start:str(value)}}eliffield_typecheckbox:return{checkbox:bool(value)}eliffield_typeurl:return{url:str(value)}然后批量构建创建请求的bodydefbuild_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:fBearer{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?vyyyxxx部分就是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_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxmsg{msgtype:markdown,markdown:{content:f任务更新通知\n 今日新增任务:{new_count}个\n 已完成:{done_count}个}}requests.post(webhook_url,jsonmsg)邮件通知同上影刀内置邮件指令或者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_TOKENsecret_xxxTASK_DB_IDxxx-yyy-zzzPROJECT_DB_IDaaa-bbb-cccFEISHU_WEBHOOKhttps://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 UnauthorizedToken无效。检查Token是否过期、是否多空格、Integration是否已创建。404 Not Found数据库找不到。检查Database ID是否正确提取URL中的那串字符、数据库是否已共享给Integration。400 Bad Request请求体格式错误。检查properties结构是否符合Notion规范每个字段有没有包type外壳。429 Too Many Requests请求频率超限。加sleep降低并发。Notion限制每秒3次请求。504 Gateway TimeoutNotion服务器超时。重试机制用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 #自动化办公 #任务管理 #新手教程作者林焱
返回列表