## 一、API概述### 1.1 什么是微信机器人API微信机器人API是一组用于与微信客户端交互的编程接口开发者可以通过这些接口实现自动化操作┌─────────────────────────────────────────────────────────────────┐│ 微信机器人API架构 │├─────────────────────────────────────────────────────────────────┤│ ││ 用户层 应用层 协议层 ││ ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ││ │ 开发者 │ ── API ─│ SDK/框架 │ ── 协议 ─│ 微信协议 │ ││ │ 业务系统 │ │ WTAPI等 │ │ iPad/手机 │ ││ └──────────┘ └──────────┘ └──────────┘ ││ │└─────────────────────────────────────────────────────────────────┘### 1.2 API能力矩阵| API类别 | 功能描述 | 适用场景 ||:---|:---|:---|| 消息API | 发送/接收文本、图片、语音、视频 | 客服自动化 || 联系人API | 获取好友列表、添加好友、备注管理 | 用户管理 || 群管理API | 创建群、拉人、踢人、群设置 | 社群运营 || 账号API | 获取账号信息、状态查询 | 监控告警 || 媒体API | 图片上传、文件下载、语音转文字 | 多媒体处理 |---## 二、核心API详解### 2.1 消息发送APIpythonimport requestsimport jsonclass WeChatAPI:微信API客户端def __init__(self, api_key: str, base_url: str https://api.wtapi.io):self.api_key api_keyself.base_url base_urlself.headers {Authorization: fBearer {api_key},Content-Type: application/json}def send_text_message(self, to_user: str, content: str) - dict:发送文本消息:param to_user: 目标用户ID:param content: 消息内容:return: API响应payload {action: sendText,toUser: to_user,content: content}response requests.post(f{self.base_url}/message,headersself.headers,datajson.dumps(payload))return response.json()def send_image_message(self, to_user: str, image_path: str) - dict:发送图片消息:param to_user: 目标用户ID:param image_path: 图片本地路径或URL:return: API响应files {image: open(image_path, rb)}data {action: sendImage,toUser: to_user}response requests.post(f{self.base_url}/message/image,headers{Authorization: fBearer {self.api_key}},datadata,filesfiles)return response.json()def send_group_message(self, group_id: str, content: str) - dict:发送群消息:param group_id: 群ID:param content: 消息内容:return: API响应payload {action: sendGroupText,groupID: group_id,content: content}response requests.post(f{self.base_url}/group/message,headersself.headers,datajson.dumps(payload))return response.json()### 2.2 联系人管理APIpythonclass ContactAPI:联系人管理APIdef __init__(self, client: WeChatAPI):self.client clientdef get_friend_list(self) - list:获取好友列表response self.client._request(GET, /contact/friends)return response.get(data, [])def search_friend(self, keyword: str) - list:搜索好友params {keyword: keyword}response self.client._request(GET, /contact/search, paramsparams)return response.get(data, [])def get_friend_info(self, user_id: str) - dict:获取好友详细信息response self.client._request(GET, f/contact/{user_id})return response.get(data, {})def update_friend_remark(self, user_id: str, remark: str) - dict:更新好友备注payload {remark: remark}response self.client._request(PUT, f/contact/{user_id}/remark, datapayload)return responsedef add_friend(self, v3: str, v4: str, message: str 你好) - dict:添加好友:param v3: 好友V3标识:param v4: 好友V4标识:param message: 验证消息:return: 添加结果payload {v3: v3,v4: v4,message: message}response self.client._request(POST, /contact/add, datapayload)return response### 2.3 群管理APIpythonclass GroupAPI:群管理APIdef __init__(self, client: WeChatAPI):self.client clientdef get_group_list(self) - list:获取群列表response self.client._request(GET, /group/list)return response.get(data, [])def get_group_info(self, group_id: str) - dict:获取群详细信息response self.client._request(GET, f/group/{group_id})return response.get(data, {})def get_group_members(self, group_id: str) - list:获取群成员列表response self.client._request(GET, f/group/{group_id}/members)return response.get(data, [])def add_group_member(self, group_id: str, user_ids: list) - dict:拉人进群payload {userIds: user_ids}response self.client._request(POST, f/group/{group_id}/members, datapayload)return responsedef remove_group_member(self, group_id: str, user_ids: list) - dict:移出群成员payload {userIds: user_ids}response self.client._request(DELETE, f/group/{group_id}/members, datapayload)return responsedef create_group(self, user_ids: list, topic: str ) - dict:创建群聊payload {userIds: user_ids,topic: topic}response self.client._request(POST, /group/create, datapayload)return response---## 三、API使用最佳实践### 3.1 错误处理pythonclass APIError(Exception):API错误异常def __init__(self, code: int, message: str):self.code codeself.message messagesuper().__init__(fAPI Error {code}: {message})class SafeWeChatAPI(WeChatAPI):安全的API客户端def _request(self, method: str, endpoint: str, **kwargs) - dict:安全请求封装max_retries 3retry_delay 1for attempt in range(max_retries):try:response requests.request(method,f{self.base_url}{endpoint},headersself.headers,**kwargs)response.raise_for_status()result response.json()if result.get(code) ! 0:raise APIError(result[code], result.get(message, Unknown error))return resultexcept requests.exceptions.RequestException as e:if attempt max_retries - 1:time.sleep(retry_delay * (attempt 1))continueraiseexcept APIError as e:if e.code in [429, 503]: # 限流或服务不可用if attempt max_retries - 1:time.sleep(retry_delay * (attempt 1))continueraise### 3.2 限流控制pythonimport timefrom collections import dequeclass RateLimiter:API限流控制器def __init__(self, max_requests: int, time_window: int)::param max_requests: 时间窗口内最大请求数:param time_window: 时间窗口秒self.max_requests max_requestsself.time_window time_windowself.request_times deque()def wait(self):等待直到可以发送请求now time.time()while self.request_times and now - self.request_times[0] self.time_window:self.request_times.popleft()if len(self.request_times) self.max_requests:wait_time self.time_window - (now - self.request_times[0])if wait_time 0:time.sleep(wait_time)self.request_times.append(time.time())class RateLimitedAPI(WeChatAPI):带限流的API客户端def __init__(self, api_key: str):super().__init__(api_key)self.rate_limiter RateLimiter(max_requests100, time_window60)def _request(self, method: str, endpoint: str, **kwargs) - dict:self.rate_limiter.wait()return super()._request(method, endpoint, **kwargs)---## 四、实战案例### 4.1 自动回复机器人pythonclass AutoReplyBot:自动回复机器人def __init__(self, api_key: str):self.api RateLimitedAPI(api_key)self.keyword_map {你好: 您好有什么可以帮到您,价格: 我们提供多种价格方案详情请咨询客服。,功能: 我们的产品支持消息自动回复、群管理、数据分析等功能。,试用: 可以免费试用7天需要帮您开通吗}def start(self):启动机器人print(机器人已启动开始监听消息...)while True:try:messages self.api.get_latest_messages()for msg in messages:if msg[type] text and not msg[isSelf]:reply self._get_reply(msg[content])if reply:self.api.send_text_message(msg[fromUser], reply)time.sleep(1)except Exception as e:print(fError: {e})time.sleep(5)def _get_reply(self, content: str) - str:获取回复内容for keyword, reply in self.keyword_map.items():if keyword in content:return replyreturn None### 4.2 群成员统计工具pythonclass GroupStatsTool:群成员统计工具def __init__(self, api_key: str):self.api WeChatAPI(api_key)def analyze_groups(self):分析所有群groups self.api.get_group_list()for group in groups:stats self._analyze_group(group[groupID])self._print_stats(group[name], stats)def _analyze_group(self, group_id: str) - dict:分析单个群members self.api.get_group_members(group_id)stats {total: len(members),male: 0,female: 0,unknown: 0,new_members: 0}for member in members:if member.get(gender) 1:stats[male] 1elif member.get(gender) 2:stats[female] 1else:stats[unknown] 1if self._is_new_member(member):stats[new_members] 1return statsdef _is_new_member(self, member: dict) - bool:判断是否新成员join_time member.get(joinTime, 0)return join_time (time.time() - 86400 * 7)def _print_stats(self, group_name: str, stats: dict):打印统计信息print(f\n群名称: {group_name})print(f总人数: {stats[total]})print(f男性: {stats[male]} ({stats[male]/stats[total]*100:.1f}%))print(f女性: {stats[female]} ({stats[female]/stats[total]*100:.1f}%))print(f未知: {stats[unknown]})print(f近7天新成员: {stats[new_members]})---## 五、API安全### 5.1 安全最佳实践| 安全措施 | 说明 | 重要性 ||:---|:---|:---:|| API密钥保护 | 不要硬编码密钥使用环境变量 | ⭐⭐⭐⭐⭐ || HTTPS加密 | 始终使用HTTPS传输数据 | ⭐⭐⭐⭐⭐ || 请求签名 | 对请求进行签名验证 | ⭐⭐⭐⭐ || IP白名单 | 限制允许访问的IP | ⭐⭐⭐ || 操作日志 | 记录所有API调用 | ⭐⭐⭐ |### 5.2 密钥管理pythonimport osfrom dotenv import load_dotenvclass SecureAPI(WeChatAPI):安全API客户端def __init__(self):load_dotenv()api_key os.getenv(WECHAT_API_KEY)if not api_key:raise ValueError(API密钥未配置)super().__init__(api_key)---## 六、总结微信机器人API为开发者提供了强大的自动化能力- **消息处理**发送、接收各类消息- **联系人管理**好友列表、备注管理- **群管理**创建群、拉人、踢人- **账号管理**状态监控、信息查询掌握这些API您可以构建各种自动化工具和机器人应用。---*本文仅用于技术交流和学习目的请在遵守相关法律法规和平台规则的前提下使用API。*