企业微信自建应用通知中心搭建笔记

一、 核心逻辑与通路选择

在搭建过程中,曾尝试通过“微信客服(KF)”通道获取客户加密 ID(external_userid),但由于未认证企业受到腾讯接口的严格脱敏与权限拦截(例如 4009648002 错误),最终选择走企业微信最稳固的基础盘——自建应用消息通道

优势

  • 无权限拦截:规避了微信客服接口对未认证企业的限制。
  • 账号清晰明确:无需使用 wmwo 开头的加密密文 ID,直接使用企业通讯录中的员工账号(userid)作为接收人。
  • 支持多用户:支持单发或通过竖线(|)分隔实现多用户同时推送。

二、 环境配置文件 r.sh

通过环境变量将企业凭证注入容器中。请确保创建或修改 r.sh 脚本,将以下内容写入其中:

podman run -d \
  --name myz \
  -p 8000:8000 \
  -m 50m \
  -e CORP_ID="ww71……96d0" \
  -e KF_SECRET="wvFz……T82aQ" \
  -e TO_USER_ID="TaiHeXin" \
  localhost/zwechat-adapter:v1

参数说明

  • CORP_ID:企业微信的最高唯一标识(企业ID)。
  • KF_SECRET:自建应用(如 ZJiQi)的独立 Secret,用于获取发信 Token。
  • TO_USER_ID:接收通知的员工账号(userid)。
  • 仅发给自己:填写当前用户的账号(例如 "TaiHeXin")。
  • 共同接收:使用竖线分隔(例如 "TaiHeXin|ZhaoPengPeng")。

三、 服务端核心代码 app.py

基于 FastAPI 框架编写的正式发送版代码。服务暴露出一个 /webhook 接口,负责自动向微信服务器申请临时的 access_token,并调用 message/send 接口把消息推送到指定手机。

import os
import requests
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 从环境变量中读取容器启动时传入的配置参数
CORP_ID = os.environ.get("CORP_ID", "")
KF_SECRET = os.environ.get("KF_SECRET", "")
TO_USER_ID = os.environ.get("TO_USER_ID", "")

@app.get("/webhook")
def send_wechat_msg(msg: str):
    if not msg:
        raise HTTPException(status_code=400, detail="msg 参数不能为空")
        
    # 1. 获取通用的全局 Access Token
    token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={KF_SECRET}"
    try:
        token_res = requests.get(token_url).json()
        token = token_res.get("access_token")
        if not token:
            return {"status": "error", "message": "获取Token失败", "weixin_error": token_res}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"请求Token发生异常: {str(e)}")

    # 2. 调用企业微信核心的应用消息发送接口
    send_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    payload = {
        "touser": TO_USER_ID,        # 接收消息的员工账号
        "msgtype": "text",
        "agentid": 1000002,          # 自建应用 ZJiQi 的 AgentId
        "text": {
            "content": msg
        },
        "safe": 0
    }
    
    try:
        send_res = requests.post(send_url, json=payload).json()
        if send_res.get("errcode") == 0:
            return {"status": "success", "message": "消息发送成功"}
        else:
            return {"status": "error", "message": "应用消息接口返回错误", "weixin_error": send_res}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"发送应用消息发生异常: {str(e)}")
注意:若自建应用对应的 AgentId 发生变更,请及时修正代码中第 32 行的数字。

四、 编译与部署流程

完成代码和脚本配置后,在项目根目录下依次执行以下命令进行编译与启动:

  1. 编译镜像

    podman build -t localhost/zwechat-adapter:v1 .
    

2. **启动容器**

sh r.sh


---

## 五、 通车测试与验证

容器成功运行后,可以在服务器终端通过执行 `curl` 命令来验证消息发送通路:

curl -G "http://localhost:8000/webhook" --data-urlencode "msg=绕过客服泥潭,应用通道全线通车!"


### 预期结果

* 终端返回标准成功 JSON:`{"status":"success","message":"消息发送成功"}`。

标签: none

添加新评论