Python 开发企微第三方 API:RPA 模式下外部群主动调用实现

Python 开发企业微信第三方 API:RPA 模式下外部群主动调用实现

重阳,你好!从你的查询来看,你想用 Python 开发企业微信(WeChat Work)的第三方应用 API,在 RPA(Robotic Process Automation,机器人过程自动化)模式下,实现外部群(客户群)的主动调用。这是一个典型的自动化测试/运维/业务集成场景。

根据企业微信官方开发者文档(2026 年最新版),企业微信第三方 API 不支持直接从外部群“主动调用”应用(即群成员或机器人直接触发应用逻辑),因为外部群的交互主要是被动响应或应用侧推送。企业微信的 RPA 模式也不是原生内置的“模式”,而是通过 API + 外部调度工具(如 cron、Airflow 或 Python 脚本)实现的自动化流程。

  • 核心机制:使用 /cgi-bin/message/send API 从应用侧主动推送消息到外部群成员(间接影响群)。如果需要“响应式调用”,可以结合 webhook(群机器人回调)实现近似“主动”效果。
  • RPA 集成:RPA 工具(如 UiPath、Automation Anywhere)可调用企业微信 API,但用 Python 纯脚本实现 RPA 时,通常结合 requests 库 + 调度器(如 schedule 或 APScheduler)模拟自动化流程。
  • 外部群限制:外部群(客户群)消息发送基于成员 externalUserId,不支持直接群 ID 推送。需确保应用有“客户联系”权限。

下面我一步步拆解实现,包括准备工作、代码示例和注意事项。代码基于 Python 3.10+,使用 requests 库(安装:pip install requests)。

1. 准备工作(获取权限和 Token)

  • 注册第三方应用:在企业微信管理后台创建应用,获取 agentidcorpidcorpsecret。启用“客户联系”权限以访问外部群。
  • 获取 access_token:这是所有 API 的入口。Token 有效期 7200 秒,需缓存或定时刷新。
  • 外部群成员 ID:通过 /cgi-bin/externalcontact/groupchat/list API 获取群列表和成员 externalUserId

Python 获取 Token 示例:

import requests
import time
import json

# 配置(替换为你的实际值)
CORP_ID = "your_corpid"          # 企业 ID
CORP_SECRET = "your_corpsecret"  # 应用 Secret
AGENT_ID = "your_agentid"        # 应用 ID

def get_access_token():
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={CORP_SECRET}"
    response = requests.get(url)
    data = response.json()
    if data.get("errcode") == 0:
        return data["access_token"]
    else:
        raise ValueError(f"获取 Token 失败: {data['errmsg']}")

# 测试获取
token = get_access_token()
print(f"Access Token: {token}")

2. RPA 模式下实现外部群主动调用

  • RPA 核心思路:用 Python 脚本模拟 RPA 流程,通过定时任务或事件触发(如监控数据库/文件变化)主动调用 API,向外部群成员推送消息。实现“主动”效果。
  • 不支持直接从群调用:外部群无法直接“调用”你的应用(无 webhook 入口)。替代方案:应用侧监控 + 推送(如每分钟检查条件,满足时推送)。
  • 示例场景:监控外部事件(如订单变化),主动推送通知到外部群成员。

完整 Python 示例(主动推送文本消息到外部群成员):

import requests
import schedule  # pip install schedule,用于 RPA 定时任务
import time

# 从步骤1获取 Token
TOKEN = get_access_token()  # 实际中需定时刷新

def send_message_to_external_group(user_ids, content):
    """
    主动推送消息到外部群成员(RPA 模式下模拟主动调用)
    :param user_ids: 外部用户 ID 列表,格式 "user1|user2"(从群详情获取)
    :param content: 消息内容
    """
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={TOKEN}"
    payload = {
        "touser": user_ids,           # 外部用户 ID,用 | 分隔
        "msgtype": "text",            # 支持 text/image/voice 等
        "agentid": AGENT_ID,
        "text": {"content": content},
        "enable_id_trans": 0,         # 0: 不翻译 ID;1: 翻译
        "enable_duplicate_check": 0,  # 0: 不检查重复;1: 检查(默认1800秒内)
    }

    response = requests.post(url, json=payload)
    data = response.json()
    if data.get("errcode") == 0:
        print(f"消息发送成功,msgid: {data.get('msgid')}")
    else:
        print(f"发送失败: {data['errmsg']}")

# RPA 示例:每分钟检查条件并主动推送
def rpa_task():
    # 模拟检查外部事件(实际可连数据库/文件/其他 API)
    if some_condition_met():  # 替换为你的逻辑,如订单状态变化
        user_ids = "external_user1|external_user2"  # 从群详情 API 获取
        content = "订单更新:您的订单 #123 已发货!请查收。"
        send_message_to_external_group(user_ids, content)

# 调度 RPA 任务
schedule.every(1).minutes.do(rpa_task)  # 每分钟执行一次

# 运行 RPA 循环
while True:
    schedule.run_pending()
    time.sleep(1)
  • 扩展:发送其他类型消息(如 markdown 或卡片):
    修改 msgtype 和对应字段。例如 markdown:
  payload = {
      "touser": user_ids,
      "msgtype": "markdown",
      "agentid": AGENT_ID,
      "markdown": {"content": "# 标题\n**粗体** *斜体* [链接](https://example.com)"}
  }
  • 获取外部群成员(辅助 API):
  def get_external_group_members(group_id):
      url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/groupchat/get?access_token={TOKEN}"
      payload = {"chat_id": group_id}
      response = requests.post(url, json=payload)
      data = response.json()
      if data.get("errcode") == 0:
          members = [member['userid'] for member in data['group_chat']['member_list'] if member['type'] == 2]  # type=2 为外部联系人
          return "|".join(members)
      else:
          raise ValueError(f"获取群成员失败: {data['errmsg']}")

3. 注意事项 & 常见坑

  • 权限问题:确保应用有“客户联系 > 客户群管理”权限。无权限会返回 invaliduserunlicenseduser
  • Token 刷新:每 2 小时刷新一次,避免失效。实际项目用缓存(如 Redis)。
  • 频率限制:每个应用每天最多 200 次推送(每 1000 人次);每用户 30 次/分;每小时 1000 次。RPA 需控制节奏。
  • RPA 优化:用 APScheduler 代替 schedule,支持更复杂调度。集成 Selenium(如需模拟 UI 操作)。
  • 调试:用企业微信的“API 调试工具”测试接口。日志用 logging 模块记录。
  • 安全:Token 和 Secret 不要硬编码,用环境变量或配置文件。
  • 2026 年趋势:企业微信正加强 AI/RPA 集成,未来可能有更多 webhook 支持“群主动触发”。

如果这个实现不完全匹配你的场景(比如你指的 RPA 是特定工具集成,或需要响应式 webhook),可以提供更多细节(如具体外部群调用什么功能),我可以调整代码或再查文档!

文章已创建 4725

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

相关文章

开始在上面输入您的搜索词,然后按回车进行搜索。按ESC取消。

返回顶部