""" 企业微信回调服务端 功能:接收企业微信的回调验证和消息推送 作者:开发工程师 日期:2025年01月 """ import os import logging import json from urllib.parse import unquote from flask import Flask, request, abort from WXBizMsgCrypt import WXBizMsgCrypt from config import Config # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # 创建Flask应用 app = Flask(__name__) def get_wecom_config(): """获取企业微信配置,优先从环境变量获取""" corp_id = Config.CORP_ID token = Config.TOKEN encoding_aes_key = Config.ENCODING_AES_KEY if not token: logger.error("缺少TOKEN配置,请设置环境变量TOKEN或修改代码中的TOKEN变量") return None, None, None if not encoding_aes_key: logger.error("缺少ENCODING_AES_KEY配置,请设置环境变量WECOM_ENCODING_AES_KEY或修改代码中的ENCODING_AES_KEY变量") return None, None, None return corp_id, token, encoding_aes_key @app.before_request def log_request_info(): """记录请求信息""" # 记录请求的基本信息 logger.info(f"=== 收到HTTP请求 ===") logger.info(f"请求方法: {request.method}") logger.info(f"请求URL: {request.url}") logger.info(f"请求路径: {request.path}") logger.info(f"客户端IP: {request.remote_addr}") # 记录请求头 headers = dict(request.headers) logger.info(f"请求头: {json.dumps(headers, indent=2, ensure_ascii=False)}") # 记录查询参数 if request.args: args = dict(request.args) logger.info(f"查询参数: {json.dumps(args, indent=2, ensure_ascii=False)}") # 记录请求体(仅对POST/PUT等方法) if request.method in ['POST', 'PUT', 'PATCH']: try: if request.content_type and 'application/json' in request.content_type: # JSON数据 json_data = request.get_json() if json_data: logger.info(f"请求JSON: {json.dumps(json_data, indent=2, ensure_ascii=False)}") else: # 原始数据 raw_data = request.get_data(as_text=True) if raw_data: logger.info(f"请求体: {raw_data[:500]}...") # 限制长度避免日志过长 except Exception as e: logger.warning(f"读取请求体时出错: {e}") @app.after_request def log_response_info(response): """记录响应信息""" logger.info(f"=== HTTP响应 ===") logger.info(f"响应状态码: {response.status_code}") logger.info(f"响应状态: {response.status}") # 记录响应头 headers = dict(response.headers) logger.info(f"响应头: {json.dumps(headers, indent=2, ensure_ascii=False)}") # 记录响应内容 try: if response.content_type and 'application/json' in response.content_type: # JSON响应 logger.info(f"响应JSON: {response.get_data(as_text=True)}") else: # 其他类型响应 response_data = response.get_data(as_text=True) if response_data: if len(response_data) > 500: logger.info(f"响应内容: {response_data[:500]}...") else: logger.info(f"响应内容: {response_data}") else: logger.info("响应内容: 无内容") except Exception as e: logger.warning(f"读取响应内容时出错: {e}") logger.info(f"=== 请求处理完成 ===\n") return response def get_wxcrypt(): """ 获取企业微信加解密工具对象 """ corp_id, token, encoding_aes_key = get_wecom_config() if not all([corp_id, token, encoding_aes_key]): logger.error("企业微信配置不完整") abort(500) return WXBizMsgCrypt(token, encoding_aes_key, corp_id) @app.route('/callback/command', methods=['GET']) def wecom_callback_get(): """ 企业微信回调接口 GET请求:用于验证回调URL的有效性 """ # 获取加解密工具对象 wxcrypt = get_wxcrypt() # 获取URL参数 msg_signature = request.args.get('msg_signature', '') timestamp = request.args.get('timestamp', '') nonce = request.args.get('nonce', '') logger.info(f"收到回调请求 - Method: {request.method}, msg_signature: {msg_signature}, timestamp: {timestamp}, nonce: {nonce}") # URL验证 echostr = request.args.get('echostr', '') if not echostr: logger.error("GET请求缺少echostr参数") abort(400) if not all([msg_signature, timestamp, nonce]): logger.error("GET请求缺少必要参数") abort(400) try: # URL解码echostr参数 echostr = unquote(echostr) logger.info(f"开始验证URL - echostr: {echostr[:50]}...") logger.info(f"echostr: {echostr}") # 验证URL并解密echostr result = wxcrypt.VerifyURL(msg_signature, timestamp, nonce, echostr) logger.info(f"验证URL结果: {result}") # 检查返回值类型 if isinstance(result, tuple): ret, reply_echostr = result else: ret = result reply_echostr = None logger.info(f"验证结果 - 返回码: {ret}") if ret != 0: logger.error(f"URL验证失败,错误码: {ret}") if ret == -40001: logger.error("签名验证失败 - 请检查Token配置是否与企业微信后台一致") elif ret == -40002: logger.error("AES解密失败或CorpID不匹配 - 请检查EncodingAESKey和CorpID配置") else: logger.error(f"未知错误码: {ret}") abort(403) logger.info("URL验证成功") return reply_echostr except Exception as e: logger.error(f"URL验证过程发生错误: {str(e)}") import traceback logger.error(f"错误详情: {traceback.format_exc()}") abort(500) @app.route('/callback/command', methods=['POST']) def wecom_callback_post(): """ 企业微信回调接口 POST请求:接收企业微信推送的消息和事件 """ # 获取加解密工具对象 wxcrypt = get_wxcrypt() # 获取URL参数 msg_signature = request.args.get('msg_signature', '') timestamp = request.args.get('timestamp', '') nonce = request.args.get('nonce', '') logger.info(f"收到回调请求 - Method: {request.method}, msg_signature: {msg_signature}, timestamp: {timestamp}, nonce: {nonce}") # 接收消息 if not all([msg_signature, timestamp, nonce]): logger.error("POST请求缺少必要参数") abort(400) try: # 获取POST数据 post_data = request.get_data(as_text=True) if not post_data: logger.error("POST请求体为空") abort(400) logger.info(f"收到POST消息: {post_data[:200]}...") # 解密消息 ret, msg = wxcrypt.DecryptMsg(post_data, msg_signature, timestamp, nonce) if ret != 0: logger.error(f"消息解密失败,错误码: {ret}") abort(403) logger.info(f"消息解密成功: {msg}") # 处理消息(这里可以根据业务需求进行扩展) response_msg = handle_message(msg) if response_msg: # 加密响应消息 ret, encrypted_msg = wxcrypt.EncryptMsg(response_msg, nonce, timestamp) if ret == 0: logger.info("响应消息加密成功") return encrypted_msg else: logger.error(f"响应消息加密失败,错误码: {ret}") # 返回空字符串表示成功接收但不回复 return '' except Exception as e: logger.error(f"处理POST请求时发生错误: {str(e)}") abort(500) def handle_message(msg): """ 处理解密后的消息 Args: msg: 解密后的XML消息 Returns: str: 要回复的消息(XML格式),如果不需要回复则返回None """ try: import xml.etree.ElementTree as ET import time # 解析XML消息 root = ET.fromstring(msg) msg_type = root.find('MsgType').text if root.find('MsgType') is not None else '' msg_id = root.find('MsgId').text if root.find('MsgId') is not None else '' logger.info(f"处理消息 - 类型: {msg_type}, 消息ID: {msg_id}") # 根据消息类型进行处理 if msg_type == 'event': # 处理事件消息 event = root.find('Event').text if root.find('Event') is not None else '' logger.info(f"收到事件: {event}") return process_event_message(event, root) # 对于其他类型的消息,记录日志但不回复 logger.info(f"收到其他类型消息,暂不处理: {msg_type}") return None except Exception as e: logger.error(f"处理消息时发生错误: {str(e)}") import traceback logger.error(f"错误详情: {traceback.format_exc()}") return None def process_event_message(event, root): """ 处理事件消息 Args: event: 事件类型 from_user: 发送者 to_user: 接收者 root: XML根节点 Returns: str: 回复消息,如果不需要回复则返回None """ if event == 'kf_msg_or_event': # 微信客服消息或事件 logger.info("收到微信客服事件,开始处理客服消息") # 获取Token和OpenKfId token = root.find('Token').text if root.find('Token') is not None else '' open_kfid = root.find('OpenKfId').text if root.find('OpenKfId') is not None else '' if token and open_kfid: # 异步处理客服消息(避免阻塞回调响应) import threading def process_kf_messages(): try: # 延迟导入避免循环导入 from kf_message_handler import KfMessageHandler kf_handler = KfMessageHandler() kf_handler.process_kf_event(token, open_kfid) except Exception as e: logger.error(f"处理客服消息异常: {e}") # 启动后台线程处理 thread = threading.Thread(target=process_kf_messages) thread.daemon = True thread.start() logger.info(f"已启动客服消息处理线程 - OpenKfId: {open_kfid}") else: logger.error("客服事件缺少必要参数 - Token或OpenKfId为空") # 客服事件不需要回复 return None # 其他事件类型 logger.info(f"收到其他事件类型: {event}") return None @app.route('/health', methods=['GET']) def health_check(): """健康检查接口""" return {'status': 'ok', 'message': '企业微信回调服务运行正常'} @app.route('/', methods=['GET']) def index(): return '''

企业微信回调服务

服务正在运行...

''' if __name__ == '__main__': # 检查配置 corp_id, token, encoding_aes_key = get_wecom_config() if not all([corp_id, token, encoding_aes_key]): logger.error("配置不完整,请检查企业微信相关配置") logger.info("请设置以下环境变量或修改代码中的配置:") logger.info("- WECOM_CORP_ID: 企业ID") logger.info("- WECOM_TOKEN: 应用Token") logger.info("- WECOM_ENCODING_AES_KEY: 应用EncodingAESKey") exit(1) logger.info("企业微信回调服务启动中...") logger.info(f"企业ID: {corp_id}") logger.info(f"Token: {token[:10]}..." if token else "Token: 未配置") logger.info(f"EncodingAESKey: {encoding_aes_key[:10]}..." if encoding_aes_key else "EncodingAESKey: 未配置") # 启动服务 port = Config.PORT logger.info(f"启动服务 - 端口: {port}") host = Config.HOST logger.info(f"启动服务 - 主机: {host}") debug = Config.DEBUG logger.info(f"启动服务 - 调试模式: {debug}") app.run(host=host, port=port, debug=debug, threaded=True, processes=1)