一、前期准备
1、企业微信具备管理企业权限。
2、服务器有公网IP或者可以将本机端口通过net映射到公网。
二、通过脚本向企业微信发送消息
1、创建sh脚本用来发送消息。
vim 2.sh
注意:脚本中xxxx信息需要在企业微信管理后台获取。
#!/bin/bash
# 设置企业微信相关信息
#每个企业都拥有唯一的corpid,获取此信息可在管理后台“我的企业”-“企业信息”下查看“企业ID”(需要有管理员权限)
corpid="xxxx"
#secret是企业应用里面用于保障数据安全的“钥匙”,每一个应用都有一个独立的访问密钥,为了保证数据的安全,secret务必不能泄漏。secret查看方法:在管理后台->“应用管理”->“应用”->“自建”,点进某个应用,即可看到。
corpsecret="xxxx"
#每个应用都有唯一的agentid。在管理后台->“应用管理”->“应用”,点进某个应用,即可看到agentid。
agentid="xxxx"
#每个部门都有唯一的id,在管理后台->“通讯录”->“组织架构”->点击某个部门右边的小圆点可以看到
#toparty="xxxx"
#每个成员都有唯一的userid,即所谓“账号”。在管理后台->“通讯录”->点进某个成员的详情页,可以看到。
userids="xxxx|xxxxx"
#获取access_token URL
url="https://qyapi.weixin.qq.com/cgi-bin/gettoken"
# access_token是企业后台去企业微信的后台获取信息时的重要票据,由corpid和secret产生。所有接口在通信时都需要携带此信息用于验证接口的访问权限,jq 可通过 yum install jq 安装。
access_token=$(curl -s -G "$url" --data-urlencode "corpid=$corpid" --data-urlencode "corpsecret=$corpsecret" | jq -r .access_token)
# 发送的消息
message="{\"touser\" : \"$userids\", \"toparty\" : \"$toparty\", \"msgtype\" : \"text\", \"agentid\" : \"$agentid\", \"text\" : {\"content\" : \"您的ssh被登录,请及时处理!\"}}"
curl -s -X POST "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=$access_token" -d "$message"
2、 执行脚本
vim 2.sh
当发现错误码:60020,通过微信接口文档,发现需要添加可信IP地址。全局错误码 - 接口文档 - 企业微信开发者中心 (qq.com)
三、用python代码通过URL认证后配置可信IP配置
1、引入WXBizMsgCrypt模块
# 下载项目
wget https://github.com/sbzhu/weworkapi_python/archive/refs/heads/master.zip
# 解压
unzip master.zip
# 改名
mv weworkapi_python-master weworkapi_python
2、随便起个名vim到一个文件即可 我这里叫weixinapi.py
#-*- encoding:utf-8 -*-
from flask import abort, request
from flask import Flask
from xml.dom.minidom import parseString
import _thread
import time
import os
import sys
sys.path.append("weworkapi_python/callback") # 正确的模块导入路径
from WXBizMsgCrypt3 import WXBizMsgCrypt # https://github.com/sbzhu/weworkapi_python 项目地址
app = Flask(__name__)
# 对应步骤4中接受消息回调模式中的URL,如域名是'www.example.com' 那么在步骤4中填入的url就为"http://www.example.com/hook_path"
@app.route('/hook_path', methods=['GET','POST'])
def douban():
if request.method == 'GET':
echo_str = signature(request, 0)
return(echo_str)
elif request.method == 'POST':
echo_str = signature2(request, 0)
return(echo_str)
qy_api = [
WXBizMsgCrypt("xxxx", "xxxxx", "xxxxx"),
] #对应接受消息回调模式中的token,EncodingAESKey 和 企业信息中的企业id
# 开启消息接受模式时验证接口连通性
def signature(request, i):
msg_signature = request.args.get('msg_signature', '')
timestamp = request.args.get('timestamp', '')
nonce = request.args.get('nonce', '')
echo_str = request.args.get('echostr', '')
ret,sEchoStr=qy_api[i].VerifyURL(msg_signature, timestamp,nonce,echo_str)
if (ret != 0):
print("ERR: VerifyURL ret: " + str(ret))
return("failed")
else:
return(sEchoStr)
# 实际接受消息
def signature2(request, i):
msg_signature = request.args.get('msg_signature', '')
timestamp = request.args.get('timestamp', '')
nonce = request.args.get('nonce', '')
data = request.data.decode('utf-8')
ret,sMsg=qy_api[i].DecryptMsg(data,msg_signature, timestamp,nonce)
if (ret != 0):
print("ERR: DecryptMsg ret: " + str(ret))
return("failed")
else:
with open ("/var/log/qywx.log", 'a+') as f: # 消息接收日志
doc = parseString(sMsg)
collection = doc.documentElement
name_xml = collection.getElementsByTagName("FromUserName")
msg_xml = collection.getElementsByTagName("Content")
type_xml = collection.getElementsByTagName("MsgType")
pic_xml = collection.getElementsByTagName("PicUrl")
msg = ""
name = ""
msg_type = type_xml[0].childNodes[0].data
if msg_type == "text": #文本消息
name = name_xml[0].childNodes[0].data #发送者id
msg = msg_xml[0].childNodes[0].data #发送的消息内容
f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:%s\n" % (i, name, msg))
_thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, msg, i, 0), )) #此处将消息进行外部业务处理
elif msg_type == "image": #图片消息
name = name_xml[0].childNodes[0].data
pic_url = pic_xml[0].childNodes[0].data
f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:图片消息\n" % (i, name))
_thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, pic_url, i, 1), )) #此处将消息进行外部业务处理
f.close()
return("ok")
if __name__=='__main__':
app.run("0.0.0.0", 888) #本地监听端口,可自定义
3、运行python3 2.py 进行验证
缺少什么模块安装什么模块
pip3 install 模块名
注意!!!Crypto 模块无法安装,直接安装pycryptodome
yum install pycryptodome
4、至此url认证通过,现在可以将公网可信IP加入可信列表了。
5、再次 执行 weixinapi.py,发现发送消息到企业微信成功。
python3 weixinapi.py
参考地址:https://developer.work.weixin.qq.com/document/path/90664 https://blog.csdn.net/xinhainan/article/details/134467271