视频号去水印解析机器人完整搭建教程(包含机器人源码)

🎯核心目标
用户将微信视频号视频转发给机器人 → 机器人自动提取视频号原始链接 → 解析无水印视频地址 → 回复给用户
最终效果:个人微信转发视频号给企业微信机器人,10 秒内收到无水印链接(全程免费搭建,新手也能搞定)
⚡第一阶段:注册企业微信 + 创建机器人应用
步骤 1:注册企业微信
1.打开企业微信官网 → 点击 “立即注册” → 选择 “企业 / 团队” → 填写企业名称(随便填,如 “视频号解析助手”)、行业(随便选)、人员规模(1-20 人) → 用个人微信扫码验证 → 完成注册。
2.登录企业微信管理后台(https://work.weixin.qq.com/wework_admin/loginpage_wx),记录右上角 “企业 ID”(后续开发用,格式:wxXXXXXXX)。
步骤 2:创建机器人应用
- • 1.管理后台左侧菜单 → 「应用管理」→ 「应用」→ 「创建应用」。
- • 2.填写应用信息:
应用名称: 视频号解析助手(用户可见,可自定义)
应用头像: 随便上传一张图片(方便用户识别)
可见范围: 选 “全部成员”(或自己的账号) - • 3.点击 “创建应用” → 进入应用详情页,记录 3 个关键信息(后续代码要用,抄在记事本上):
- • AgentId(应用 ID,如 1000002)
- • Secret(应用密钥,点击 “查看” 需用个人微信扫码验证,如 XXXXXXXXXXXXXXXXXX)
- • 企业 ID(已在步骤 1 记录)
步骤 3:配置应用接收消息权限
- • 1.应用详情页 → 「接收消息」→ 开启「消息接收 API」(蓝色开关打开)。
- • 2 填写配置信息(暂时先填占位符,服务器部署后再修改)
- • 消息接收 URL:暂时填 http://你的服务器公网IP/wechat/callback(后续替换为真实服务器地址)
- • Token:自定义一个字符串(如videoparse2025,记下来,代码里要一致)
- • EncodingAESKey:点击 “随机生成” → 复制生成的字符串记下来,代码里要一致
- • 点击 “保存” → 弹出 “验证 URL” 提示(暂时验证失败没关系,服务器部署后再回来验证)
步骤 4:添加机器人为好友
1.企业微信管理后台 → 「我的企业」→ 「企业信息」→ 记录 “企业微信名称”(如 “视频号解析助手”)。
2.打开个人微信 → 搜索企业微信名称 → 找到你创建的应用(机器人)→ 发送消息 “测试”,确认能正常对话(暂时机器人不会回复,后续开发完就会响应)
🎨 第二阶段:配置云服务器
步骤 1:登录服务器
1.云服务器控制台(阿里云 / 腾讯云)→ 找到你的服务器 → 记录「公网 IP」、「登录账号」(默认 ubuntu 或 root)、「登录密码」(创建服务器时设置的)
2.打开 Xshell → 点击 “新建” → 填写会话信息
名称: 视频号机器人服务器
主机: 填写服务器公网 IP
端口: 22(默认 SSH 端口)
3.点击 “连接” → 输入登录账号(ubuntu)→ 输入密码 → 成功登录服务器(界面显示ubuntu@xxx:~$即为成功)。
步骤 2:安装基础环境
复制以下命令,逐条在 Xshell 中执行(每行输完按回车,等待执行完成再输下一条):
# 1. 更新系统软件(Ubuntu系统)
sudo apt update && sudo apt upgrade -y
# 2. 安装Python3、pip3(开发必备)
sudo apt install python3 python3-pip -y
# 3. 安装Flask(搭建HTTP服务,接收企业微信回调)、requests(调用接口)、加密库
pip3 install flask requests pycryptodome gunicorn
# 4. 安装Nginx(反向代理,后续配置SSL证书)
sudo apt install nginx -y
# 5. 安装Certbot(申请免费SSL证书)
sudo apt install certbot python3-certbot-nginx -y
执行完后,输入python3 --version,显示Python 3.8.x以上即为成功。
步骤 3:开放服务器端口
- • 1.云服务器控制台 → 找到「安全组」→ 「配置规则」→ 「入方向规则」→ 「添加规则」:
- • 端口范围:80(HTTP)、443(HTTPS)、22(SSH,默认已开放)
- • 授权对象:0.0.0.0/0(允许所有 IP 访问,方便测试,后续可限制中国境内)
- • 描述:允许 HTTP/HTTPS 访问
- • 2.点击 “确定” → 确保 3 个端口都已开放(入方向状态为 “允许”)
步骤 4:申请免费 SSL 证书
1.在 Xshell 中执行以下命令(替换为你的服务器公网 IP,或自己的域名,没有域名就用 IP):
# 替换为你的服务器公网IP(如123.45.67.89),没有域名就填IP
sudo certbot --nginx -d 123.45.67.89
2.提示 “Successfully obtained certificate” 即为申请成功,证书会自动配置到 Nginx。
🧠 第三阶段:开发机器人代码
步骤 1:创建代码文件
1.在 Xshell 中执行以下命令,创建代码目录并进入:
mkdir -p /home/ubuntu/video_parse_bot # 创建文件夹
cd /home/ubuntu/video_parse_bot # 进入文件夹
2.创建代码文件(用 nano 编辑器):
nano video_bot.py
步骤 2:完整代码(直接复制,替换 3 处关键信息)
from flask import Flask, request, jsonify
import requests
from Crypto.Cipher import AES
import base64
import json
import time
# -------------------------- 替换为你的企业微信信息(必改!)--------------------------
CORP_ID = "你的企业ID" # 步骤1.2记录的企业ID(wxXXXXXXX)
CORP_SECRET = "你的应用Secret" # 步骤1.2记录的应用Secret
AGENT_ID = "你的AgentId" # 步骤1.2记录的AgentId(数字,如1000002)
TOKEN = "videoparse2025" # 步骤1.3自定义的Token(要和企业微信后台一致)
ENCODING_AES_KEY = "你的EncodingAESKey" # 步骤1.3生成的EncodingAESKey
# -------------------------- 替换为你的解析接口(必改!)--------------------------
PARSE_API = "https://api.qianxin918.com/api/v1/video/parse" # 替换为你的解析接口
# ----------------------------------------------------------------------------------
app = Flask(__name__)
# 1. 缓存access_token(2小时过期,避免重复调用接口)
access_token_cache = {
"token": None,
"expire_time": 0
}
def get_access_token():
"""获取企业微信access_token(接口调用凭证)"""
global access_token_cache
# 检查缓存是否有效
if access_token_cache["token"] and time.time() < access_token_cache["expire_time"]:
return access_token_cache["token"]
# 缓存失效,重新获取
url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={CORP_SECRET}"
try:
res = requests.get(url, timeout=10)
res.raise_for_status()
data = res.json()
if data.get("errcode") == 0:
access_token = data["access_token"]
access_token_cache["token"] = access_token
access_token_cache["expire_time"] = time.time() + 7100 # 缓存7100秒(2小时-100秒)
return access_token
else:
print(f"获取access_token失败:{data}")
return None
except Exception as e:
print(f"获取access_token异常:{str(e)}")
return None
def decrypt_msg(msg_signature, timestamp, nonce, encrypt_msg):
"""解密企业微信加密消息(兼容模式下也需要解密)"""
try:
# 解码EncodingAESKey
aes_key = base64.b64decode(ENCODING_AES_KEY + "=")
# AES-CBC解密
cipher = AES.new(aes_key, AES.MODE_CBC, aes_key[:16]) # IV为密钥前16位
decrypt_data = cipher.decrypt(base64.b64decode(encrypt_msg))
# 提取真实消息(格式:16位随机数 + 4位消息长度 + 消息内容 + 企业ID)
msg_len = int.from_bytes(decrypt_data[16:20], byteorder="big")
real_msg = decrypt_data[20:20+msg_len].decode("utf-8")
return json.loads(real_msg)
except Exception as e:
print(f"解密消息失败:{str(e)}")
return None
def extract_weixin_video_url(video_id):
"""通过视频号临时素材ID,提取原始链接"""
access_token = get_access_token()
if not access_token:
return None
url = f"https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token={access_token}&media_id={video_id}"
try:
res = requests.get(url, timeout=15)
res.raise_for_status()
data = res.json()
# 视频号原始链接在data["url"]中(格式:https://channels.weixin.qq.com/xxx)
if data.get("url"):
print(f"提取到视频号链接:{data['url']}")
return data["url"]
else:
print(f"提取视频号链接失败:{data}")
return None
except Exception as e:
print(f"获取临时素材异常:{str(e)}")
return None
def parse_video_no_watermark(original_url):
"""调用解析接口,获取无水印链接"""
if not original_url:
return "提取视频号链接失败,请重试"
try:
# 对原始链接进行URL编码(避免特殊字符报错)
encoded_url = requests.utils.quote(original_url)
# 调用解析接口(替换为你的接口参数格式,不同接口格式不同!)
parse_url = f"{PARSE_API}?url={encoded_url}"
res = requests.get(parse_url, timeout=20)
res.raise_for_status()
data = res.json()
# 不同接口返回格式不同,需根据接口文档调整(以下是示例格式)
if data.get("code") == 0 and data.get("data", {}).get("no_watermark_url"):
return data["data"]["no_watermark_url"]
elif data.get("success") and data.get("url"):
return data["url"]
else:
print(f"解析无水印失败:{data}")
return f"解析失败,错误信息:{data.get('msg', '未知错误')}"
except Exception as e:
print(f"解析接口异常:{str(e)}")
return "解析接口超时,请稍后再试"
def send_msg_to_user(user_id, content):
"""给用户发送消息(企业微信文本消息接口)"""
access_token = get_access_token()
if not access_token:
return
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
data = {
"touser": user_id, # 接收用户ID(企业微信用户ID)
"agentid": int(AGENT_ID), # 应用AgentId(必须是数字)
"msgtype": "text",
"text": {"content": content},
"safe": 0, # 0=普通消息,1=保密消息
"enable_id_trans": 0,
"enable_duplicate_check": 0
}
try:
res = requests.post(url, json=data, timeout=10)
res.raise_for_status()
result = res.json()
if result.get("errcode") == 0:
print(f"回复用户{user_id}成功:{content}")
else:
print(f"回复用户失败:{result}")
except Exception as e:
print(f"发送消息异常:{str(e)}")
@app.route("/wechat/callback", methods=["POST"])
def wechat_callback():
"""接收企业微信消息回调(用户转发视频号触发)"""
# 获取回调参数
msg_signature = request.args.get("msg_signature")
timestamp = request.args.get("timestamp")
nonce = request.args.get("nonce")
encrypt_msg = request.data.decode("utf-8", errors="ignore")
# 解密消息
real_msg = decrypt_msg(msg_signature, timestamp, nonce, encrypt_msg)
if not real_msg:
return jsonify({"errcode": -1, "errmsg": "decrypt failed"})
print(f"收到用户消息:{real_msg}")
# 识别视频号消息(MsgType=video且包含VideoMsg)
if real_msg.get("MsgType") == "video" and real_msg.get("VideoMsg"):
user_id = real_msg.get("FromUserName") # 发送用户ID
video_id = real_msg["VideoMsg"].get("MsgId") # 视频号临时素材ID
# 提取原始链接
original_url = extract_weixin_video_url(video_id)
# 解析无水印链接
no_watermark_url = parse_video_no_watermark(original_url)
# 回复用户
reply_content = f"✅ 视频号无水印链接已生成:\n{no_watermark_url}\n\n💡 提示:链接有效期24小时,建议及时保存"
send_msg_to_user(user_id, reply_content)
else:
# 非视频号消息,回复提示
user_id = real_msg.get("FromUserName")
send_msg_to_user(user_id, "❌ 请转发「微信视频号视频」给我,我会自动生成无水印链接~")
# 必须返回200 OK,否则企业微信会重试回调
return jsonify({"errcode": 0, "errmsg": "success"})
if __name__ == "__main__":
# 本地测试用(服务器部署用gunicorn启动)
app.run(host="0.0.0.0", port=8080, debug=True)
步骤 3:修改代码中的关键信息
- • CORP_ID:步骤 1.2 记录的企业 ID(如 wx1234567890abcdef)
- • CORP_SECRET:步骤 1.2 记录的应用 Secret(如 abcdef1234567890)
- • AGENT_ID:步骤 1.2 记录的 AgentId(数字,如 1000002)
- • ENCODING_AES_KEY:步骤 1.3 生成的 EncodingAESKey(如 abcdef1234567890...)
- • PARSE_API:替换为你找到的视频号解析接口(如阿里云市场的接口地址)
步骤 4:配置 Nginx 反向代理
1.执行以下命令,编辑 Nginx 配置文件:
sudo nano /etc/nginx/sites-available/default
2.找到server { ... }块,替换为以下内容(替换你的服务器公网IP为实际 IP):
server {
listen 80;
listen [::]:80;
server_name 你的服务器公网IP; # 替换为你的服务器IP或域名
# 强制HTTPS
location / {
return 301 https://$host$request_uri;
}
}
server {
listen 443 ssl;
listen [::]:443 ssl;
server_name 你的服务器公网IP; # 替换为你的服务器IP或域名
# SSL证书配置(Certbot自动生成,无需修改)
ssl_certificate /etc/letsencrypt/live/你的服务器公网IP/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/你的服务器公网IP/privkey.pem;
include /etc/letsencrypt/options-ssl-nginx.conf;
ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem;
# 反向代理到机器人代码(Flask服务)
location /wechat/callback {
proxy_pass http://127.0.0.1:8080; # 转发到本地8080端口
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
步骤 5:启动机器人服务
1.在代码目录下,执行以下命令启动服务(后台运行,关闭 Xshell 也不影响):
cd /home/ubuntu/video_parse_bot
nohup gunicorn -w 4 -b 127.0.0.1:8080 video_bot:app &
2.查看服务是否启动成功:
ps aux | grep gunicorn
3.查看日志(排查错误用):
tail -f nohup.out
🎯 第四阶段:验证企业微信回调 + 测试机器人
步骤 1:更新企业微信回调 URL
1.登录企业微信管理后台 → 「应用管理」→ 找到你的机器人应用 → 「接收消息」。
2.修改「消息接收 URL」为:https://你的服务器公网IP/wechat/callback(注意是 HTTPS 开头)。
3.确认「Token」和「EncodingAESKey」与代码中一致 → 点击「保存」→ 提示 “验证成功” 即为配置完成。
步骤 2:测试机器人功能
1.打开个人微信 → 找到你的机器人(企业微信应用)→ 转发一个视频号视频给机器人。
2.等待 10 秒内,机器人回复无水印链接 → 点击链接,即可下载无水印视频(测试成功!)。
至此!大功告成!你的视频号解析机器人就搭建完成啦!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)