CogVideoX-2b应用拓展:对接Notion API实现笔记→短视频自动同步

CogVideoX-2b应用拓展:对接Notion API实现笔记→短视频自动同步 CogVideoX-2b应用拓展对接Notion API实现笔记→短视频自动同步你是不是也遇到过这种情况在Notion里精心整理了一份产品介绍、一个项目复盘或者一个知识点的学习笔记想把它做成短视频分享出去却卡在了“从文字到视频”这一步。手动剪辑费时费力找外包又成本高昂。今天我们就来解决这个痛点。我将手把手带你把智谱AI开源的CogVideoX-2b视频生成工具和你的Notion笔记库打通搭建一个“笔记自动变视频”的流水线。你只需要在Notion里写好内容剩下的交给代码。1. 这个方案能帮你做什么简单来说它能实现一个自动化流程你在Notion的指定数据库里新建或更新一条笔记比如标题是“本周产品亮点”内容是一些描述性文字。系统自动监测到这条新笔记提取出标题和关键内容。系统调用CogVideoX-2b模型将这些文字转换成一段短视频。系统将生成的视频文件链接自动添加回那条Notion笔记中。整个过程无需你手动干预。最终你的Notion笔记不仅保留了文字还附带了一个自动生成的、可视化的视频摘要。这对于内容创作、知识沉淀、产品演示等场景效率提升是巨大的。核心价值效率倍增省去从文稿到视频的繁琐制作过程。内容复用将已有的文字资产笔记、文档轻松转化为视频资产。自动化一次搭建持续运行解放双手。隐私安全基于AutoDL的CogVideoX-2b镜像所有处理在本地GPU完成你的笔记内容和生成的视频都不会上传到第三方公有服务。2. 准备工作搞定Notion API和CogVideoX环境在开始写代码之前我们需要拿到两把“钥匙”Notion的访问权限和CogVideoX的服务地址。2.1 获取Notion API集成令牌访问 Notion开发者页面登录你的账号。点击“ New integration”创建一个新的集成。给你的集成起个名字例如“Auto Video Generator”。选择这个集成要关联的工作区Workspace。点击“Submit”创建。创建成功后你会看到“Internal Integration Token”字段下的密钥。复制并保存好这个令牌它看起来像secret_abc123...。这个令牌就像密码不要泄露。接下来你需要邀请这个集成到你的特定Notion页面或数据库。打开你想监控的那个Notion数据库页面点击右上角的“...”菜单选择“Add connections”然后找到并连接你刚刚创建的“Auto Video Generator”集成。2.2 获取你的Notion数据库ID打开你想要同步的Notion数据库页面。浏览器地址栏的URL通常格式是https://www.notion.so/yourworkspace/a8aec43384f447ed84390e8e42c2a9a7?v...其中a8aec43384f447ed84390e8e42c2a9a7这部分就是该数据库的ID。复制并保存这个ID。2.3 部署并启动CogVideoX-2b服务我们使用在AutoDL上优化过的CogVideoX-2b镜像省去了环境配置的麻烦。在AutoDL平台搜索并租用带有合适GPU如RTX 3090/4090或同等算力的实例。在社区镜像中搜索“CogVideoX-2b”并选择CSDN专用版进行部署。实例启动后在控制台点击“自定义服务”或“HTTP链接”打开WebUI界面。你会看到一个简单的输入框和生成按钮。这证明服务已经正常运行。记下这个WebUI的内网地址通常格式如http://127.0.0.1:7860或类似。我们的代码将通过这个地址与视频生成服务通信。3. 核心实现搭建自动化同步脚本现在我们开始编写Python脚本作为整个流程的“大脑”。这个脚本主要做三件事查笔记、生视频、更新笔记。首先安装必要的Python库pip install notion-client requests3.1 第一步连接Notion并查询新笔记我们创建一个notion_video_sync.py文件。import os import time import requests from notion_client import Client from datetime import datetime, timedelta # 配置你的密钥和信息 NOTION_TOKEN 你的Notion集成令牌_secret_abc123 DATABASE_ID 你的Notion数据库ID_a8aec43384f COGVIDEOX_URL http://127.0.0.1:7860 # 你的CogVideoX WebUI内网地址 # 初始化Notion客户端 notion Client(authNOTION_TOKEN) def get_recent_notes(minutes5): 获取最近N分钟内创建或更新的、且没有视频附件的笔记。 # 计算时间点 since_time (datetime.utcnow() - timedelta(minutesminutes)).isoformat() Z # 构建查询过滤器查找最近更新且“视频链接”属性为空 filter_params { and: [ { timestamp: last_edited_time, last_edited_time: { on_or_after: since_time } }, { property: Video, # 假设你数据库里有一个“Video”属性URL类型 url: { is_empty: True } } ] } try: response notion.databases.query( database_idDATABASE_ID, filterfilter_params ) return response.get(results, []) except Exception as e: print(f查询Notion数据库失败: {e}) return []代码解释我们使用notion_client库与Notion API交互。get_recent_notes函数会查找在最近5分钟内被编辑过并且“Video”字段为空的笔记。你可以根据需要调整时间窗口。这里假设你的Notion数据库有一个名为“Video”的URL类型属性用于存放生成视频的链接。你需要在Notion数据库中提前创建好这个属性。3.2 第二步调用CogVideoX-2b生成视频接下来我们编写调用本地CogVideoX服务的函数。由于CogVideoX-2b的WebUI通常通过Gradio构建我们可以模拟前端请求来触发生成。def generate_video_from_text(prompt, note_title): 调用CogVideoX-2b服务生成视频并保存文件。 返回生成视频的本地文件路径或可访问URL。 # 优化提示词将中文标题和内容结合并建议使用英文 # 这里做一个简单处理如果提示词是中文可以添加一个英文翻译或关键词 enhanced_prompt f{prompt}. {note_title}. cinematic, high quality, 4k # 在实际应用中你可能需要一个更复杂的中译英逻辑但模型对中文也有一定理解 # 构造请求数据根据CogVideoX WebUI的实际接口调整 # 注意这里需要你根据实际部署的WebUI的API格式来调整以下为示例 payload { prompt: enhanced_prompt, num_frames: 24, # 生成帧数影响视频长度 width: 512, # 视频宽度 height: 512, # 视频高度 seed: -1, # 随机种子-1表示随机 } print(f正在为笔记《{note_title}》生成视频提示词: {enhanced_prompt[:50]}...) try: # 示例假设WebUI的生成接口是 /api/generate # 你需要查看CogVideoX镜像的具体API文档或通过浏览器开发者工具查看网络请求 response requests.post(f{COGVIDEOX_URL}/api/generate, jsonpayload, timeout300) # 设置长超时 response.raise_for_status() # 假设接口返回一个包含视频文件路径或URL的JSON result response.json() video_url result.get(video_url) # 例如 http://127.0.0.1:7860/file/videos/abc.mp4 # 或者可能是本地路径需要转换为可通过Web访问的URL # 这里我们假设服务直接返回一个可访问的HTTP链接 if video_url: print(f视频生成成功: {video_url}) return video_url else: print(生成成功但未获取到视频链接。) return None except requests.exceptions.Timeout: print(视频生成请求超时这可能需要更长时间。) return None except Exception as e: print(f调用视频生成服务失败: {e}) return None重要提示这部分代码的payload和请求端点 (/api/generate) 是示例。CogVideoX-2b WebUI的实际API可能不同。你需要通过打开浏览器开发者工具F12在WebUI上点击生成按钮查看它实际发送的网络请求来确定正确的API地址和参数格式。这是最关键的一步。生成视频需要2-5分钟所以设置了较长的超时时间。3.3 第三步将视频链接更新回Notion生成视频后我们需要把链接写回对应的Notion笔记。def update_note_with_video(note_id, video_url): 更新指定的Notion笔记添加视频链接。 update_data { Video: { # 属性名需要与你的数据库中的“Video”属性名完全一致 url: video_url }, # 你也可以同时更新其他属性比如状态 Status: { # 假设有一个“Status”选择属性 select: { name: 视频已生成 } } } try: notion.pages.update(page_idnote_id, propertiesupdate_data) print(f已更新笔记 {note_id}添加视频链接。) return True except Exception as e: print(f更新Notion笔记失败: {e}) return False3.4 第四步组装主循环实现自动化最后我们把所有功能串联起来形成一个可以持续运行的守护进程。def extract_note_content(notion_page): 从Notion页面对象中提取标题和文本内容。 这是一个简化示例实际可能需要递归解析复杂的块结构。 title 无标题 content_text # 提取标题假设标题在“Name”属性 if hasattr(notion_page[properties].get(Name), title): title_list notion_page[properties][Name][title] if title_list: title title_list[0][plain_text] # 提取页面正文内容需要调用blocks API try: blocks notion.blocks.children.list(block_idnotion_page[id]) for block in blocks.get(results, []): block_type block[type] if block_type paragraph: # 提取段落文本 rich_text block[paragraph].get(rich_text, []) for text in rich_text: content_text text.get(plain_text, ) # 可以继续处理其他块类型如标题、列表等 except Exception as e: print(f提取页面内容时出错: {e}) # 如果内容太长可以截取前一部分作为视频提示词 if len(content_text) 500: content_text content_text[:500] ... return title, content_text.strip() def main_loop(): 主循环每隔一段时间检查Notion处理新笔记。 print(启动Notion到CogVideoX同步守护进程...) check_interval 60 # 检查间隔单位秒 while True: print(f\n[{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}] 检查新笔记...) recent_notes get_recent_notes(minutes10) # 查找过去10分钟内的笔记 if not recent_notes: print(未发现需要处理的新笔记。) else: print(f发现 {len(recent_notes)} 条待处理笔记。) for note in recent_notes: note_id note[id] title, content extract_note_content(note) if not content: print(f笔记《{title}》内容为空跳过。) continue print(f处理笔记: 《{title}》) # 生成视频 video_url generate_video_from_text(content, title) if video_url: # 更新Notion success update_note_with_video(note_id, video_url) if success: print(f笔记《{title}》处理完成) else: print(f笔记《{title}》视频生成成功但更新Notion失败。) else: print(f笔记《{title}》视频生成失败。) # 处理完一条笔记后可以稍作停顿避免对GPU请求过载 time.sleep(30) # 等待下一次检查 time.sleep(check_interval) if __name__ __main__: # 你可以选择直接运行主循环或者使用此脚本作为函数被其他调度器调用 # main_loop() # 或者先测试单次运行 test_single_run()为了测试我们可以先写一个测试函数def test_single_run(): 测试单次运行流程 print( 测试单次运行 ) notes get_recent_notes(minutes30) # 查找最近30分钟的笔记 if notes: note notes[0] note_id note[id] title, content extract_note_content(note) print(f测试笔记: 《{title}》) print(f内容预览: {content[:100]}...) # 实际运行时取消注释下面两行 # video_url generate_video_from_text(content, title) # if video_url: # update_note_with_video(note_id, video_url) else: print(没有找到可用于测试的近期笔记。)4. 部署与优化建议脚本写好了怎么让它长期稳定运行呢在AutoDL实例上运行既然CogVideoX服务跑在AutoDL上我们可以把同步脚本也放在同一个实例上运行避免网络延迟。使用nohup或tmux让脚本在后台运行。tmux new -s notion-sync python notion_video_sync.py # 按 CtrlB, 然后按 D 分离会话处理视频存储生成的视频文件需要有一个固定的存储和访问位置。确保CogVideoX服务生成的视频文件保存在一个可以通过Web服务器访问的目录例如WebUI的static或output目录并且返回的URL是有效的。错误处理与日志在生产环境中需要增强错误处理如网络重试、API限流处理并将日志写入文件方便排查问题。提示词优化为了获得更好的视频效果可以对从Notion提取的原始文本进行清洗和优化。例如提取关键词、总结核心观点、翻译成效果更佳的英文提示词等。你可以集成一个简单的文本处理API或库来实现。使用更可靠的调度对于重要应用可以用系统的cron定时任务来定期执行脚本而不是使用脚本内的while循环这样更易于管理。5. 总结通过以上步骤我们成功搭建了一个连接Notion知识库和CogVideoX-2b视频生成能力的自动化桥梁。这个方案的核心思路清晰监听定期检查Notion数据库的变更。转换将文本内容转化为视频生成提示词并调用本地模型服务。反馈将成果视频链接写回原笔记形成闭环。它不仅仅是一个技术Demo更是一个可以切实提升内容工作流的生产力工具。你可以在此基础上进行无限扩展例如支持更多笔记源如飞书文档、语雀。增加视频风格选择通过在Notion中添加属性。实现批量处理历史笔记。将生成的视频自动发布到视频平台或社交媒体。技术的价值在于解决实际问题。希望这个从笔记到视频的自动化方案能为你打开一扇新的效率之门。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。