stefanfeng
0cc43a0d8c
Prevents stale signature values from leaking into the MD5 signing calculation, matching the news_service.py pattern. Adds a test that passes a stale signature in extra params and verifies the returned signature is freshly computed.
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""盒子 IM 客户端 — 封装网易云信 IM 接口调用"""
|
|
import hashlib
|
|
import random
|
|
import string
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
|
|
class BoxIMClient:
|
|
"""盒子 IM 客户端，通过会会平台网关调用网易云信 IM"""
|
|
|
|
def __init__(self, config: dict):
|
|
self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040")
|
|
self.app_id = config.get("HUIHUI_APP_ID", "")
|
|
self.access_id = config.get("HUIHUI_ACCESS_ID", "")
|
|
self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "")
|
|
|
|
def _build_sign_params(self, extra: dict) -> dict:
|
|
"""构建带签名的请求参数（复用 news_service 签名模式）"""
|
|
nonce = "".join(random.choices(string.ascii_lowercase + string.digits, k=12))
|
|
timestamp = datetime.now().strftime("%Y%m%d%H%M%S") # 24小时制
|
|
params = {
|
|
"appId": self.app_id,
|
|
"accessId": self.access_id,
|
|
"nonce": nonce,
|
|
"timestamp": timestamp,
|
|
**extra,
|
|
}
|
|
# 计算签名 — 排序 key, 过滤空值, 拼接后加 accessSecret, MD5 大写
|
|
keys = sorted(params.keys())
|
|
sign_parts = []
|
|
for k in keys:
|
|
if k in ("signature", "accessSecret"):
|
|
continue
|
|
v = params.get(k)
|
|
if v and v != "" and v != []:
|
|
sign_parts.append(f"{k}={v}")
|
|
sign_str = "&".join(sign_parts) + f"&accessSecret={self.access_secret}"
|
|
signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
|
|
params["signature"] = signature
|
|
params["signType"] = "MD5"
|
|
params["signVersion"] = "1.0"
|
|
return params
|
|
|
|
async def get_credentials(self, user_id: str) -> Optional[dict]:
|
|
"""获取用户的网易云信 IM 凭证 (accid, token)"""
|
|
params = self._build_sign_params({"userId": user_id})
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
r = await client.post(
|
|
f"{self.base_url}/box/netease",
|
|
params=params,
|
|
)
|
|
data = r.json()
|
|
if data.get("code") in (0, 200):
|
|
return data.get("data", {})
|
|
return None
|
|
|
|
async def send_p2p_message(
|
|
self, from_accid: str, to_accid: str, content: str
|
|
) -> bool:
|
|
"""发送单聊消息（文本）"""
|
|
params = self._build_sign_params({
|
|
"from": from_accid,
|
|
"to": to_accid,
|
|
"msgType": "text",
|
|
"content": content,
|
|
})
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
r = await client.post(
|
|
f"{self.base_url}/box/message/send/p2p",
|
|
params=params,
|
|
)
|
|
data = r.json()
|
|
return data.get("code") in (0, 200)
|