## 从代码到通知:当AI视频流遇上钉钉机器人
项目启动那周,项目经理在晨会上说,客户要求每天凌晨自动生成一段库存预警视频,然后推送到钉钉群里。当时第一反应是,这活儿要么找人半夜盯数据,要么写个脚本把截图拼成视频。后来发现,用现成的AI工具生成视频内容,再通过钉钉机器人推送,比想象中简单得多,也灵活得多。
它到底是什么
说白了,这是一套把“AI生成视频”和“钉钉消息推送”粘合起来的自动化流程。核心是两个独立模块:一边是调用大模型或AI视频生成接口(比如腾讯混元、百度文心一言的API,或者本地部署的Stable Video Diffusion),根据结构化数据生成带有解说或动画的视频片段;另一边是用钉钉开放平台的机器人Webhook,把生成好的视频文件作为消息卡片发到指定群聊。
关键点在于“串联”。不是每次生成视频都手动上传,而是让Python脚本充当调度中心:从数据库或API拉取数据,塞进视频生成模板,得到MP4文件,再调用钉钉接口上传并发送。整个链条里,Python的requests库和subprocess模块是主力,偶尔会用到FFmpeg做格式转换或拼接。
真正能解决什么问题
最早接触这个需求时,有人提议用录屏软件定时录播。但录屏无法动态适应数据变化——比如库存低于警戒线时,视频里需要自动生成红色警告字幕。AI视频生成的优势在于内容可变:同一个模板,填入不同的数字、图表、文案,每次产出的视频都不一样。
实际场景里,最常见的是三类:
- 数据日报视频化:把Excel表格或数据库查询结果,变成带语音解说的短视频,晨会前自动发到群里。
- 监控预警:当系统检测到异常流量或设备离线,自动生成一段15秒的故障描述视频,附带截图和文字说明。
- 批量培训材料:把操作文档转化为AI配音的演示视频,按部门分组推送到不同钉钉群。
有个客户把这种流程用在工厂质检环节:AI根据当日良品率数据生成柱状图动画,配上语音播报,生产主管每天通过视频而不是表格来掌握质量趋势。
从零搭一套的常见路径
第一步是明确数据源。假设从MySQL读库存数据:
importmysql.connector conn=mysql.connector.connect(host='localhost',user='root',password='pwd',database='warehouse')cursor=conn.cursor()cursor.execute("SELECT product_name, stock_quantity FROM products WHERE stock_quantity < 50 ORDER BY stock_quantity ASC")low_stock_items=cursor.fetchall()第二步是用AI生成视频。如果选腾讯混元,需要先申请API,然后组装参数。一个取巧的做法是先生成图片序列,再用FFmpeg合成视频:
importrequestsimportjsonfrompathlibimportPathdefgenerate_alert_image(product,quantity):# 调用文生图API,生成带文字的预警图片url="https://api.hunyuan.cloud.tencent.com/v1/images/generations"headers={"Authorization":"Bearer YOUR_API_KEY"}prompt=f"一张深色背景的工业风格卡片,中间显示产品{product}库存仅剩{quantity}件,红色警示边框"payload={"prompt":prompt,"size":"1280x720"}resp=requests.post(url,json=payload,headers=headers)# 下载返回的图片并保存到本地image_url=resp.json()["data"][0]["url"]img_data=requests.get(image_url).content save_path=Path(f"./frames/alert_{product}.png")save_path.write_bytes(img_data)returnstr(save_path)生成多张图片后,用FFmpeg拼成带过渡效果的视频:
importsubprocess frames_dir="./frames"output_video="./output/daily_alert.mp4"cmd=["ffmpeg","-y","-framerate","1",# 每张图片展示1秒"-pattern_type","glob","-i",f"{frames_dir}/*.png","-c:v","libx264","-pix_fmt","yuv420p",output_video]subprocess.run(cmd,check=True)第三步是发送到钉钉。先在企业钉钉后台创建一个机器人,拿到Webhook地址。注意要设置自定义关键词或者加签,不然会被拦截。上传文件需要先获取上传凭证,然后拼接消息卡片:
defsend_dingtalk_video(webhook,video_path):# 第一步:获取access_token(这里假设已拿到,实际要调用获取token接口)upload_url=f"https://oapi.dingtalk.com/media/upload?access_token={token}&type=file"withopen(video_path,'rb')asf:files={'media':f}upload_resp=requests.post(upload_url,files=files)media_id=upload_resp.json()["media_id"]# 第二步:发送消息卡片card_data={"msgtype":"action_card","action_card":{"title":"📊 今日库存预警视频","markdown":"请查看附件,红底标注为紧急补货项","btn_orientation":"1","btn_json_list":[{"title":"查看视频","action_url":"dingtalk://dingtalkclient/page/link?url=直接播放链接"}]}}# 或者直接发文件消息file_msg={"msgtype":"file","file":{"media_id":media_id}}requests.post(webhook,json=file_msg)最后,用cron或APScheduler调度整个流程,设定每天5:30执行。
踩过的一些坑
最初尝试直接用现成的“AI视频生成+钉钉发送”工具,比如某些低代码平台的预制组件。但很快发现,这类工具生成视频的模板太死板,只能替换文本,无法控制图表位置或动画时长。后来换成自研方案,虽然代码量大了些,但定制空间大得多。
另一个坑是视频文件大小。钉钉机器人对文件大小有限制(免费版最大20MB),AI生成视频如果分辨率太高或时长过长,很容易超限。解决方案是在FFmpeg阶段控制码率:
cmd=["ffmpeg","-i",output_video,"-b:v","800k",# 降低码率"-maxrate","1000k","-bufsize","1500k","output_compressed.mp4"]此外,视频配音如果不做语音识别同步,容易造成音画不同步。建议先用AI语音合成生成音频文件,再通过FFmpeg精确对齐时间戳。
与同类技术的差异
市面上有一些专门的“智能视频生成平台”,比如剪映的商业版、腾讯智影、Mootion等,它们支持直接导出并分享到钉钉。这种方案的好处是完全不需要写代码,拖拽就能用。但局限性在于:无法嵌入到已有的业务系统里,数据更新全靠手动导入。如果每天的数据从内部ERP自动同步,这类工具就显得笨重。
另一种方案是纯用Python库,比如moviepy生成视频,再用钉钉sdk推送。moviepy的优势是原生Python,不需要调API,适合生成简单的文字动画。缺点是对复杂场景(比如接入外部AI模型做画面生成)支持较弱,性能也差一些,渲染较长视频时容易内存溢出。
对比下来,综合方案灵活度最高:AI负责内容生成,FFmpeg负责后期处理,钉钉API负责分发。每一层都可以独立替换——今天用混元,明天换成sd-webui,不影响其他模块。而且所有代码都控制在千行以内,维护成本并不高。
如果团队里有人对视频质量要求特别高,也可以改成先通过AI生成脚本描述,再调用专业渲染引擎(比如Blender的Python接口)输出视频。但这就属于进阶玩法了,日常预警场景完全不用这么复杂。