CogVideoX-2b应用拓展：对接Notion API实现笔记→短视频自动同步-Seo优化-合肥市网站建设公司

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 requests

3.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(auth=NOTION_TOKEN) def get_recent_notes(minutes=5): """ 获取最近N分钟内创建或更新的、且没有视频附件的笔记。 """ # 计算时间点 since_time = (datetime.utcnow() - timedelta(minutes=minutes)).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_id=DATABASE_ID, filter=filter_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", json=payload, timeout=300) # 设置长超时 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_id=note_id, properties=update_data) print(f"已更新笔记 {note_id}，添加视频链接。") return True except Exception as e: print(f"更新Notion笔记失败: {e}") return False

3.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_id=notion_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(minutes=10) # 查找过去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(minutes=30) # 查找最近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 # 按 Ctrl+B, 然后按 D 分离会话
```
处理视频存储：生成的视频文件需要有一个固定的存储和访问位置。确保CogVideoX服务生成的视频文件保存在一个可以通过Web服务器访问的目录（例如，WebUI的static或output目录），并且返回的URL是有效的。
错误处理与日志：在生产环境中，需要增强错误处理（如网络重试、API限流处理），并将日志写入文件，方便排查问题。
提示词优化：为了获得更好的视频效果，可以对从Notion提取的原始文本进行清洗和优化。例如，提取关键词、总结核心观点、翻译成效果更佳的英文提示词等。你可以集成一个简单的文本处理API或库来实现。
使用更可靠的调度：对于重要应用，可以用系统的cron定时任务来定期执行脚本，而不是使用脚本内的while循环，这样更易于管理。