""" E云管家API客户端 """ import requests from typing import Optional, Dict, Any, List from loguru import logger from config import settings class ECloudClient: """E云管家API客户端""" def __init__(self): self.base_url = settings.ecloud_base_url.rstrip("/") self.headers = { "Content-Type": "application/json", "Authorization": settings.ecloud_authorization, } self.session = requests.Session() self.session.headers.update(self.headers) def get_contact_info(self, w_id: str, wc_id: str) -> Optional[Dict[str, Any]]: """ 获取联系人信息 Args: w_id: 登录实例标识 wc_id: 好友微信id/群id，多个使用英文逗号分隔 Returns: 联系人信息字典或列表，失败返回None - 单个wcId时返回字典 - 多个wcId时返回列表 """ try: url = f"{self.base_url}/getContact" payload = {"wId": w_id, "wcId": wc_id} logger.info(f"获取联系人信息: wId={w_id}, wcId={wc_id}") response = self.session.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get("code") == "1000": contact_data = result.get("data", []) if contact_data and len(contact_data) > 0: # 如果是单个wcId，返回第一个联系人信息 if "," not in wc_id: logger.info(f"成功获取联系人信息: wcId={wc_id}") return contact_data[0] else: # 如果是多个wcId，返回完整列表 logger.info(f"成功获取批量联系人信息: count={len(contact_data)}") return contact_data else: logger.warning(f"联系人信息为空: wcId={wc_id}") return None else: logger.error( f"获取联系人信息失败: wcId={wc_id}, code={result.get('code')}, message={result.get('message')}" ) return None except requests.exceptions.RequestException as e: logger.error(f"获取联系人信息网络错误: wcId={wc_id}, error={str(e)}") return None except Exception as e: logger.error(f"获取联系人信息异常: wcId={wc_id}, error={str(e)}") return None def send_text_message(self, w_id: str, wc_id: str, content: str, max_retries: int = None) -> bool: """ 发送文本消息 Args: w_id: 登录实例标识 wc_id: 接收人微信id/群id content: 文本内容消息 max_retries: 最大重试次数 Returns: 发送成功返回True，失败返回False """ if max_retries is None: from config import settings max_retries = settings.max_retry_count retry_count = 0 while retry_count <= max_retries: try: url = f"{self.base_url}/sendText" payload = {"wId": w_id, "wcId": wc_id, "content": content} logger.info( f"发送文本消息: wId={w_id}, wcId={wc_id}, content_length={len(content)}, retry={retry_count}" ) response = self.session.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get("code") == "1000": logger.info(f"文本消息发送成功: wcId={wc_id}") return True else: logger.error( f"文本消息发送失败: wcId={wc_id}, code={result.get('code')}, message={result.get('message')}" ) except requests.exceptions.RequestException as e: logger.error(f"发送文本消息网络错误: wcId={wc_id}, retry={retry_count}, error={str(e)}") except Exception as e: logger.error(f"发送文本消息异常: wcId={wc_id}, retry={retry_count}, error={str(e)}") retry_count += 1 if retry_count <= max_retries: from config import settings wait_time = settings.retry_delay * retry_count logger.info(f"等待重试: wcId={wc_id}, wait_time={wait_time}s") import time time.sleep(wait_time) logger.error( f"文本消息发送失败，已达最大重试次数: wcId={wc_id}, max_retries={max_retries}" ) return False def send_group_message(self, w_id: str, group_id: str, content: str) -> bool: """ 发送群聊消息 Args: w_id: 登录实例标识 group_id: 群id content: 文本内容消息 Returns: 发送成功返回True，失败返回False """ return self.send_text_message(w_id, group_id, content) def init_address_list(self, w_id: str) -> bool: """ 初始化通讯录列表 Args: w_id: 登录实例标识 Returns: 初始化成功返回True，失败返回False """ try: url = f"{self.base_url}/initAddressList" payload = {"wId": w_id} logger.info(f"初始化通讯录列表: wId={w_id}") response = self.session.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get("code") == "1000": logger.info(f"初始化通讯录列表成功: wId={w_id}") return True else: logger.error( f"初始化通讯录列表失败: wId={w_id}, code={result.get('code')}, message={result.get('message')}" ) return False except requests.exceptions.RequestException as e: logger.error(f"初始化通讯录列表网络错误: wId={w_id}, error={str(e)}") return False except Exception as e: logger.error(f"初始化通讯录列表异常: wId={w_id}, error={str(e)}") return False def get_address_list(self, w_id: str) -> Optional[Dict[str, Any]]: """ 获取通讯录列表 Args: w_id: 登录实例标识 Returns: 通讯录数据字典，失败返回None 返回格式: { "chatrooms": [...], # 群组列表 "friends": [...], # 好友列表 "ghs": [...], # 公众号列表 "others": [...] # 其他 } """ try: url = f"{self.base_url}/getAddressList" payload = {"wId": w_id} logger.info(f"获取通讯录列表: wId={w_id}") response = self.session.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get("code") == "1000": address_data = result.get("data", {}) logger.info(f"成功获取通讯录列表: wId={w_id}, friends_count={len(address_data.get('friends', []))}") return address_data else: logger.error( f"获取通讯录列表失败: wId={w_id}, code={result.get('code')}, message={result.get('message')}" ) return None except requests.exceptions.RequestException as e: logger.error(f"获取通讯录列表网络错误: wId={w_id}, error={str(e)}") return None except Exception as e: logger.error(f"获取通讯录列表异常: wId={w_id}, error={str(e)}") return None def send_group_at_message(self, w_id: str, wc_id: str, content: str, at_wc_ids: List[str], max_retries: int = None) -> bool: """ 发送群聊@消息 Args: w_id: 登录实例标识 wc_id: 接收方群id content: 文本内容消息（@的微信昵称需要自己拼接，必须拼接艾特符号，不然不生效） at_wc_ids: 艾特的微信id列表 max_retries: 最大重试次数 Returns: 发送成功返回True，失败返回False """ if max_retries is None: from config import settings max_retries = settings.max_retry_count retry_count = 0 while retry_count <= max_retries: try: url = f"{self.base_url}/sendText" # 将at_wc_ids列表用逗号拼接 at_str = ",".join(at_wc_ids) payload = { "wId": w_id, "wcId": wc_id, "content": content, "at": at_str } logger.info( f"发送群聊@消息: wId={w_id}, wcId={wc_id}, at={at_str}, content_length={len(content)}, retry={retry_count}" ) response = self.session.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get("code") == "1000": logger.info(f"群聊@消息发送成功: wcId={wc_id}, at={at_str}") return True else: logger.error( f"群聊@消息发送失败: wcId={wc_id}, at={at_str}, code={result.get('code')}, message={result.get('message')}" ) except requests.exceptions.RequestException as e: logger.error(f"发送群聊@消息网络错误: wcId={wc_id}, at={at_str}, retry={retry_count}, error={str(e)}") except Exception as e: logger.error(f"发送群聊@消息异常: wcId={wc_id}, at={at_str}, retry={retry_count}, error={str(e)}") retry_count += 1 if retry_count <= max_retries: from config import settings wait_time = settings.retry_delay * retry_count logger.info(f"等待重试: wcId={wc_id}, wait_time={wait_time}s") import time time.sleep(wait_time) logger.error( f"群聊@消息发送失败，已达最大重试次数: wcId={wc_id}, at={at_str}, max_retries={max_retries}" ) return False # 全局E云管家客户端实例 ecloud_client = ECloudClient()