import requests
# !!!请替换为你的真实凭证 !!!
CORP_ID = "ww123" # 你的企业ID
CORP_SECRET = "123" # 具有“客户联系”权限的应用Secret
def get_wecom_token(corp_id, corp_secret):
"""
获取企业微信 access_token
"""
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
params = {
"corpid": corp_id,
"corpsecret": corp_secret
}
try:
resp = requests.get(url, params=params).json()
if resp.get("errcode") == 0:
print(f"✅ Token 获取成功 (有效期 {resp['expires_in']} 秒)")
print(f"access_token: {resp['access_token']}")
return resp['access_token']
else:
print(f"❌ 获取失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
return None
except Exception as e:
print(f"🚨 请求异常: {e}")
return None
if __name__ == "__main__":
token = get_wecom_token(CORP_ID, CORP_SECRET)import requests
import json
import logging
import time
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 你的固定 Token
FIXED_TOKEN = "GEqSXVi54nTynmaJ4q5VC09aOEbhppNj0_-8-1Ce0J_tQrfoGmJdE47xJXw9cKHbxdKYrRVy_6pnjRwCcGXUTIJtxi6ppHjGFV2pu9CV1N8rrCCXthQtE_EOgnoQrGeb0-9ujo88OklHC0JuYj0aXCaIa5NIeHthABbtpuTBDGAlFxU2E7cbf9V8vHuz0mqCuVl9kDHj1sik8Wu0DdyEmg"
class WeComAPI:
def __init__(self, fixed_token=FIXED_TOKEN):
"""
使用固定 Token 初始化
:param fixed_token: 你的固定 Token
"""
self.fixed_token = fixed_token
self.base_url = "https://qyapi.weixin.qq.com/cgi-bin" # 修正基础URL
logger.info("✅ 使用固定 Token 初始化成功")
def get_external_groupchat_list(self, owner_userid_list=None, limit=100, cursor=""):
"""
获取客户群列表
:param owner_userid_list: 群主ID列表,例如 ["userid1", "userid2"]
:param limit: 分页大小,建议 100
:param cursor: 分页游标,首次调用传空字符串
:return: 群聊列表和下一页游标
"""
url = f"{self.base_url}/externalcontact/groupchat/list"
params = {"access_token": self.fixed_token} # 直接使用固定 Token
# 构建请求体
request_body = {
"limit": limit,
"cursor": cursor,
"status_filter": 0
}
# 如果有指定群主,则添加 owner_filter
if owner_userid_list:
request_body["owner_filter"] = {
"userid_list": owner_userid_list
}
try:
resp = requests.post(url, params=params, json=request_body, timeout=10).json()
logger.info(f"调用 groupchat/list 返回: {json.dumps(resp, ensure_ascii=False)}")
if resp.get("errcode") == 0:
group_chat_list = resp.get("group_chat_list", [])
next_cursor = resp.get("next_cursor", "")
logger.info(f"✅ 成功获取到 {len(group_chat_list)} 个客户群,下一页游标: '{next_cursor}'")
return group_chat_list, next_cursor
else:
logger.error(f"❌ 接口调用失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
return None, None
except Exception as e:
logger.error(f"🚨 请求异常: {e}")
return None, None
def send_to_external_group(self, chat_id, content):
"""
向指定的外部群(客户群)发送文本消息
注意:这是"代办任务"模式,需群主在手机端确认后才会真正发送
"""
url = f"{self.base_url}/externalcontact/groupchat/send"
params = {"access_token": self.fixed_token}
request_body = {
"chat_id": chat_id,
"msgtype": "text",
"text": {"content": content}
}
try:
# 1. 先打印请求信息(用于调试)
logger.info(f"请求URL: {url}")
logger.info(f"请求参数: {params}")
logger.info(f"请求体: {json.dumps(request_body, ensure_ascii=False)}")
# 2. 发送请求,捕获原始响应
response = requests.post(url, params=params, json=request_body, timeout=10)
raw_text = response.text
logger.info(f"接口返回原始文本: {raw_text}") # 关键调试信息
# 3. 尝试解析JSON
try:
resp = response.json()
except json.JSONDecodeError as e:
logger.error(f"JSON解析失败: {e}")
# 如果不是空响应,打印前200个字符
if raw_text:
logger.error(f"原始响应内容: {raw_text[:200]}")
return None
# 4. 处理解析后的响应
if resp.get("errcode") == 0:
msgid = resp.get("msgid")
logger.info(f"✅ 消息发送任务创建成功,msgid: {msgid}")
logger.info("⚠️ 请提醒群主在企业微信「群发助手」中确认发送")
return msgid
else:
logger.error(f"❌ 发送失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
return None
except requests.exceptions.Timeout:
logger.error("🚨 请求超时,请检查网络或增加超时时间")
return None
except requests.exceptions.ConnectionError:
logger.error("🚨 网络连接错误,请检查网络设置")
return None
except Exception as e:
logger.error(f"🚨 请求发送接口异常: {e}")
return None
# ====================== 使用示例 ======================
if __name__ == "__main__":
# 初始化客户端(使用固定Token)
client = WeComAPI()
# 1. 获取客户群列表
TEST_OWNER_USERID = ["HuJing"] # 测试用的群主ID
logger.info("正在获取客户群列表(第一页)...")
group_list, next_cursor = client.get_external_groupchat_list(
owner_userid_list=TEST_OWNER_USERID,
limit=100,
cursor=""
)
if group_list is not None:
# 打印获取到的群聊信息
for idx, group in enumerate(group_list, 1):
print(f"{idx}. chat_id: {group.get('chat_id')}, status: {group.get('status')}")
# 2. 向第一个群发送消息
if group_list:
test_chat_id = group_list[0].get('chat_id')
order_content = "【酒店新订单】\n订单号:201111111111\n酒店:XX温泉酒店\n房型:xxxxx房\n入住时间:2026-55-55"
print(f"\n正在向群 {test_chat_id} 发送消息...")
msgid = client.send_to_external_group(test_chat_id, order_content)
if msgid:
print(f"✅ 消息任务已创建,任务ID: {msgid}")
print("⚠️ 重要:请通知群主在企业微信手机端的「群发助手」中确认发送")
else:
print("❌ 消息发送失败")
else:
logger.error("获取客户群列表失败,请检查配置。")客户与上下游---客户联系---客户---企业全部客户数
可以看到。所有的外部联系人。点--已服务的外部联系人 可以看到列表。
客户与上下游---客户联系---客户(客户 --成员可用企业微信添加客户的微信,并在这里与他们联系,企业可统一管理这些客户。)
这个地方有个api 小字,点开 可以配置我的应用。(可调用接口的应用)
配置到这一步,我们可以获取 微信群了。
INFO:__main__:调用 groupchat/list 返回: {"errcode": 0, "errmsg": "ok", "group_chat_list": [{"chat_id": "wrG3K7QAAAlWZybyF3ZZ47lqPrJI0PLA", "status": 0}]}

https://developer.work.weixin.qq.com/document/path/90244

怎么获取 内部群的 chatid ~~~~~~~~~~~~~~~~~~~~~~~~~~ 很遗憾,企业微信目前没有提供任何 API 接口,可以直接获取员工手动创建的内部群 chatid。 这和之前能拉取“客户群(外部群)”列表不一样,内部群的 chatid 获取受到严格限制,目前仅有以下两种可行途径: 通过 API 创建群聊时获取:只有使用“创建群聊会话”接口(appchat/create)由你的应用新建的群,才会在接口的返回值中直接拿到 chatid。 通过会话内容存档(收费功能)间接获取:如果你的企业开通了“会话内容存档”功能,在处理归档的聊天消息记录时,消息数据里会携带 chatid,你可以从中提取并记录下来。 如果是员工早就在客户端手动拉好、且非 API 创建的内部群,现阶段无法通过后台接口批量查到它们的 chatid,只能在群内产生流经存档的消息后,去存档数据里反查。 注意点,我们用api创建的时候要保存好chatid
import requests
import json
FIXED_TOKEN = "U_7xWRvoZKHrRhINT3-CY5NJsDjfGJnuw2Q2g7Q6ja1dPvuvKUzSomH2ganXCv_C-XTqczQmM5vNVKxoq5r8CmWSYu6Ll8oRxV4NbZbO4vfwMZxGc5A2lSYqSH15-sH2iUKqll7TX8H5bibDNMqXbTadk8nI8hwLh-2IxtE_OG-qpJrmaiL8vNH7y3C9xHLyLHAUwBfelnlvKWV-BXFPxw"
def create_internal_group(token, group_name, user_list, owner_user=None):
"""
创建企业微信内部群
:param token: 自建应用的 access_token
:param group_name: 群名称
:param user_list: 群成员 userid 列表(至少2人)
:param owner_user: 指定群主 userid(可选,不填则系统随机选)
:return: 创建成功的 chatid 或 None
"""
url = f"https://qyapi.weixin.qq.com/cgi-bin/appchat/create?access_token={token}"
body = {
"name": group_name,
"userlist": user_list
}
if owner_user:
body["owner"] = owner_user
try:
resp = requests.post(url, json=body, timeout=10).json()
if resp.get("errcode") == 0:
chatid = resp.get("chatid")
print(f"✅ 内部群创建成功,chatid 为:{chatid}")
return chatid
else:
print(f"❌ 创建群失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
return None
except Exception as e:
print(f"🚨 请求异常: {e}")
return None
if __name__ == "__main__":
# 示例:创建内部群
members = ["HuJing", "ChenRun", "ZhangMeng"] # 替换为真实 userid,至少 2 人
create_internal_group(
token=FIXED_TOKEN,
group_name="自动创建的内部群001",
user_list=members,
owner_user="HuJing"
)
import requests
FIXED_TOKEN = "U_7xWRvoZKHrRhINT3-CY5NJsDjfGJnuw2Q2g7Q6ja1dPvuvKUzSomH2ganXCv_C-XTqczQmM5vNVKxoq5r8CmWSYu6Ll8oRxV4NbZbO4vfwMZxGc5A2lSYqSH15-sH2iUKqll7TX8H5bibDNMqXbTadk8nI8hwLh-2IxtE_OG-qpJrmaiL8vNH7y3C9xHLyLHAUwBfelnlvKWV-BXFPxw"
CHAT_ID = "wrG3K7QAAA_FbnXaM91pS3ibQwkpCAMg" # 你刚刚创建成功的 chatid
def send_text_to_internal_group(token, chatid, content):
url = f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send?access_token={token}"
payload = {
"chatid": chatid,
"msgtype": "text",
"text": {
"content": content
},
"safe": 0
}
resp = requests.post(url, json=payload, timeout=10).json()
if resp.get("errcode") == 0:
print("✅ 消息发送成功")
else:
print(f"❌ 发送失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
if __name__ == "__main__":
send_text_to_internal_group(
token=FIXED_TOKEN,
chatid=CHAT_ID,
content="🤖 这是来自 API 的第一条测试消息!"
)
可以换行发送
if __name__ == "__main__":
poem_parts = [
"《热爱生命》\n作者:汪国真\n\n我不去想是否能够赢得爱情",
"既然钟情于玫瑰\n就勇敢地吐露真诚",
"我不去想身后会不会袭来寒风冷雨",
"既然目标是地平线\n留给世界的只能是背影",
"我不去想未来是平坦还是泥泞",
"只要热爱生命\n一切,都在意料之中",
"\n—— 来自 API 的诗意推送 🌹"
]
for part in poem_parts:
send_text_to_internal_group(FIXED_TOKEN, CHAT_ID, part)
time.sleep(1) # 防止过快,看起来更像人工发送
import requests
FIXED_TOKEN = "U_7xWRvoZKHrRhINT3-CY5NJsDjfGJnuw2Q2g7Q6ja1dPvuvKUzSomH2ganXCv_C-XTqczQmM5vNVKxoq5r8CmWSYu6Ll8oRxV4NbZbO4vfwMZxGc5A2lSYqSH15-sH2iUKqll7TX8H5bibDNMqXbTadk8nI8hwLh-2IxtE_OG-qpJrmaiL8vNH7y3C9xHLyLHAUwBfelnlvKWV-BXFPxw"
BASE = "https://qyapi.weixin.qq.com/cgi-bin"
def get_all_departments(token):
"""获取所有部门"""
url = f"{BASE}/department/list?access_token={token}"
r = requests.get(url).json()
if r.get("errcode") != 0:
raise Exception(f"获取部门失败: {r}")
return r.get("department", [])
def get_users_in_department(token, dept_id):
"""获取指定部门下的员工详情"""
url = f"{BASE}/user/list?access_token={token}&department_id={dept_id}&fetch_child=0"
r = requests.get(url).json()
if r.get("errcode") != 0:
# 有些部门可能无权限,跳过
return []
return r.get("userlist", [])
if __name__ == "__main__":
departments = get_all_departments(FIXED_TOKEN)
all_users = []
for dept in departments:
dept_id = dept["id"]
dept_name = dept["name"]
users = get_users_in_department(FIXED_TOKEN, dept_id)
for u in users:
u["_dept_name"] = dept_name # 方便看属于哪个部门
all_users.append(u)
print(f"共获取到 {len(all_users)} 名员工:")
for u in all_users:
print(u.get("name"), u.get("userid"), u.get("_dept_name"))
import requests
FIXED_TOKEN = "U_7xWRvoZKHrRhINT3-CY5NJsDjfGJnuw2Q2g7Q6ja1dPvuvKUzSomH2ganXCv_C-XTqczQmM5vNVKxoq5r8CmWSYu6Ll8oRxV4NbZbO4vfwMZxGc5A2lSYqSH15-sH2iUKqll7TX8H5bibDNMqXbTadk8nI8hwLh-2IxtE_OG-qpJrmaiL8vNH7y3C9xHLyLHAUwBfelnlvKWV-BXFPxw"
CHAT_ID = "wrG3K7QAAA_FbnXaM91pS3ibQwkpCAMg"
url = f"https://qyapi.weixin.qq.com/cgi-bin/appchat/get?access_token={FIXED_TOKEN}&chatid={CHAT_ID}"
resp = requests.get(url).json()
if resp.get("errcode") == 0:
chat_info = resp.get("chat_info", {})
members = chat_info.get("userlist", [])
owner = chat_info.get("owner")
name = chat_info.get("name")
print(f"群名:{name}")
print(f"群主:{owner}")
print(f"成员数:{len(members)}")
print("成员列表:")
for user_id in members:
print(user_id)
else:
print(f"❌ 获取失败 [{resp.get('errcode')}]: {resp.get('errmsg')}")
import json
import requests
import time
FIXED_TOKEN = "U_7xWRvoZKHrRhINT3-CY5NJsDjfGJnuw2Q2g7Q6ja1dPvuvKUzSomH2ganXCv_C-XTqczQmM5vNVKxoq5r8CmWSYu6Ll8oRxV4NbZbO4vfwMZxGc5A2lSYqSH15-sH2iUKqll7TX8H5bibDNMqXbTadk8nI8hwLh-2IxtE_OG-qpJrmaiL8vNH7y3C9xHLyLHAUwBfelnlvKWV-BXFPxw"
BASE_URL = "https://qyapi.weixin.qq.com/cgi-bin"
#获取群 成员
def get_group_members(chatid):
url = f"{BASE_URL}/appchat/get?access_token={FIXED_TOKEN}&chatid={chatid}"
r = requests.get(url).json()
if r.get("errcode") == 0:
return r["chat_info"].get("userlist", [])
else:
raise Exception(f"获取成员失败: {r}")
# 添加群成员(修复版)
def add_member_to_group(chatid, userid):
url = f"{BASE_URL}/appchat/update?access_token={FIXED_TOKEN}"
# 接口要求必须是 数组格式!!!
payload = {
"chatid": chatid,
"add_user_list": [userid] # 单个成员也要放列表里
}
try:
resp = requests.post(url, json=payload, timeout=10).json()
print(f"添加成员返回: {json.dumps(resp, ensure_ascii=False)}")
return resp.get("errcode") == 0
except Exception as e:
print(f"请求异常: {e}")
return False
# 删除群成员
def remove_member_from_group(chatid, userid):
url = f"{BASE_URL}/appchat/update?access_token={FIXED_TOKEN}"
payload = {
"chatid": chatid,
"del_user_list": [userid] # 删除也要数组格式
}
r = requests.post(url, json=payload).json()
print(f"删除成员返回: {json.dumps(r, ensure_ascii=False)}")
return r.get("errcode") == 0
#解散群
def dissolve_group(chatid):
members = get_group_members(chatid)
url = f"{BASE_URL}/appchat/update?access_token={FIXED_TOKEN}"
r = requests.post(url, json={
"chatid": chatid,
"del_user_list": members
}).json()
return r.get("errcode") == 0
def rename_group(chatid, new_name):
"""
修改内部群群名
:param chatid: 群 chatid
:param new_name: 新群名
"""
url = f"{BASE_URL}/appchat/update?access_token={FIXED_TOKEN}"
payload = {
"chatid": chatid,
"name": new_name
}
try:
resp = requests.post(url, json=payload, timeout=10).json()
print(f"修改群名返回: {json.dumps(resp, ensure_ascii=False)}")
return resp.get("errcode") == 0
except Exception as e:
print(f"请求异常: {e}")
return False
if __name__ == "__main__":
CHAT_ID = "wrG3K7QAAA_FbnXaM91pS3ibQwkpCAMg" # 你刚刚创建成功的 chatid
# remove_member_from_group(CHAT_ID, "LiuXiaoHui")
# add_member_to_group(CHAT_ID, "LiuXiaoHui")
#
# user_list = get_group_members(CHAT_ID)
# print(user_list)
# 3. 修改群名
rename_group(CHAT_ID, "API自动化测试群-已改名")站长微信:xiaomao0055
站长QQ:14496453