【OpenClaw】视频批量生成

️Tavily Search定义一个专为 AI Agent 优化的搜索引擎。使用准备在ClawHub中搜索下载将skill文件保存到~/.openclaw/workspace/skills/中在Tavily API 平台获取API Key/Users/dudumac003/.openclaw/workspace-plot/fetch_hotspots.pyimport json import re import requests from datetime import datetime from pathlib import Path def fetch_and_parse_hotspots(): # 调用Tavily API url https://api.tavily.com/search headers { Content-Type: application/json, Authorization: Bearer tvly-dev-……………………………………………… } payload { query: 小红书热榜 微博热搜榜, search_depth: advanced, max_results: 15 } print(正在获取热榜数据...) response requests.post(url, headersheaders, jsonpayload) data response.json() # 解析热榜数据 hotspots { date: datetime.now().strftime(%Y-%m-%d), timestamp: datetime.now().isoformat(), weibo: [], xiaohongshu: [], } # 从搜索结果中提取热榜数据 for result in data.get(results, []): title result.get(title, ) url_text result.get(url, ).lower() content result.get(content, ) # 提取微博热搜 - 从 tophub.today 的微博数据 if 微博 in title and tophub in url_text: hotspots[weibo] extract_weibo_from_tophub(content) print(f✅ 从微博热榜提取到 {len(hotspots[weibo])} 条) # 提取小红书热榜 - 从 tophub.today 的小红书数据 if 小红书 in title and tophub in url_text: hotspots[xiaohongshu] extract_xiaohongshu_from_tophub(content) print(f✅ 从小红书热榜提取到 {len(hotspots[xiaohongshu])} 条) # 保存结果 output_dir Path(/Users/dudumac003/.openclaw/workspace-plot/hotspot) output_dir.mkdir(parentsTrue, exist_okTrue) output_file output_dir / f{datetime.now().strftime(%Y-%m-%d)}.json with open(output_file, w, encodingutf-8) as f: json.dump(hotspots, f, ensure_asciiFalse, indent2) print(f\n✅ 热榜数据已保存到: {output_file}) print(f 最终统计:) print(f 微博热搜: {len(hotspots[weibo])} 条) print(f 小红书热榜: {len(hotspots[xiaohongshu])} 条) def extract_weibo_from_tophub(content): 从 tophub.today 的微博热榜内容中提取数据 hotspots [] # 匹配 Markdown 表格格式: | 1. | 标题 | 热度 | 图标 | # 例如: | 1. | 乘风2026定档 | 113万 |  | pattern r\|\s*(\d)\.\s*\|\s*([^|]?)\s*\|\s*(\d\.?\d*万)\s*\| matches re.findall(pattern, content) for rank, title, heat in matches: title title.strip() if title and len(title) 2: hotspots.append({ rank: int(rank), title: title, heat: heat, source: 微博热搜 }) if len(hotspots) 20: break return hotspots def extract_xiaohongshu_from_tophub(content): 从 tophub.today 的小红书热榜内容中提取数据 hotspots [] # 匹配 Markdown 表格格式: | 1. | 标题 | 热度 | 图标 | # 例如: | 1. | 用万能旅行拍照姿势美美出片 | 909.2w |  | pattern r\|\s*(\d)\.\s*\|\s*([^|]?)\s*\|\s*(\d\.?\d*[wW万])\s*\| matches re.findall(pattern, content) for rank, title, heat in matches: title title.strip() if title and len(title) 2: hotspots.append({ rank: int(rank), title: title, heat: heat, source: 小红书热榜 }) if len(hotspots) 20: break # 如果没有匹配到尝试备用格式有时表格格式略有不同 if not hotspots: pattern2 r\|\s*(\d)\.\s*\|\s*([^|]?)\s*\|\s*([^|]?)\s*\| matches2 re.findall(pattern2, content) for rank, title, heat in matches2: if w in heat or 万 in heat: title title.strip() if title and len(title) 2: hotspots.append({ rank: int(rank), title: title, heat: heat.strip(), source: 小红书热榜 }) if len(hotspots) 20: break return hotspots if __name__ __main__: fetch_and_parse_hotspots()执行代码,搜集热点python3 /Users/dudumac003/.openclaw/workspace-plot/fetch_hotspots.py剧本大师创建,分配独立工作区openclaw agents add plot --workspace ~/.openclaw/workspace-plot设置身份openclaw agents set-identity --agent plot --name 剧本大师 --emoji ✍️/Users/dudumac003/.openclaw/workspace-plot/prompt.md# 角色设定 你是专业的短视频剧本专家专注于创作15秒“嘟嘟巴士”线路推广剧本。你的作品需要兼具传播力与品牌调性让观众在15秒内被吸引、记住品牌、产生出行欲望。 # 输入数据 - 热榜数据/Users/dudumac003/.openclaw/workspace-plot/hotspot/2026-03-27.json - 价格数据/Users/dudumac003/.openclaw/workspace-plot/dudu_prices.json # 核心创作要求 ## 1. 车辆呈现规范必须严格遵守 剧本中必须出现一辆**红色旅游大巴车**车身侧面清晰可见“嘟嘟巴士”字样。车辆特征 - **车型**旅游大巴非市内公交车 - **车内场景**必须体现旅游大巴内部特征——独立软座座椅、座椅上方行李架、宽体过道、窗帘、空调出风口 - **严禁出现**竖立扶手拉杆、站立区域、刷卡机、后门下车铃等公交车元素 - **氛围**干净、舒适、明亮传递安心出行的品牌质感 ## 2. 品牌优势传达真实可信 体现嘟嘟巴士的核心优势 - 线路覆盖热门目的地 - 价格实惠透明 - 座位舒适、空间宽敞 - 准时可靠、体验稳定 ⚠️ **禁止虚假宣传**不夸大、不承诺不存在的高端配置保持“普通但干净舒适”的真实定位 ## 3. 剧情结构15秒起承转合 - **0-3秒起**钩子——制造悬念/情绪共鸣/视觉冲击 - **3-8秒承**展开——引入热点/场景/人物 - **8-12秒转**推进——展示巴士/播报线路/揭示优势 - **12-15秒合**收尾——情绪升华/互动引导/品牌记忆点 ## 4. 视听语言描述画面感强化 请用文字生动描绘以下要素 - **声音**车门开合声、发动机启动声、车内播报声、环境音海浪/鸟鸣/城市声、背景音乐风格 - **光线**清晨柔光、午后暖阳、黄昏金色光线、车内自然光透过窗帘 - **运镜**推聚焦细节、拉展示全景、摇跟随动线、移车内穿行、跟人物移动 - **构图**特写表情/车票/手机屏幕、中景人物互动、全景大巴与环境、俯拍车内座位布局 - **色调**温暖橙黄治愈感、清新蓝白干净感、高饱和度活力感、低饱和度高级感 ## 5. 热点结合策略 从热榜数据中筛选最适合融入剧本的热点 - 优先选择旅行相关、周末出行、亲子话题、情绪共鸣类、生活方式类 - 融合方式台词呼应、场景设定、情绪基调、话题标签 - 自然不硬凑热点是情绪的引子不是剧本的全部 ## 6. 价格信息植入规则 从 dudu_prices.json 中 - **播报1条线路**在车内以语音播报形式呈现如“下一站XX票价仅需XX元” - **展示3条线路**通过以下方式之一在画面中呈现 - 手机屏幕乘客正在查票 - 车内电子屏滚动显示线路信息 - 字幕形式画面下方浮现 - 车票/宣传单特写 ## 7. 输出格式纯文本无格式标记 **请直接输出纯文字剧本不要使用以下内容** - ❌ “画面开始”、“第X秒”、“字幕”、“内景”等前置标题 - ❌ 表格、Markdown、代码块 - ❌ 任何格式标记加粗、斜体、列表符号 # 输出要求 请严格按照以上规则生成5个完整的15秒剧本。剧本需自然融入热点与价格数据画面感强品牌传达清晰具备短视频传播潜力。 **保存要求** - 将生成的5个剧本分别保存为Markdown文件 - 保存路径/Users/dudumac003/.openclaw/workspace-plot/script/ - 文件命名格式YYYY-MM-DD-序号.md例如2026-03-30-1.md、2026-03-30-2.md **每个文件内容需包含** - ##视频描述用于发布在社媒平台的简介50-100字简洁吸引人可包含emoji - ##剧本内容15秒完整剧本纯文字描述 - ##发布话题推荐的话题标签8个左右生成剧本openclaw agent --agent plot --message /Users/dudumac003/.openclaw/workspace-plot/prompt.md️视频生成/Users/dudumac003/.openclaw/workspace-plot/wanxiang.pyimport os import re import requests from http import HTTPStatus from dashscope import VideoSynthesis import dashscope from datetime import datetime import glob # 配置基础参数 dashscope.base_http_api_url https://dashscope.aliyuncs.com/api/v1 api_key ……………………………… # 剧本文件目录 SCRIPT_DIR /Users/dudumac003/.openclaw/workspace-plot/script # -------------------------- 根据时间动态生成文件名 -------------------------- def generate_filename(script_index): 根据当前时间和剧本序号生成文件名 now datetime.now() timestamp now.strftime(%Y%m%d_%H%M%S) return f嘟嘟巴士_{timestamp}_剧本{script_index}.mp4 # -------------------------- 读取剧本内容 -------------------------- def extract_script_content(md_file_path): 从Markdown文件中提取## 剧本内容部分的内容 :param md_file_path: Markdown文件路径 :return: 剧本内容字符串 try: with open(md_file_path, r, encodingutf-8) as f: content f.read() # 使用正则匹配## 剧本内容之后的内容直到下一个二级标题或文件结束 pattern r##\s*剧本内容\s*\n(.*?)(?\n##\s|\Z) match re.search(pattern, content, re.DOTALL) if match: script_content match.group(1).strip() print(f✅ 成功读取剧本: {os.path.basename(md_file_path)}) return script_content else: print(f⚠️ 未找到## 剧本内容部分: {md_file_path}) return None except Exception as e: print(f❌ 读取文件失败 {md_file_path}: {str(e)}) return None # -------------------------- 获取所有剧本文件 -------------------------- def get_script_files(): 获取指定目录下所有符合命名规则的剧本文件 格式: YYYY-MM-DD-序号.md # 获取今天的日期 today datetime.now().strftime(%Y-%m-%d) pattern os.path.join(SCRIPT_DIR, f{today}-*.md) script_files sorted(glob.glob(pattern)) if not script_files: print(f⚠️ 未找到今天的剧本文件: {pattern}) return [] print(f 找到 {len(script_files)} 个剧本文件:) for f in script_files: print(f - {os.path.basename(f)}) return script_files # -------------------------- 下载视频 -------------------------- def download_video(video_url, save_path): 下载远程视频到本地 try: print(f开始下载视频保存路径{os.path.abspath(save_path)}) response requests.get(video_url, streamTrue, timeout300) response.raise_for_status() with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) print(f✅ 视频下载成功本地文件路径{os.path.abspath(save_path)}) return True except Exception as e: print(f❌ 视频下载失败{str(e)}) return False # -------------------------- 生成视频任务 -------------------------- def generate_video_from_script(script_content, script_index, output_dir): 根据剧本内容生成视频 :param script_content: 剧本内容文本 :param script_index: 剧本序号 :param output_dir: 输出目录 :return: 是否成功 print(f\n{*60}) print(f 开始生成第 {script_index} 个视频) print(f{*60}) # 生成输出文件名 output_filename generate_filename(script_index) output_path os.path.join(output_dir, output_filename) # 创建异步任务 try: rsp VideoSynthesis.async_call( api_keyapi_key, modelwan2.6-t2v, promptscript_content, # 使用剧本内容作为prompt size720*1280, # 手机竖屏分辨率 shot_typemulti, # 开启多镜头叙事 duration15, # 视频时长15秒 prompt_extendTrue, watermarkTrue, negative_prompt模糊,失真,不良画面,虚假宣传,公交车,竖立扶手,刷卡机,站立区, seed12345 script_index # 不同剧本使用不同种子 ) if rsp.status_code ! HTTPStatus.OK: print(f❌ 任务创建失败状态码: {rsp.status_code}, 错误信息: {rsp.message}) return False print(f✅ 任务创建成功task_id: {rsp.output.task_id}) # 等待任务完成 print(⏳ 正在生成视频请耐心等待1-5分钟...) rsp_result VideoSynthesis.wait(taskrsp, api_keyapi_key) if rsp_result.status_code HTTPStatus.OK: video_url rsp_result.output.video_url print(f✅ 视频生成成功远程链接{video_url}) # 下载视频 if download_video(video_url, output_path): print(f✅ 第 {script_index} 个视频处理完成) return True else: return False else: print(f❌ 视频生成失败状态码: {rsp_result.status_code}) print(f错误信息: {rsp_result.message}) if link hit security strategy in str(rsp_result.message): print(\n 报错解决方案) print(1. 检查剧本内容是否包含敏感词) print(2. 精简prompt内容) print(3. 核对地域一致性) return False except Exception as e: print(f❌ 生成视频时发生异常: {str(e)}) return False # -------------------------- 主函数 -------------------------- def main(): 主函数读取所有剧本并依次生成视频 print( 嘟嘟巴士视频自动生成器启动) print(f 剧本目录: {SCRIPT_DIR}) # 创建输出目录如果不存在 output_dir /Users/dudumac003/.openclaw/workspace-plot/video os.makedirs(output_dir, exist_okTrue) # 获取所有剧本文件 script_files get_script_files() if not script_files: print(❌ 没有找到今天的剧本文件请先运行剧本生成脚本) return # 统计成功/失败数量 success_count 0 fail_count 0 # 遍历每个剧本文件 for idx, script_file in enumerate(script_files, 1): # 提取剧本内容 script_content extract_script_content(script_file) if not script_content: print(f⚠️ 跳过第 {idx} 个剧本内容为空) fail_count 1 continue # 生成视频 success generate_video_from_script(script_content, idx, output_dir) if success: success_count 1 else: fail_count 1 # 每个视频之间稍作延迟避免API请求过于频繁 if idx len(script_files): print(\n⏸️ 等待5秒后处理下一个剧本...) import time time.sleep(5) # 输出总结 print(f\n{*60}) print(f 任务完成统计) print(f{*60}) print(f✅ 成功: {success_count} 个) print(f❌ 失败: {fail_count} 个) print(f 视频保存位置: {output_dir}) if __name__ __main__: # 执行主函数 main()生成视频python3 /Users/dudumac003/.openclaw/workspace-plot/wanxiang.py⏰Cron注意:应该配置OpenClaw内置cron而不是系统cron。添加一个cron每天00:00触发主协调器openclaw cron add \ --name daily-video-pipeline \ --cron 0 0 * * * \ --session isolated \ --agent coordinator \ --message cd ~/.openclaw/workspace-coordinator ./coordinate.sh \ --announce