Python构建企业微信自动消息转发服务端

一、背景

目前有在项目分组,就小组成员中,微信群消息回复较多的情况下,想根据组来转发特定消息,包含文字、图片、语言等。在此只是自己实现仅供参考,可以根据自身需求修改更多功能。

二、代码

2.1 企业微信相关信息

  • 企业ID:corpid
  • 自建应用appid
  • 自建应用secret

2.2 服务端部署 运行环境:

python 版本 2.7

git clone https://github.com/redhatxl/wechatmsg.git
nohup python2.7 wechatmsg/wx_msg_server.py &

2.3 参考RUL:

获取access_token

发送消息

2.4 代码

# flask 框架后台
  def server_run(self):
        app = Flask(__name__)
        @app.route('/index', methods=['GET', 'POST'])
        def index():

            wxcpt = WXBizMsgCrypt(self.sToken, self.sEncodingAESKey, self.sCorpID)
            # 获取url验证时微信发送的相关参数
            sVerifyMsgSig = request.args.get('msg_signature')
            sVerifyTimeStamp = request.args.get('timestamp')
            sVerifyNonce = request.args.get('nonce')
            sVerifyEchoStr = request.args.get('echostr')

            # 验证url
            if request.method == 'GET':
                ret, sEchoStr = wxcpt.VerifyURL(sVerifyMsgSig, sVerifyTimeStamp, sVerifyNonce, sVerifyEchoStr)
                print type(ret)
                print type(sEchoStr)

                if (ret != 0):
                    print "ERR: VerifyURL ret:" + str(ret)
                    sys.exit(1)
                return sEchoStr

            # 接收客户端消息
            if request.method == 'POST':
                sReqMsgSig = sVerifyMsgSig
                sReqTimeStamp = sVerifyTimeStamp
                sReqNonce = sVerifyNonce
                sReqData = request.data
                print(sReqData)

                ret, sMsg = wxcpt.DecryptMsg(sReqData, sReqMsgSig, sReqTimeStamp, sReqNonce)
                print ret, sMsg
                if (ret != 0):
                    print "ERR: DecryptMsg ret: " + str(ret)
                    sys.exit(1)
                # 解析发送的内容并打印

                xml_tree = ET.fromstring(sMsg)
                print('xml_tree is ', xml_tree)
  • 消息内容发送
 def _send_text_msg(self, content):
        data = {
            "touser": ('|').join(self.userid.split(',')),
            "toparty": ('|').join(self.partid.split(',')),
            # "toparty":int(self.partid),
            "msgtype": "text",
            "agentid": self.agent_id,
            "text": {
                "content": content
            },
            "safe": 0
        }
        try:
            response = requests.post(self.send_msg_url.format(self.access_token), json.dumps(data))
            self.logoper.info(response.text)
            print(response.text)
            result_msg = json.loads(response.content)['errmsg']
            return result_msg
        except Exception as e:
            self.logoper.info(e)
  • 日志
   def create_dir(self):
        """
        创建目录
        :return: 文件名称
        """
        _LOGDIR = os.path.join(os.path.dirname(__file__), self.logdir_name)
        _TIME = time.strftime('%Y-%m-%d', time.gmtime()) + '-'
        _LOGNAME = _TIME + self.logfile_name
        LOGFILENAME = os.path.join(_LOGDIR, _LOGNAME)
        if not os.path.exists(_LOGDIR):
            os.mkdir(_LOGDIR)
        return LOGFILENAME

    def create_logger(self, logfilename):
        """
        创建logger对象
        :param logfilename:
        :return: logger对象
        """
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)
        handler = logging.FileHandler(logfilename)
        handler.setLevel(logging.INFO)
        formater = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formater)
        logger.addHandler(handler)
        return logger

配置文件

# 定义微信公众号信息
[common]
# 企业微信企业ID
corpid = wxe23xxxxxxxxxxx


# 接收消息服务器配置
[recmsg]

Token = mVNAAw3xxxxxxxxxxxxxxxxx
EncodingAESKey = vwbKImxc3WPeE073xxxxxxxxxxxxxxxxxx


# 自建应用信息
[appconfig]
# 自建应用agentid
agentid = 1000002
# 自建应用secret
secret = 6HAGX7Muw36pv5anxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

# 消息接收信息
# 消息接收用户id,如果多个用户用英文','隔开
userid = xuel|yaoy

# 消息接收部门id,如果多个用英文','隔开
partid = 11


[urlconfig]
# 获取应用token的api接口
get_access_token_url = https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}
# 发送消息api接口
send_msg_url = https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}
# 上传媒体api接口,获取mediaid
upload_media_url = https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={}&type=image
# 上传高清语音接口
upload_video_url = https://qyapi.weixin.qq.com/cgi-bin/media/get/jssdk?access_token={}&media_id={}

[loginfo]
#日志目录
logdir_name = logdir
#日志文件名称
logfile_name = wechat_server.log

三、测试

在企业微信发送消息,可以修改配置文件制定转发到特定的群组,从而避免消息分流。

启用应用API,设置回调地址

测试发送消息

查看接受消息

四、优化

  • 后期可以配合数据库将每次获取的access_token 保存至数据库,待2小时过期后,再重新获取新的。
  • 更多内容转发

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏新智元

【TensorFlow重大升级】自动将Python代码转为TF Graph,大幅简化动态图处理!

作者:Alex Wiltschko, Dan Moldovan, Wolff Dobson

15720
来自专栏散尽浮华

CentOS7下安装Docker-Compose操作记录

Docker-Compose是一个部署多个容器的简单但是非常必要的工具.安装Docker-Compose之前,请先安装 python-pip

12920
来自专栏有趣的django

Memcached的使用 一、安装和启动二、telnet操作memcached三、python操作memcached

         Memcached 是一个高性能的分布式内存对象缓存系统,用于动态Web应用以减轻数据库负载。它通过在内存中缓存数据和对象来减少读取数据库的次...

10830
来自专栏有趣的django

8.Flask-Script

 Flask-script的作用是可以通过命令行的形式操作flask。安装方式:pip install flask-script

12850
来自专栏电光石火

pip 10.0.0 BUG 解决方案Traceback (most recent call last): File "/usr/bin/pip3", line 9, in <module>

Traceback (most recent call last):  File “/usr/bin/pip3”, line 9, in  from p...

1.5K20
来自专栏新智元

【卷积神经网络失陷】几行Python代码搞定,偏要用100个GPU!

【新智元导读】Uber近日一篇论文引起许多讨论:该论文称发现卷积神经网络一个引人注目的“失败”,并提出解决方案CoordConv。论文称CoordConv解决了...

18700
来自专栏抠抠空间

Flask之基本使用与配置

Flask是一个基于Python开发并且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架,对于Werkzeug本质是Socket服务端,其用于...

12020
来自专栏新智元

Guido 转身离去,Python 何去何从?

【新智元导读】Python之父Guido van Rossum因最近的“PEP 572”事件宣布放弃他在Python社区“仁慈大君”的称号,且没有任命继任者,并...

10540
来自专栏pangguoming

Git忽略规则.gitignore梳理

对于经常使用Git的朋友来说,.gitignore配置一定不会陌生。废话不说多了,接下来就来说说这个.gitignore的使用。

13230
来自专栏张俊红

Sklearn参数详解—Adaboost

今天这篇讲讲集成学习,集成学习就是将多个弱学习器集合成一个强学习器,你可以理解成现在有好多道判断题(判断对错即01),如果让学霸去做这些题,可能没啥问题,几乎全...

30030

扫码关注云+社区

领取腾讯云代金券

年度创作总结 领取年终奖励