"""
|
E云管家API客户端
|
"""
|
|
import requests
|
from typing import Optional, Dict, Any
|
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
|
"""
|
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:
|
logger.info(f"成功获取联系人信息: wcId={wc_id}")
|
return contact_data[0] # 返回第一个联系人信息
|
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) -> bool:
|
"""
|
发送文本消息
|
|
Args:
|
w_id: 登录实例标识
|
wc_id: 接收人微信id/群id
|
content: 文本内容消息
|
|
Returns:
|
发送成功返回True，失败返回False
|
"""
|
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)}"
|
)
|
|
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')}"
|
)
|
return False
|
|
except requests.exceptions.RequestException as e:
|
logger.error(f"发送文本消息网络错误: wcId={wc_id}, error={str(e)}")
|
return False
|
except Exception as e:
|
logger.error(f"发送文本消息异常: wcId={wc_id}, error={str(e)}")
|
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)
|
|
|
# 全局E云管家客户端实例
|
ecloud_client = ECloudClient()
|
