ff33edb0d1
- 新增 server_864 独立 provider 目录，接入登录、消息轮询、联系人、群资料、用户资料与朋友圈基础能力 - 扩展 gateway、robot 与配置归一化逻辑，支持 server_864/864 别名和 WECHAT_SERVER_KEY - 更新配置示例与多版本适配路线图，明确 864 第一版接入范围和后续待补项
123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
import base64
|
|
import os
|
|
from typing import Union
|
|
|
|
import aiofiles
|
|
|
|
from wechat_ipad.models.friend_circle_info import build_friend_circle_xml
|
|
from wechat_ipad.providers.server_864.base import Server864APIClientBase
|
|
|
|
|
|
class FriendCircleMixin(Server864APIClientBase):
|
|
"""864 朋友圈相关接口。"""
|
|
|
|
async def get_friend_circle_list(self, max_id: int = 0, first_page_md5: str = "") -> dict:
|
|
data = await self._request_data(
|
|
"post",
|
|
"/sns/SendSnsTimeLine",
|
|
json_body={"UserName": "", "FirstPageMD5": first_page_md5, "MaxID": int(max_id)},
|
|
timeout=30,
|
|
)
|
|
return dict(data or {})
|
|
|
|
async def get_friend_circle_detail(self, towxid: str, max_id: int = 0, first_page_md5: str = "") -> dict:
|
|
data = await self._request_data(
|
|
"post",
|
|
"/sns/SendSnsUserPage",
|
|
json_body={"UserName": towxid, "FirstPageMD5": first_page_md5, "MaxID": int(max_id)},
|
|
timeout=30,
|
|
)
|
|
return dict(data or {})
|
|
|
|
async def get_friend_circle_id_detail(self, object_id: Union[str, int], towxid: str = "") -> dict:
|
|
del towxid
|
|
data = await self._request_data(
|
|
"post",
|
|
"/sns/SendSnsObjectDetailById",
|
|
json_body={"Id": str(object_id), "BlackList": [], "LocationVal": 0, "Location": {}},
|
|
timeout=30,
|
|
)
|
|
return dict(data or {})
|
|
|
|
async def publish_friend_circle(
|
|
self,
|
|
content: str,
|
|
media_items: list[dict] | None = None,
|
|
blacklist: str = "",
|
|
with_user_list: str = "",
|
|
) -> dict:
|
|
payload = {
|
|
"ContentStyle": 1 if media_items else 0,
|
|
"ContentUrl": "",
|
|
"Description": "",
|
|
"Privacy": 0,
|
|
"Content": content,
|
|
"MediaList": list(media_items or []),
|
|
"WithUserList": [item for item in str(with_user_list or "").split(",") if item],
|
|
"GroupUserList": [],
|
|
"BlackList": [item for item in str(blacklist or "").split(",") if item],
|
|
"LocationInfo": None,
|
|
}
|
|
data = await self._request_data("post", "/sns/SendFriendCircle", json_body=payload, timeout=60)
|
|
return dict(data or {})
|
|
|
|
async def friend_circle_comment(
|
|
self,
|
|
object_id: str,
|
|
content: str = "",
|
|
type: int = 2,
|
|
reply_comment_id: int = 0,
|
|
) -> dict:
|
|
payload = {
|
|
"SnsCommentList": [{
|
|
"OpType": int(type),
|
|
"ItemID": str(object_id),
|
|
"ToUserName": "",
|
|
"Content": content,
|
|
"CreateTime": 0,
|
|
"ReplyCommentID": int(reply_comment_id),
|
|
"ReplyItem": {"UserName": "", "NickName": "", "OpType": 0, "Source": 0},
|
|
}],
|
|
"Tx": False,
|
|
}
|
|
data = await self._request_data("post", "/sns/SendSnsComment", json_body=payload, timeout=30)
|
|
return {"result": data}
|
|
|
|
async def friend_circle_operation(self, object_id: str, type: int, comment_id: int = 0) -> dict:
|
|
payload = {
|
|
"SnsObjectOpList": [{
|
|
"SnsObjID": str(object_id),
|
|
"OpType": int(type),
|
|
"DataLen": 0,
|
|
"Data": [],
|
|
"Ext": int(comment_id),
|
|
}]
|
|
}
|
|
data = await self._request_data("post", "/sns/SendSnsObjectOp", json_body=payload, timeout=30)
|
|
return {"result": data}
|
|
|
|
async def upload_friend_circle_media(self, media: Union[str, bytes, os.PathLike]) -> dict:
|
|
if isinstance(media, str):
|
|
media_base64 = media.split(",", 1)[1] if "," in media else media
|
|
elif isinstance(media, bytes):
|
|
media_base64 = base64.b64encode(media).decode()
|
|
elif isinstance(media, os.PathLike):
|
|
async with aiofiles.open(media, "rb") as f:
|
|
media_base64 = base64.b64encode(await f.read()).decode()
|
|
else:
|
|
raise ValueError("media should be str, bytes, or path")
|
|
|
|
data = await self._request_data(
|
|
"post",
|
|
"/sns/UploadFriendCircleImage",
|
|
json_body={"ImageDataList": [media_base64], "VideoDataList": []},
|
|
timeout=60,
|
|
)
|
|
return {"result": data}
|
|
|
|
async def publish_friend_circle_by_xml(self, content: str, media_items: list[dict] | None = None) -> dict:
|
|
"""为后续扩展保留 XML 版本朋友圈发送入口。"""
|
|
xml_content = build_friend_circle_xml(self.wxid, content, media_items=media_items)
|
|
data = await self._request_data("post", "/sns/SendFriendCircleByXMl", json_body=xml_content, timeout=60)
|
|
return {"result": data}
|