WTAPI群聊操控:建群、邀人、踢人的完整代码方案
·
群管理是微信机器人需求最密集的场景。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)) # 模拟真人节奏
二是幂等,每步操作落状态标记,重试时跳过已完成步骤,避免重复邀人、重复欢迎。
这样搭出来的群机器人,常规流程全自动跑,敏感动作有人把关,既省人力又不越界。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐



所有评论(0)