群管理是微信机器人需求最密集的场景。WTAPI的群操作接口覆盖建群、改名、邀请成员(group/inviteMember)、移除成员等完整生命周期。单个接口不难,难的是把多个操作编排成一条自动跑的链路。这篇给可直接套用的代码方案,并把"能全自动"和"必须留人工确认"的动作分清楚。

一、统一调用层封装

每个群操作都走同一个请求函数,集中处理鉴权、超时和成功码:

import requests, json, redis, time

BASE = "https://wx.chuapi.com"
HEADERS = {
    "X-finder-TOKEN": "<平台Token>",
    "Authorization": "Bearer <Bearer>",
    "Content-Type": "application/json",
}
APP_ID = "<appId>"

def call(path, instance_id, **params):
    body = {"appId": APP_ID, "instanceId": instance_id, **params}
    r = requests.post(f"{BASE}{path}", headers=HEADERS,
                      json=body, timeout=10).json()
    if r.get("code") != "1000":
        raise RuntimeError(f"{path} 失败: {r}")
    return r.get("data", {})

二、原子操作逐个封装

def create_group(inst, name):
    return call("/finder/v2/api/group/create", inst, name=name)

def rename_group(inst, room_id, name):
    return call("/finder/v2/api/group/rename", inst,
                chatRoomId=room_id, name=name)

def invite_members(inst, room_id, wxids):
    return call("/finder/v2/api/group/inviteMember", inst,
                chatRoomId=room_id, wxids=wxids)

def send_group_text(inst, room_id, content):
    return call("/finder/v2/api/postText", inst,
                toWxid=room_id, content=content)

三、全自动编排:建群一条龙

建群→改名→邀人→欢迎语这条链路可以全自动,配合状态持久化支持失败续跑:

rds = redis.Redis()

def auto_build_group(inst, name, wxids):
    state_key = f"group_flow:{name}"
    state = json.loads(rds.get(state_key) or "{}")

    try:
        if "created" not in state:
            room_id = create_group(inst, name)["chatRoomId"]
            state["room_id"] = room_id
            state["step"] = "created"
            rds.setex(state_key, 3600, json.dumps(state))

        if "renamed" not in state:
            rename_group(inst, state["room_id"], f"{name}-交流群")
            state["step"] = "renamed"
            rds.setex(state_key, 3600, json.dumps(state))

        if "invited" not in state:
            invite_members(inst, state["room_id"], wxids)
            state["step"] = "invited"
            rds.setex(state_key, 3600, json.dumps(state))

        send_group_text(inst, state["room_id"], "欢迎进群,请先看群规")
        rds.delete(state_key)
        return state["room_id"]
    except Exception as e:
        alert_admin(f"建群流程中断 {name}: {e}")
        raise

四、踢人:自动识别,人工执行

踢人不要做成全自动。正确做法是机器人自动识别违规(如命中广告关键词),但把动作推进审核队列,管理员确认后才真正调用移除接口:

def on_violation(inst, room_id, wxid, reason):
    rds.lpush("kick_review_queue", json.dumps({
        "inst": inst, "room_id": room_id,
        "wxid": wxid, "reason": reason, "status": "pending",
    }))
    notify_admin(f"踢人申请待审:{wxid} / {reason}")

def approve_kick(task):
    call("/finder/v2/api/group/removeMember", task["inst"],
         chatRoomId=task["room_id"], wxids=[task["wxid"]])
    mark_reviewed(task["id"], "approved")

五、两条不能省的红线

一是限流,建群和批量邀人之间加随机延迟,短时间高频操作必触发风控:

import random
time.sleep(random.uniform(3, 15))  # 模拟真人节奏

二是幂等,每步操作落状态标记,重试时跳过已完成步骤,避免重复邀人、重复欢迎。

这样搭出来的群机器人,常规流程全自动跑,敏感动作有人把关,既省人力又不越界。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐