2026/8/4 14:51:10

从Notion到RIVALS:构建自动化数据流水线生成集锦素材

从Notion到RIVALS:构建自动化数据流水线生成集锦素材 在实际项目开发中我们经常需要将 Notion 这类知识管理工具中的结构化数据与 RIVALS 这类游戏或竞技数据分析平台的内容进行整合以制作出高质量的集锦视频或数据可视化报告。这个过程涉及数据提取、格式转换、自动化脚本编写和最终渲染如果手动操作不仅效率低下也容易出错。本文将围绕如何构建一个从 Notion 到 RIVALS 数据集的自动化处理流程最终生成一个可用的集锦素材包Montage展开。本文适合有一定 Python 或 Node.js 基础希望将 Notion 作为数据源并自动化处理游戏、体育或任何竞技类数据的开发者。我们将从理解 Notion API 和 RIVALS 数据格式开始逐步完成环境搭建、数据抓取、清洗转换、到最终生成可用于视频剪辑软件如 Adobe Premiere, DaVinci Resolve或数据可视化库如 Matplotlib, Chart.js的素材和脚本的全过程。学完后你将能搭建一个可复用的自动化流水线将 Notion 中的比赛记录、选手数据、精彩时刻描述自动转换为时间线文件、字幕文件或数据图表。1. 理解 Notion API 与 RIVALS 数据集的桥梁作用在开始写代码之前必须理清数据从哪里来要到哪里去以及中间需要经历哪些变形。Notion 作为数据源其优势在于灵活的数据结构和便捷的协作编辑而“RIVALS Montage”的目标产出通常是视频片段、时间线标记或数据图表这些都需要高度结构化和时间序列化的数据。1.1 Notion 数据库你的结构化数据仓库Notion 的核心是数据库Database。一个用于记录“RIVALS”比赛数据的典型数据库可能包含以下属性Properties比赛名称 (Title): 主标题字段。比赛日期 (Date): 记录事件发生时间。参赛选手/队伍 (Multi-select 或 Relation): 记录对战的双方。精彩时刻时间点 (Text 或 Number): 例如01:23.456表示视频中的时间戳。时刻描述 (Text): 对该精彩时刻的文字描述用于生成字幕。数据指标 (Number): 如击杀数、得分、反应时间等。关联媒体文件 (Files media): 可能存储了原始录像或截图。标签 (Multi-select): 如“绝杀”、“逆转”、“高光操作”等用于分类筛选。Notion API 允许我们以编程方式查询这些数据。API 返回的是 JSON 格式的原始数据其中包含了数据库的结构Schema和每一条记录Page的具体内容。我们的第一个任务就是学会如何认证并请求这些数据。1.2 RIVALS Montage 的最终产物时间线、字幕与数据处理后的数据需要服务于最终的集锦制作。通常我们需要生成以下几种格式的文件EDL (Edit Decision List) 或 XML 时间线文件: 供专业视频剪辑软件如 DaVinci Resolve, Premiere识别用于自动在时间线上标记片段或进行粗剪。它本质上是一个包含入点In Point、出点Out Point和片段描述的文件。SRT 或 ASS 字幕文件: 包含时间戳和对应文字的字幕文件可以直接导入视频剪辑软件或播放器为精彩时刻添加说明。CSV/JSON 数据文件: 包含清洗后的结构化数据可用于导入数据分析工具如 Pandas Matplotlib生成战绩走势图、雷达图等可视化图表。剪辑清单 (Markdown/Text): 一个给人看的、格式化的清单列出所有需要剪辑的片段及其描述方便剪辑师手动操作。我们的自动化脚本核心就是将 Notion API 返回的 JSON经过解析和转换生成上述一种或多种文件。1.3 技术选型与工作流设计整个流程可以设计为一个线性的工作流Notion Database-Notion API Client-数据清洗与转换脚本-输出生成器-EDL/XML/SRT/CSV 文件根据你的技术栈可以选择不同的工具链Python 方案: 使用notion-client库配合pandas进行数据处理用内置的csv、json模块输出文件。适合数据清洗和复杂分析。Node.js 方案: 使用官方notionhq/clientSDK配合一些时间处理库如dayjs直接生成文件。适合与前端可视化或构建工具链集成。本文将主要使用Python方案进行演示因为其在数据处理和脚本编写上对初学者更为友好和直观。2. 环境准备与依赖配置在开始编码前需要准备好开发环境和必要的访问凭证。2.1 创建 Notion 集成并获取密钥访问 Notion Developers 页面并登录。点击 “ New integration” 创建一个新的集成。填写名称如RIVALS Montage Generator并选择关联的工作区。点击 “Submit” 创建。创建成功后页面会显示 “Internal Integration Token”。复制并妥善保存这个令牌它相当于你的密码。在同一个页面你还可以配置集成的能力比如“Read content”、“Update content”等。对于只读的数据抓取勾选“Read content”即可。2.2 将集成分享给你的数据库仅有令牌还不够必须让这个集成有权限访问你存储数据的特定数据库。打开你的 Notion 数据库页面。点击页面右上角的 “...” 菜单选择 “Add connections”。在弹出的列表中找到你刚刚创建的集成如RIVALS Montage Generator并点击它。现在你的集成就可以读取这个数据库的内容了。记下数据库的ID。数据库的URL通常格式为https://www.notion.so/yourworkspace/{database_id}?v...。复制{database_id}部分一串32位的16进制数字。2.3 配置 Python 开发环境确保你的电脑上安装了 Python建议 3.8 及以上版本。我们将使用pip安装必要的包。创建一个新的项目目录并初始化虚拟环境推荐mkdir notion-rivals-montage cd notion-rivals-montage python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate安装核心依赖库pip install notion-client pandasnotion-client: 非官方的 Notion API 客户端API 设计更 Pythonic。pandas: 强大的数据处理库用于清洗、筛选和转换数据。2.4 安全存储配置信息永远不要将令牌和数据库ID硬编码在代码中或上传到公开仓库。推荐使用环境变量或配置文件。创建一个.env文件确保在.gitignore中忽略它# .env NOTION_TOKENyour_secret_integration_token_here NOTION_DATABASE_IDyour_database_id_here然后安装python-dotenv来读取它pip install python-dotenv3. 构建数据抓取与清洗脚本现在我们开始编写核心脚本。我们将创建一个montage_generator.py文件。3.1 初始化客户端与读取数据库首先读取环境变量并初始化 Notion 客户端。# montage_generator.py import os from notion_client import Client from dotenv import load_dotenv import pandas as pd # 加载 .env 文件中的环境变量 load_dotenv() # 初始化 Notion 客户端 notion Client(authos.environ.get(NOTION_TOKEN)) database_id os.environ.get(NOTION_DATABASE_ID) def fetch_notion_database(): 从 Notion 数据库获取原始数据 try: # 查询数据库可以添加筛选和排序 response notion.databases.query( database_iddatabase_id, # 示例筛选出“标签”包含“高光”的记录并按“比赛日期”升序排列 # filter{ # property: 标签, # multi_select: { # contains: 高光 # } # }, # sorts[{ # property: 比赛日期, # direction: ascending # }] ) return response.get(results, []) except Exception as e: print(f获取 Notion 数据失败: {e}) return [] if __name__ __main__: pages fetch_notion_database() print(f成功获取到 {len(pages)} 条记录) # 打印第一条记录看看结构 if pages: print(pages[0])运行这个脚本python montage_generator.py如果配置正确你会看到打印出的第一条记录的复杂 JSON 结构。这证明了连接成功。3.2 解析 Notion API 的复杂响应Notion API 返回的数据是嵌套很深的 JSON。我们需要编写辅助函数来提取不同类型属性的值。# montage_generator.py (续) def extract_property_value(page, property_name, property_typeNone): 从 Notion Page 对象中提取指定属性的值 props page.get(properties, {}) prop props.get(property_name) if not prop: return None # 根据属性类型提取值 prop_type prop.get(type) if prop_type title: # 标题属性 titles prop.get(title, []) return .join([t.get(plain_text, ) for t in titles]) if titles else None elif prop_type rich_text: # 富文本属性 rich_texts prop.get(rich_text, []) return .join([rt.get(plain_text, ) for rt in rich_texts]) if rich_texts else None elif prop_type number: # 数字属性 return prop.get(number) elif prop_type date: # 日期属性 date_obj prop.get(date) if date_obj: return date_obj.get(start) # 返回 ISO 格式字符串 return None elif prop_type multi_select: # 多选属性 options prop.get(multi_select, []) return [opt.get(name) for opt in options] elif prop_type files: # 文件属性 files prop.get(files, []) # 返回第一个文件的 URL如果有 return files[0].get(file, {}).get(url) if files else None # 可以继续添加其他属性类型如 select, people, checkbox 等 else: print(f警告: 未处理的属性类型 {prop_type} 对于属性 {property_name}) return None def parse_pages_to_df(pages): 将 Notion Page 列表解析为 pandas DataFrame data [] for page in pages: page_id page.get(id) row { id: page_id, 比赛名称: extract_property_value(page, 比赛名称) or extract_property_value(page, Name), # 有些数据库主属性叫 Name 比赛日期: extract_property_value(page, 比赛日期), 参赛选手: extract_property_value(page, 参赛选手), # 假设是 multi-select 时间点: extract_property_value(page, 精彩时刻时间点), # 格式如 01:23.456 时刻描述: extract_property_value(page, 时刻描述), 击杀数: extract_property_value(page, 击杀数), 标签: extract_property_value(page, 标签), } data.append(row) return pd.DataFrame(data) # 在主函数中使用 if __name__ __main__: pages fetch_notion_database() df parse_pages_to_df(pages) print(df.head()) # 查看前几行数据 print(df.info()) # 查看数据概览3.3 数据清洗与格式标准化从 Notion 提取的数据可能存在缺失值、格式不一致等问题。我们需要进行清洗特别是时间点数据。# montage_generator.py (续) def clean_and_transform_data(df): 清洗和转换数据 df_clean df.copy() # 1. 处理时间点将 MM:SS.mmm 或 HH:MM:SS.mmm 转换为总秒数浮点数 def time_str_to_seconds(t): if pd.isna(t): return None try: parts list(map(float, t.split(:))) if len(parts) 3: # HH:MM:SS h, m, s parts return h * 3600 m * 60 s elif len(parts) 2: # MM:SS m, s parts return m * 60 s else: return float(t) # 可能已经是秒数 except ValueError: print(f无法解析时间点: {t}) return None df_clean[时间点_秒] df_clean[时间点].apply(time_str_to_seconds) # 2. 假设每个片段持续3秒计算结束时间点用于EDL df_clean[结束时间点_秒] df_clean[时间点_秒].apply(lambda x: x 3.0 if pd.notna(x) else None) # 3. 将秒数转换回 EDL 需要的格式: HH:MM:SS:FF (假设帧率 25fps) # 注意不同帧率计算方式不同这里以25fps为例 FRAME_RATE 25 def seconds_to_edl_time(total_seconds): if total_seconds is None: return 00:00:00:00 hours int(total_seconds // 3600) minutes int((total_seconds % 3600) // 60) seconds int(total_seconds % 60) frames int((total_seconds - int(total_seconds)) * FRAME_RATE) return f{hours:02d}:{minutes:02d}:{seconds:02d}:{frames:02d} df_clean[入点] df_clean[时间点_秒].apply(seconds_to_edl_time) df_clean[出点] df_clean[结束时间点_秒].apply(seconds_to_edl_time) # 4. 处理多选字段将列表转换为逗号分隔的字符串方便CSV输出 df_clean[参赛选手_str] df_clean[参赛选手].apply(lambda x: , .join(x) if isinstance(x, list) else x) df_clean[标签_str] df_clean[标签].apply(lambda x: , .join(x) if isinstance(x, list) else x) # 5. 按时间点排序 df_clean df_clean.sort_values(by时间点_秒).reset_index(dropTrue) return df_clean if __name__ __main__: pages fetch_notion_database() df_raw parse_pages_to_df(pages) df_clean clean_and_transform_data(df_raw) print(df_clean[[比赛名称, 时间点, 时间点_秒, 入点, 出点, 时刻描述]].head())4. 生成最终集锦素材文件数据清洗完毕后我们就可以将其转换为各种目标格式了。4.1 生成 EDL (Edit Decision List) 文件EDL 是一种简单的文本格式许多专业软件都支持。一个基本的 EDL 格式如下TITLE: RIVALS Highlights Montage FCM: NON-DROP FRAME 001 AX V C 00:00:00:00 00:00:00:00 00:01:23:12 00:01:23:12 * FROM CLIP NAME: 精彩瞬间1 * 选手: PlayerA, PlayerB * 描述: 这是一次精彩的绝杀。 002 AX V C 00:00:00:00 00:00:00:00 00:02:15:04 00:02:15:04 * FROM CLIP NAME: 精彩瞬间2 ...我们来编写生成函数# montage_generator.py (续) def generate_edl(df, output_pathoutput/montage.edl): 生成 EDL 文件 os.makedirs(os.path.dirname(output_path), exist_okTrue) with open(output_path, w, encodingutf-8) as f: f.write(TITLE: RIVALS Highlights Montage\n) f.write(FCM: NON-DROP FRAME\n\n) for idx, row in df.iterrows(): if pd.isna(row[入点]) or pd.isna(row[出点]): continue # EDL 行: 序号, 轨道, 类型, 转换, 源入点, 源出点, 时间线入点, 时间线出点 # 这里简化处理假设源文件从头开始时间线按顺序拼接 timeline_in seconds_to_edl_time(idx * 3.0) # 简化每个片段在时间线上间隔3秒 timeline_out seconds_to_edl_time(idx * 3.0 3.0) f.write(f{idx1:03d} AX V C {row[入点]} {row[出点]} {timeline_in} {timeline_out}\n) f.write(f* FROM CLIP NAME: {row.get(比赛名称, Clip)}\n) if row.get(参赛选手_str): f.write(f* 选手: {row[参赛选手_str]}\n) if row.get(时刻描述): f.write(f* 描述: {row[时刻描述]}\n) f.write(\n) print(fEDL 文件已生成: {output_path})4.2 生成 SRT 字幕文件SRT 格式更通用可以被绝大多数视频播放器和剪辑软件识别。# montage_generator.py (续) def seconds_to_srt_time(total_seconds): 将秒数转换为 SRT 时间格式: HH:MM:SS,mmm if total_seconds is None: return 00:00:00,000 hours int(total_seconds // 3600) minutes int((total_seconds % 3600) // 60) seconds int(total_seconds % 60) milliseconds int((total_seconds - int(total_seconds)) * 1000) return f{hours:02d}:{minutes:02d}:{seconds:02d},{milliseconds:03d} def generate_srt(df, output_pathoutput/subtitles.srt): 生成 SRT 字幕文件 os.makedirs(os.path.dirname(output_path), exist_okTrue) with open(output_path, w, encodingutf-8) as f: for idx, row in df.iterrows(): if pd.isna(row[时间点_秒]): continue start_sec row[时间点_秒] end_sec start_sec 3.0 # 假设字幕显示3秒 start_time seconds_to_srt_time(start_sec) end_time seconds_to_srt_time(end_sec) f.write(f{idx1}\n) f.write(f{start_time} -- {end_time}\n) description row.get(时刻描述, ) players row.get(参赛选手_str, ) subtitle_text description if players: subtitle_text f[{players}] {subtitle_text} f.write(f{subtitle_text}\n\n) print(fSRT 字幕文件已生成: {output_path})4.3 生成数据报告 (CSV) 和剪辑清单# montage_generator.py (续) def generate_outputs(df): 生成所有输出文件 # 1. 生成清洗后的数据 CSV csv_path output/rivals_data.csv df.to_csv(csv_path, indexFalse, encodingutf-8-sig) print(f数据 CSV 已生成: {csv_path}) # 2. 生成给人看的剪辑清单 (Markdown) md_path output/edit_notes.md with open(md_path, w, encodingutf-8) as f: f.write(# RIVALS 集锦剪辑清单\n\n) f.write(f共 {len(df)} 个片段\n\n) for idx, row in df.iterrows(): f.write(f## 片段 {idx1}\n) f.write(f- **比赛**: {row.get(比赛名称, N/A)}\n) f.write(f- **时间点**: {row.get(时间点, N/A)}\n) f.write(f- **选手**: {row.get(参赛选手_str, N/A)}\n) f.write(f- **描述**: {row.get(时刻描述, N/A)}\n) f.write(f- **标签**: {row.get(标签_str, N/A)}\n) f.write(f- **数据(击杀)**: {row.get(击杀数, N/A)}\n) f.write(\n---\n\n) print(f剪辑清单已生成: {md_path}) # 3. 调用生成EDL和SRT的函数 generate_edl(df) generate_srt(df)4.4 整合主流程最后我们将所有步骤串联起来。# montage_generator.py (续) def main(): 主函数执行完整流程 print(开始从 Notion 获取 RIVALS 数据...) pages fetch_notion_database() if not pages: print(未获取到数据请检查令牌、数据库ID和网络连接。) return print(解析 Notion 数据...) df_raw parse_pages_to_df(pages) print(清洗和转换数据...) df_clean clean_and_transform_data(df_raw) print(生成输出文件...) generate_outputs(df_clean) print( 处理完成 ) print(f总计处理了 {len(df_clean)} 个精彩时刻。) print(生成的文件位于 output/ 目录下。) if __name__ __main__: main()运行python montage_generator.py如果一切顺利你将在项目根目录下的output/文件夹中找到生成的所有文件。5. 运行验证与结果分析脚本运行成功后你需要验证生成的文件是否有效以及数据是否正确。5.1 文件输出验证检查output/目录应该包含以下文件rivals_data.csv: 包含所有清洗后数据的表格可以用 Excel 或 Numbers 打开查看。edit_notes.md: Markdown 格式的剪辑清单清晰列出了每个片段的信息。montage.edl: EDL 时间线文件。subtitles.srt: SRT 字幕文件。5.2 验证数据准确性打开 CSV 文件检查“时间点_秒”列是否被正确计算。对比 Notion 中原始的“MM:SS.mmm”格式看转换是否正确。打开 SRT 文件用文本编辑器打开检查时间戳格式HH:MM:SS,mmm是否正确字幕文本是否与描述一致。导入视频软件测试EDL: 在 DaVinci Resolve 中进入“剪辑”页面在媒体池右键 - 时间线 - 导入 - AAFEDLXML...选择生成的.edl文件。如果成功时间线上会出现一系列标记好的片段。SRT: 在 Premiere Pro 或 DaVinci Resolve 中直接将.srt文件拖入时间线或字幕轨道检查字幕是否能正确加载并与视频对齐。5.3 验证逻辑与参数调整第一次运行时最常见的偏差是时间计算。你需要根据你的源视频和需求调整两个关键参数片段持续时间在clean_and_transform_data函数中我们假设每个高光片段持续3.0秒。如果你的片段长度不一需要在 Notion 数据库中增加一个“持续时间”字段并在脚本中读取它。帧率 (Frame Rate)在seconds_to_edl_time函数中我们假设帧率为25FPS。你必须根据你的视频项目的帧率修改这个值。常见的还有 24电影、30、60 FPS。错误的帧率会导致 EDL 文件中的时间码不准确。6. 常见问题排查在开发和运行过程中你可能会遇到以下问题。6.1 Notion API 连接失败问题现象可能原因检查方式处理建议运行脚本报错notion_client.errors.APIResponseError或返回空数据1. 集成令牌无效或过期。2. 数据库ID错误。3. 集成未被分享到目标数据库。4. 数据库属性名与代码中提取的名称不匹配。1. 检查.env文件中的NOTION_TOKEN和NOTION_DATABASE_ID是否正确。2. 在 Notion 集成页面确认集成状态为 “Active”。3. 在数据库页面确认已连接该集成。4. 打印pages变量的原始 JSON查看properties下的确切属性名。1. 重新生成集成令牌并更新.env。2. 复制正确的数据库ID。3. 在数据库页面重新连接集成。4. 修改extract_property_value函数中使用的属性名使其与 Notion 数据库中的属性名完全一致注意大小写和空格。6.2 数据解析错误或为空问题现象可能原因检查方式处理建议df_clean数据框为空或某些列为空。1. 数据库中没有数据。2. 属性类型判断错误。3. 时间格式解析失败。1. 运行print(df_raw.head())和print(df_raw.columns)查看原始解析结果。2. 检查extract_property_value函数是否涵盖了数据库中所有用到的属性类型如select,checkbox。3. 打印出原始的时间字符串row[‘时间点’]看是否符合MM:SS.mmm格式。1. 在 Notion 数据库中添加测试数据。2. 根据 Notion API 文档补充对新属性类型的解析逻辑。3. 修改time_str_to_seconds函数增加对更多时间格式如1:23.45的兼容性或先清洗 Notion 中的数据格式。6.3 生成的文件无法被软件识别问题现象可能原因检查方式处理建议EDL 文件导入后时间码错乱或无法识别。1. 帧率设置错误。2. 时间码格式不符合软件要求。3. EDL 文件头信息缺失或错误。1. 确认你的视频项目设置的帧率并与seconds_to_edl_time函数中的FRAME_RATE变量对比。2. 用文本编辑器打开 EDL 文件检查时间码格式是否为HH:MM:SS:FF。3. 查阅你所使用视频软件的官方文档看其对 EDL 格式的具体要求。1. 将FRAME_RATE变量修改为正确的值。2. 尝试在 EDL 文件头添加或修改FCM: DROP FRAME或FCM: NON-DROP FRAME这取决于你的视频是丢帧还是非丢帧时间码。3. 使用更简单的测试数据如整秒数生成 EDL先排除时间计算错误的可能。SRT 文件导入后字幕不同步。1. 视频的起始时间不是 00:00:00。2. 字幕显示时长固定为3秒与实际片段长度不符。1. 检查视频文件本身的元数据或时间线起点。2. 对比 SRT 文件中的时间戳与视频中实际事件发生的时间。1. 如果视频有片头需要在计算时间点时进行偏移。可以在seconds_to_srt_time函数中加入一个全局偏移量offset_seconds。2. 在 Notion 中增加“片段时长”字段并在生成 SRT 时使用该字段计算结束时间而不是固定的3秒。7. 最佳实践与扩展方向一个能稳定运行的基础脚本已经完成但要用于实际生产或更复杂的场景还需要考虑以下几点。7.1 生产环境建议错误处理与日志当前的脚本只有基础的try...except。在生产环境中应该引入logging模块对不同级别的信息INFO, WARNING, ERROR进行记录并写入文件方便后期排查。配置化管理将数据库字段映射、时间偏移、帧率、输出路径等参数提取到独立的配置文件如config.yaml中避免硬编码。增量处理如果数据库数据量很大每次全量拉取效率低。可以在 Notion 数据库中增加“最后导出时间”属性或者利用 API 的last_edited_time进行增量查询。自动化调度使用系统的定时任务如 Linux 的cron Windows 的“任务计划程序”或 CI/CD 工具如 GitHub Actions定期运行此脚本实现自动化素材更新。结果通知脚本运行完成后可以通过邮件、Slack 或钉钉机器人发送通知告知处理结果和文件链接。7.2 功能扩展方向支持更多输出格式XML for FCPX/ Premiere: 生成 Final Cut Pro XML 或 Adobe Premiere Pro 的.xml项目文件兼容性更好。JSON for Data Visualization: 生成一个结构化的 JSON 文件直接供前端图表库如 ECharts, D3.js使用制作交互式数据看板。集成视频处理结合moviepy或ffmpeg-python库根据 EDL 信息直接对源视频进行自动剪切和拼接生成初步的集锦视频草稿。丰富数据源除了 Notion可以考虑从其他平台如特定游戏的 API、直播平台录像列表拉取数据统一处理后生成集锦。构建 Web 界面使用 Flask 或 FastAPI 将脚本包装成一个简单的 Web 服务提供上传 Notion 数据库链接、配置参数、预览并下载生成文件的界面。7.3 代码优化清单在将脚本投入正式使用前建议按此清单检查[ ] 敏感信息令牌、ID是否已移出代码使用环境变量或密钥管理服务。[ ] 是否对 Notion API 的请求频率做了限制避免触发速率限制。[ ] 是否处理了所有可能为None的数据字段避免程序因空值崩溃。[ ] 生成的文件路径是否考虑了跨平台兼容性使用os.path.join。[ ] 输出目录output/是否存在如果不存在是否会自动创建。[ ] 时间、数值的转换函数是否考虑了各种边界情况如负数、超大数、格式错误的字符串。[ ] 是否添加了足够的日志输出能清晰看到脚本运行的每个阶段。通过以上步骤你不仅完成了一个从 Notion 到 RIVALS Montage 的自动化数据流水线更掌握了一套将非结构化或半结构化数据转化为特定领域可用素材的方法论。这个模式可以灵活应用到内容创作、数据分析报告生成等多个场景中。