922 lines
31 KiB
Python
922 lines
31 KiB
Python
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import traceback
|
||
from io import BytesIO
|
||
from typing import Union
|
||
import xml.etree.cElementTree as ET
|
||
|
||
import emoji
|
||
import requests
|
||
|
||
import config
|
||
|
||
|
||
def initialize(app_id: str, app_name: str, app_ver: str, app_key: str) -> dict:
|
||
r"""初始化[K10001]
|
||
|
||
:param app_id: 软件唯一id，暂时随意
|
||
:param app_name: 软件名称
|
||
:param app_ver: 软件版本号
|
||
:param app_key: 软件Key，暂时随意
|
||
"""
|
||
api_json = {'api': 'K10001', 'appId': app_id, 'appName': app_name, 'appVer': app_ver, 'appKey': app_key}
|
||
return api_request(api_json)
|
||
|
||
|
||
def add_log(api_type: str, msg: str) -> dict:
|
||
r"""添加日志[K10002]
|
||
|
||
:param api_type: 1调试，2警告，3提示
|
||
:param msg: 日志内容
|
||
"""
|
||
api_json = {'api': 'K10002', 'type': api_type, 'msg': msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def follow_official_account(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""关注公众号[K10003]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 公众号的wxid
|
||
"""
|
||
api_json = {'api': 'K10003', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def transfer_receipt(robot_wxid: str, from_wxid: str, transfer_id: str) -> dict:
|
||
r"""转账收款[K10004]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 转账人的wxid
|
||
:param transfer_id: 收到的转账transferid
|
||
"""
|
||
api_json = {'api': 'K10004', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'transferid': transfer_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def transfer_refund(robot_wxid: str, from_wxid: str, transfer_id: str) -> dict:
|
||
r"""转账退款[K10005]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 转账人的wxid
|
||
:param transfer_id: 收到的转账transferid
|
||
"""
|
||
api_json = {'api': 'K10005', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'transferid': transfer_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def agree_group_invite(robot_wxid: str, from_wxid: str, url: str) -> dict:
|
||
r"""同意群聊邀请[K10006]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param url: 收到的消息里获取
|
||
"""
|
||
api_json = {'api': 'K10006', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'url': url}
|
||
return api_request(api_json)
|
||
|
||
|
||
def message_not_disturb(robot_wxid: str, from_wxid: str, is_flag: str) -> dict:
|
||
r"""消息免打扰[K10007]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param is_flag: 是否免打扰，0/1
|
||
"""
|
||
api_json = {'api': 'K10007', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'isFlag': is_flag}
|
||
return api_request(api_json)
|
||
|
||
|
||
def set_chat_top(robot_wxid: str, from_wxid: str, is_flag: str) -> dict:
|
||
r"""置顶聊天[K10008]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param is_flag: 是否置顶，0/1
|
||
"""
|
||
api_json = {'api': 'K10008', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'isFlag': is_flag}
|
||
return api_request(api_json)
|
||
|
||
|
||
def change_chat(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""切换当前聊天对象[K10009]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
"""
|
||
api_json = {'api': 'K10009', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_group_members_list(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""获取群成员列表[K10010]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
"""
|
||
api_json = {'api': 'K10010', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_wecom_group_members_list(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""获取企业群成员列表[K10011]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 企业群的id
|
||
"""
|
||
api_json = {'api': 'K10011', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def create_group(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""创建群聊[K10012]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: wxid，用英文逗号分隔
|
||
"""
|
||
api_json = {'api': 'K10012', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def update_group_name(robot_wxid: str, from_wxid: str, new_name: str) -> dict:
|
||
r"""修改群聊名称[K10013]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param new_name: 新的群名称
|
||
"""
|
||
api_json = {'api': 'K10013', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newName': new_name}
|
||
return api_request(api_json)
|
||
|
||
|
||
def publish_group_announcement(robot_wxid: str, from_wxid: str, new_msg: str) -> dict:
|
||
r"""发布群公告[K10014]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param new_msg: 公告内容
|
||
"""
|
||
api_json = {'api': 'K10014', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newMsg': new_msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def add_group_members_40_more(robot_wxid: str, from_wxid: str, new_wxid: str) -> dict:
|
||
r"""邀请群成员40人以上[K10015]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param new_wxid: 邀请进群的wxid
|
||
"""
|
||
api_json = {'api': 'K10015', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newWxid': new_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def add_group_members_40_less(robot_wxid: str, from_wxid: str, new_wxid: str) -> dict:
|
||
r"""添加群成员40人以下[K10016]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param new_wxid: 邀请进群的wxid
|
||
"""
|
||
api_json = {'api': 'K10016', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newWxid': new_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def drop_group_member(robot_wxid: str, from_wxid: str, old_wxid: str) -> dict:
|
||
r"""删除群成员[K10017]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param old_wxid: 群成员的wxid
|
||
"""
|
||
api_json = {'api': 'K10017', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'oldWxid': old_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def quit_group(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""退出群聊[K10018]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
"""
|
||
api_json = {'api': 'K10018', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_group_member_nickname(robot_wxid: str, from_wxid: str, old_wxid: str) -> dict:
|
||
r"""获取群内昵称[K10019]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param old_wxid: 群成员的wxid
|
||
"""
|
||
api_json = {'api': 'K10019', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'oldWxid': old_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def update_nickname_in_group(robot_wxid: str, from_wxid: str, new_msg: str) -> dict:
|
||
r"""修改群内昵称[K10020]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param new_msg: 新的群内昵称
|
||
"""
|
||
api_json = {'api': 'K10020', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newMsg': new_msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def hide_group_member_nickname(robot_wxid: str, from_wxid: str, is_flag: str) -> dict:
|
||
r"""显示或隐藏群成员昵称[K10021]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param is_flag: 是否隐藏，0/1
|
||
"""
|
||
api_json = {'api': 'K10021', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'isFlag': is_flag}
|
||
return api_request(api_json)
|
||
|
||
|
||
def save_to_address(robot_wxid: str, from_wxid: str, is_flag: str) -> dict:
|
||
r"""保存到通讯录[K10022]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
:param is_flag: 是否保存，0/1
|
||
"""
|
||
api_json = {'api': 'K10022', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'isFlag': is_flag}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_group_member_invite_information(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""获取群成员邀请信息[K10023]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 群聊的id
|
||
"""
|
||
api_json = {'api': 'K10023', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_address_list(robot_wxid: str) -> dict:
|
||
r"""获取通讯录列表[K10024]，此处获取的好友+群等(具体自己测试)
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
"""
|
||
api_json = {'api': 'K10024', 'robotWxid': robot_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_wecom_friend_list(robot_wxid: str) -> dict:
|
||
r"""获取企业好友列表[K10025]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
"""
|
||
api_json = {'api': 'K10025', 'robotWxid': robot_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_friend_info(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""获取指定好友信息[K10026]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
"""
|
||
api_json = {'api': 'K10026', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_net_friend_info(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""网络获取指定好友信息[K10027]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
"""
|
||
api_json = {'api': 'K10027', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def del_friend(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""删除好友[K10028]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
"""
|
||
api_json = {'api': 'K10028', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def update_friend_remark(robot_wxid: str, from_wxid: str, remark: str) -> dict:
|
||
r"""修改好友备注[K10029]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param remark: 备注
|
||
"""
|
||
api_json = {'api': 'K10029', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'newMsg': remark}
|
||
return api_request(api_json)
|
||
|
||
|
||
def agree_friend_request(robot_wxid: str, from_wxid: str, v3: str, v4: str, sence: str) -> dict:
|
||
r"""同意好友请求[K10030]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param v3: 接收到的消息里获取
|
||
:param v4: 接收到的消息里获取
|
||
:param sence: 接收到的消息里获取
|
||
"""
|
||
api_json = {'api': 'K10030', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'v3': v3, 'v4': v4, 'sence': sence}
|
||
return api_request(api_json)
|
||
|
||
|
||
def friend_status_detection(robot_wxid: str, from_wxid: str) -> dict:
|
||
r"""好友状态检测[K10031]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
"""
|
||
api_json = {'api': 'K10031', 'robotWxid': robot_wxid, 'fromWxid': from_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def add_blacklist(robot_wxid: str, from_wxid: str, is_flag: str) -> dict:
|
||
r"""拉黑好友[K10032]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param is_flag: 是否拉黑，0/1
|
||
"""
|
||
api_json = {'api': 'K10032', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'isFlag': is_flag}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_text_message(robot_wxid: str, from_wxid: str, msg: str) -> dict:
|
||
r"""发送文本信息[K10033]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg: 消息内容
|
||
"""
|
||
api_json = {'api': 'K10033', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msg': msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_image_message(robot_wxid: str, from_wxid: str, msg: str) -> dict:
|
||
r"""发送图片信息[K10034]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg: 图片路径，支持网址(直链，如：http://127.0.0.1/tp.jpg)
|
||
"""
|
||
api_json = {'api': 'K10034', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msg': msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_file_message(robot_wxid: str, from_wxid: str, msg: str) -> dict:
|
||
r"""发送文件信息[K10035]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg: 文件路径，本地(鲲鹏机器人所在电脑/服务器)
|
||
"""
|
||
api_json = {'api': 'K10035', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msg': msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_gif_message(robot_wxid: str, from_wxid: str, msg: str) -> dict:
|
||
r"""发送GIF信息[K10036]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg: GIF路径，支持网址(直链，如：http://127.0.0.1/tp.gif)
|
||
"""
|
||
api_json = {'api': 'K10036', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msg': msg}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_at_message(robot_wxid: str, from_wxid: str, msg: str, wx_list: str) -> dict:
|
||
r"""发送艾特信息[K10037]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg: 消息内容
|
||
:param wx_list: 艾特的wxid列表，用英文逗号隔开
|
||
"""
|
||
api_json = {'api': 'K10037', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msg': msg, 'list': wx_list}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_card_message(robot_wxid: str, from_wxid: str, tj_wxid: str, tj_name: str) -> dict:
|
||
r"""发送名片信息[K10038]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param tj_wxid: wxid和v4都可以
|
||
:param tj_name: 推荐昵称
|
||
"""
|
||
api_json = {'api': 'K10038', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'tjWxid': tj_wxid, 'tjName': tj_name}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_link_message(robot_wxid: str, from_wxid: str, title: str, desc: str, url: str, pic: str) -> dict:
|
||
r"""发送链接信息[K10039]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param title: 标题
|
||
:param desc: 描述
|
||
:param url: 链接
|
||
:param pic: 图片路径，支持网址(直链，如：http://127.0.0.1/tp.jpg)
|
||
"""
|
||
api_json = {'api': 'K10039', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'title': title, 'desc': desc,
|
||
'url': url, 'pic': pic}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_applet_message(robot_wxid: str, from_wxid: str, xml: str, pic: str) -> dict:
|
||
r"""发送小程序信息[K10040]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param xml: xml
|
||
:param pic: 略缩图路径，支持网址(直链，如：http://127.0.0.1/tp.jpg)
|
||
"""
|
||
api_json = {'api': 'K10040', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'xml': xml, 'pic': pic}
|
||
return api_request(api_json)
|
||
|
||
|
||
def relay_message(robot_wxid: str, from_wxid: str, msg_id: str) -> dict:
|
||
r"""转发信息[K10041]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg_id: 消息Id，SvrId
|
||
"""
|
||
api_json = {'api': 'K10041', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msgId': msg_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_login_status(robot_wxid: str) -> dict:
|
||
r"""取登录状态[K10042]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
"""
|
||
api_json = {'api': 'K10042', 'robotWxid': robot_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def withdraw_message(robot_wxid: str, from_wxid: str, msg_id: str) -> dict:
|
||
r"""撤回信息[K10043]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param msg_id: 消息Id，SvrId
|
||
"""
|
||
api_json = {'api': 'K10043', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'msgId': msg_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_personal_info(robot_wxid: str) -> dict:
|
||
r"""获取个人信息[K10044]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
"""
|
||
api_json = {'api': 'K10044', 'robotWxid': robot_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_login_list() -> dict:
|
||
r"""取登录账号列表[K10045]"""
|
||
api_json = {'api': 'K10045'}
|
||
return api_request(api_json)
|
||
|
||
|
||
def download_image(robot_wxid: str, xml: str, path: str) -> dict:
|
||
r"""下载图片[K10046]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param xml: 接收到的消息xml
|
||
:param path: 保存路径，绝对路径，带后缀
|
||
"""
|
||
api_json = {'api': 'K10046', 'robotWxid': robot_wxid, 'xml': xml, 'path': path}
|
||
return api_request(api_json)
|
||
|
||
|
||
def decrypt_image(robot_wxid: str, filepath: str, path: str) -> dict:
|
||
r"""解密图片[K10047]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param filepath: 接收消息里的filepath字段
|
||
:param path: 保存路径，绝对路径，带后缀
|
||
"""
|
||
api_json = {'api': 'K10047', 'robotWxid': robot_wxid, 'filepath': filepath, 'path': path}
|
||
return api_request(api_json)
|
||
|
||
|
||
def send_xml_message(robot_wxid: str, from_wxid: str, xml: str) -> dict:
|
||
r"""发送XML信息[K10048]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param xml: XML信息
|
||
"""
|
||
api_json = {'api': 'K10048', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'xml': xml}
|
||
return api_request(api_json)
|
||
|
||
|
||
def refresh_moments(robot_wxid: str) -> dict:
|
||
r"""刷新朋友圈[K10049]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
"""
|
||
api_json = {'api': 'K10049', 'robotWxid': robot_wxid}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_moments_next_page(robot_wxid: str, circle_id: str) -> dict:
|
||
r"""获取朋友圈下一页[K10050]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param circle_id: itemid朋友圈ID，获取从该ID的下一条开始的一页
|
||
"""
|
||
api_json = {'api': 'K10050', 'robotWxid': robot_wxid, 'circleId': circle_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def get_personal_moments(robot_wxid: str, from_wxid: str, circle_id: str) -> dict:
|
||
r"""获取个人朋友圈[K10051]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param from_wxid: 目标的wxid
|
||
:param circle_id: itemid朋友圈ID，获取从该ID的下一条开始的一页
|
||
"""
|
||
api_json = {'api': 'K10051', 'robotWxid': robot_wxid, 'fromWxid': from_wxid, 'circleId': circle_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def moments_like(robot_wxid: str, circle_id: str) -> dict:
|
||
r"""朋友圈点赞[K10052]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param circle_id: itemid朋友圈ID，获取从该ID的下一条开始的一页
|
||
"""
|
||
api_json = {'api': 'K10052', 'robotWxid': robot_wxid, 'circleId': circle_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def moments_cancel_like(robot_wxid: str, circle_id: str) -> dict:
|
||
r"""朋友圈取消点赞[K10053]
|
||
|
||
:param robot_wxid: 机器人的wxid
|
||
:param circle_id: itemid朋友圈ID，获取从该ID的下一条开始的一页
|
||
"""
|
||
api_json = {'api': 'K10053', 'robotWxid': robot_wxid, 'circleId': circle_id}
|
||
return api_request(api_json)
|
||
|
||
|
||
def local_download_image(xml, path):
|
||
r"""本地下载图片
|
||
|
||
:param xml: 接收到的消息xml
|
||
:param path: 保存文件夹的绝对路径
|
||
:return: 下载图片的绝对路径
|
||
"""
|
||
emoji_msg = ET.fromstring(xml).find('emoji')
|
||
url = emoji_msg.get('cdnurl').replace('amp;', '')
|
||
r = requests.get(url)
|
||
if os.path.exists(path) is False:
|
||
os.mkdir(path)
|
||
xo, j = image_format(r.content) # 判断图片格式 并计算返回异或值 函数
|
||
if j == 1:
|
||
mat = '.png'
|
||
elif j == 2:
|
||
mat = '.gif'
|
||
else:
|
||
mat = '.jpg'
|
||
file_name = re.findall(r'm=(.*?)&', url)[0]
|
||
out = os.path.join(path, file_name + mat)
|
||
with open(out, 'wb') as f:
|
||
f.write(r.content)
|
||
f.close()
|
||
return out
|
||
|
||
|
||
def local_decrypt_image(filepath: str, path: str) -> str:
|
||
r"""本地解密dat图片
|
||
|
||
:param filepath: 接收消息里的filepath字段
|
||
:param path: 保存文件夹的绝对路径
|
||
:return: 解密图片的绝对路径
|
||
"""
|
||
file_splitext = os.path.splitext(filepath)
|
||
file_type = file_splitext[1]
|
||
file_name = os.path.split(file_splitext[0])[1]
|
||
if file_type != '.dat':
|
||
return '不支持处理非.dat文件'
|
||
if '.' in path:
|
||
return 'path参数为保存文件夹的绝对路径'
|
||
if os.path.exists(path) is False:
|
||
os.mkdir(path)
|
||
for i in range(30):
|
||
if os.path.exists(filepath) is True:
|
||
break
|
||
time.sleep(2)
|
||
if os.path.exists(filepath) is False:
|
||
return '微信下载dat图片超时'
|
||
|
||
dat_read = open(filepath, "rb") # 读取.dat 文件
|
||
xo, j = image_format(filepath) # 判断图片格式 并计算返回异或值 函数
|
||
if j == 1:
|
||
mat = '.png'
|
||
elif j == 2:
|
||
mat = '.gif'
|
||
else:
|
||
mat = '.jpg'
|
||
out = os.path.join(path, file_name + mat)
|
||
png_write = open(out, "wb") # 图片写入
|
||
dat_read.seek(0) # 重置文件指针位置
|
||
for now in dat_read: # 循环字节
|
||
for nowByte in now:
|
||
new_byte = nowByte ^ xo # 转码计算
|
||
png_write.write(bytes([new_byte])) # 转码后重新写
|
||
dat_read.close()
|
||
png_write.close()
|
||
return out
|
||
|
||
|
||
def image_format(f):
|
||
"""
|
||
计算异或值
|
||
各图片头部信息
|
||
png：89 50 4e 47
|
||
gif： 47 49 46 38
|
||
jpeg：ff d8 ff
|
||
"""
|
||
if isinstance(f, str):
|
||
dat_r = open(f, "rb")
|
||
elif isinstance(f, bytes):
|
||
dat_r = BytesIO(f)
|
||
else:
|
||
return print('不支持的类型', type(f))
|
||
try:
|
||
a = [(0x89, 0x50, 0x4e), (0x47, 0x49, 0x46), (0xff, 0xd8, 0xff)]
|
||
for now in dat_r:
|
||
j = 0
|
||
for xor in a:
|
||
j = j + 1 # 记录是第几个格式 1：png 2：gif 3：jpeg
|
||
i = 0
|
||
res = []
|
||
now2 = now[:3] # 取前三组判断
|
||
for nowByte in now2:
|
||
res.append(nowByte ^ xor[i])
|
||
i += 1
|
||
if res[0] == res[1] == res[2]:
|
||
return res[0], j
|
||
except:
|
||
pass
|
||
finally:
|
||
dat_r.close()
|
||
|
||
|
||
# 表情代码
|
||
WC_EMOTICON_CONVERSION = {
|
||
'[微笑]': '😃', '[Smile]': '😃',
|
||
'[撇嘴]': '😖', '[Grimace]': '😖',
|
||
'[色]': '😍', '[Drool]': '😍',
|
||
'[发呆]': '😳', '[Scowl]': '😳',
|
||
'[得意]': '😎', '[Chill]': '😎',
|
||
'[流泪]': '😭', '[Sob]': '😭',
|
||
'[害羞]': '☺️', '[Shy]': '☺️', '[Blush]': '☺️',
|
||
'[闭嘴]': '🤐', '[Shutup]': '🤐',
|
||
'[睡]': '😴', '[Sleep]': '😴',
|
||
'[大哭]': '😣', '[Cry]': '😣',
|
||
'[尴尬]': '😰', '[Awkward]': '😰',
|
||
'[发怒]': '😡', '[Pout]': '😡',
|
||
'[调皮]': '😜', '[Wink]': '😜',
|
||
'[呲牙]': '😁', '[Grin]': '😁',
|
||
'[惊讶]': '😱', '[Surprised]': '😱',
|
||
'[难过]': '🙁', '[Frown]': '🙁',
|
||
'[囧]': '☺️', '[Tension]': '☺️',
|
||
'[抓狂]': '😫', '[Scream]': '😫',
|
||
'[吐]': '🤢', '[Puke]': '🤢',
|
||
'[偷笑]': '🙈', '[Chuckle]': '🙈',
|
||
'[愉快]': '☺️', '[Joyful]': '☺️',
|
||
'[白眼]': '🙄', '[Slight]': '🙄',
|
||
'[傲慢]': '😕', '[Smug]': '😕',
|
||
'[困]': '😪', '[Drowsy]': '😪',
|
||
'[惊恐]': '😱', '[Panic]': '😱',
|
||
'[流汗]': '😓', '[Sweat]': '😓',
|
||
'[憨笑]': '😄', '[Laugh]': '😄',
|
||
'[悠闲]': '😏', '[Loafer]': '😏',
|
||
'[奋斗]': '💪', '[Strive]': '💪',
|
||
'[咒骂]': '😤', '[Scold]': '😤',
|
||
'[疑问]': '❓', '[Doubt]': '❓',
|
||
'[嘘]': '🤐', '[Shhh]': '🤐',
|
||
'[晕]': '😲', '[Dizzy]': '😲',
|
||
'[衰]': '😳', '[BadLuck]': '😳',
|
||
'[骷髅]': '💀', '[Skull]': '💀',
|
||
'[敲打]': '👊', '[Hammer]': '👊',
|
||
'[再见]': '🙋\u200d♂', '[Bye]': '🙋\u200d♂',
|
||
'[擦汗]': '😥', '[Relief]': '😥',
|
||
'[抠鼻]': '🤷\u200d♂', '[DigNose]': '🤷\u200d♂',
|
||
'[鼓掌]': '👏', '[Clap]': '👏',
|
||
'[坏笑]': '👻', '[Trick]': '👻',
|
||
'[左哼哼]': '😾', '[Bah！L]': '😾',
|
||
'[右哼哼]': '😾', '[Bah！R]': '😾',
|
||
'[哈欠]': '😪', '[Yawn]': '😪',
|
||
'[鄙视]': '😒', '[Lookdown]': '😒',
|
||
'[委屈]': '😣', '[Wronged]': '😣',
|
||
'[快哭了]': '😔', '[Puling]': '😔',
|
||
'[阴险]': '😈', '[Sly]': '😈',
|
||
'[亲亲]': '😘', '[Kiss]': '😘',
|
||
'[可怜]': '😻', '[Whimper]': '😻',
|
||
'[菜刀]': '🔪', '[Cleaver]': '🔪',
|
||
'[西瓜]': '🍉', '[Melon]': '🍉',
|
||
'[啤酒]': '🍺', '[Beer]': '🍺',
|
||
'[咖啡]': '☕', '[Coffee]': '☕',
|
||
'[猪头]': '🐷', '[Pig]': '🐷',
|
||
'[玫瑰]': '🌹', '[Rose]': '🌹',
|
||
'[凋谢]': '🥀', '[Wilt]': '🥀',
|
||
'[嘴唇]': '💋', '[Lip]': '💋',
|
||
'[爱心]': '❤️', '[Heart]': '❤️',
|
||
'[心碎]': '💔', '[BrokenHeart]': '💔',
|
||
'[蛋糕]': '🎂', '[Cake]': '🎂',
|
||
'[炸弹]': '💣', '[Bomb]': '💣',
|
||
'[便便]': '💩', '[Poop]': '💩',
|
||
'[月亮]': '🌃', '[Moon]': '🌃',
|
||
'[太阳]': '🌞', '[Sun]': '🌞',
|
||
'[拥抱]': '🤗', '[Hug]': '🤗',
|
||
'[强]': '👍', '[Strong]': '👍',
|
||
'[弱]': '👎', '[Weak]': '👎',
|
||
'[握手]': '🤝', '[Shake]': '🤝',
|
||
'[胜利]': '✌️', '[Victory]': '✌️',
|
||
'[抱拳]': '🙏', '[Salute]': '🙏',
|
||
'[勾引]': '💁\u200d♂', '[Beckon]': '💁\u200d♂',
|
||
'[拳头]': '👊', '[Fist]': '👊',
|
||
'[OK]': '👌',
|
||
'[跳跳]': '💃', '[Waddle]': '💃',
|
||
'[发抖]': '🙇', '[Tremble]': '🙇',
|
||
'[怄火]': '😡', '[Aaagh!]': '😡',
|
||
'[转圈]': '🕺', '[Twirl]': '🕺',
|
||
'[嘿哈]': '🤣', '[Hey]': '🤣',
|
||
'[捂脸]': '🤦\u200d♂', '[Facepalm]': '🤦\u200d♂',
|
||
'[奸笑]': '😜', '[Smirk]': '😜',
|
||
'[机智]': '🤓', '[Smart]': '🤓',
|
||
'[皱眉]': '😟', '[Concerned]': '😟',
|
||
'[耶]': '✌️', '[Yeah!]': '✌️',
|
||
'[红包]': '🧧', '[Packet]': '🧧',
|
||
'[鸡]': '🐥', '[Chick]': '🐥',
|
||
'[蜡烛]': '🕯️', '[Candle]': '🕯️',
|
||
'[糗大了]': '😥',
|
||
'[Thumbs Up]': '👍', '[Pleased]': '😊',
|
||
'[Rich]': '🀅',
|
||
'[Pup]': '🐶',
|
||
'[吃瓜]': '🙄\u200d🍉', '[Onlooker]': '🙄\u200d🍉',
|
||
'[加油]': '💪\u200d😁',
|
||
'[加油加油]': '💪\u200d😷',
|
||
'[汗]': '😓',
|
||
'[天啊]': '😱',
|
||
'[Emm]': '🤔',
|
||
'[社会社会]': '😏',
|
||
'[旺柴]': '🐶\u200d😏', '[Awesome]': '🐶\u200d😏',
|
||
'[好的]': '😏\u200d👌',
|
||
'[哇]': '🤩',
|
||
'[打脸]': '😟\u200d🤚',
|
||
'[破涕为笑]': '😂', '[破涕為笑]': '😂', '[Lol]': '😂',
|
||
'[苦涩]': '😭',
|
||
'[翻白眼]': '🙄',
|
||
'[爆竹]': '🧨',
|
||
'[烟花]': '🎆',
|
||
'[裂开]': '[裂开]',
|
||
'[福]': '🧧',
|
||
'[發]': '🀅',
|
||
'[礼物]': '🎁',
|
||
'[庆祝]': '🎉',
|
||
'[合十]': '🙏',
|
||
'[叹气]': '😮💨', '[Sigh]': '😮💨',
|
||
'[让我看看]': '👀',
|
||
'[666]': '6️⃣6️⃣6️⃣',
|
||
'[无语]': '😑',
|
||
'[失望]': '😞',
|
||
'[恐惧]': '😨',
|
||
'[脸红]': '😳',
|
||
'[生病]': '😷',
|
||
'[笑脸]': '😁',
|
||
}
|
||
|
||
|
||
def unicode_escape(chars, data_dict):
|
||
"""emoji 替换 unicode 格式 "[\\u1f602]，请勿主动调用"""
|
||
if chars in WC_EMOTICON_CONVERSION.values():
|
||
return [k for k, v in WC_EMOTICON_CONVERSION.items() if v == chars][0]
|
||
else:
|
||
chars_str = chars.encode('unicode-escape').decode().replace('U000', 'u')
|
||
chars_unicode_list = chars_str.split('\\')
|
||
chars_unicode_list.pop(0)
|
||
for i in range(len(chars_unicode_list)):
|
||
chars_unicode_list[i] = '[\\\\{}]'.format(chars_unicode_list[i])
|
||
return ''.join(chars_unicode_list)
|
||
|
||
|
||
def emoji_to_unicode(message: str) -> str:
|
||
"""将msg中的emoji表情转为鲲鹏 unicode 格式 "[\\u1f602]"
|
||
"""
|
||
return emoji.replace_emoji(message, replace=unicode_escape)
|
||
|
||
|
||
def unicode_to_emoji(message: Union[str, dict, list], msg_type='str') -> Union[str, dict, list]:
|
||
"""将msg中的鲲鹏unicode格式 "[\\u1f602]" 转为emoji表情
|
||
|
||
:param message: 建议传入json字符串，虽然提供了dict和list的解析方法，但递归遍历影响运行速度
|
||
:param msg_type: 默认为‘str'；json字符串请传入’json‘，区别在于json的转义符多；dict和list通常情况下保持默认即可
|
||
"""
|
||
if isinstance(message, str):
|
||
return de_emoji(message, msg_type)
|
||
elif isinstance(message, dict) or isinstance(message, list):
|
||
return ergodic(message)
|
||
else:
|
||
return '暂不支持此类型'
|
||
|
||
|
||
def ergodic(message: Union[dict, list]) -> Union[dict, list]:
|
||
"""递归遍历，将鲲鹏unicode格式 "[\\u1f602]" 转为emoji表情，不建议使用此方法，且请勿主动调用"""
|
||
if isinstance(message, dict):
|
||
for key in message.keys():
|
||
message[key] = de_emoji(message[key])
|
||
return message
|
||
elif isinstance(message, list):
|
||
for i in range(len(message)):
|
||
message[i] = de_emoji(message[i])
|
||
return message
|
||
print(message)
|
||
return message
|
||
|
||
|
||
def de_emoji(message, msg_type='str'):
|
||
"""将鲲鹏unicode格式 "[\\u1f602]" 转为emoji表情，请勿主动调用"""
|
||
if not message:
|
||
return message
|
||
if msg_type == 'json':
|
||
zyf = '\\'
|
||
else:
|
||
zyf = ''
|
||
if isinstance(message, dict) or isinstance(message, list):
|
||
return ergodic(message)
|
||
elif isinstance(message, str):
|
||
emoji_list = re.findall(zyf + zyf + r'\\u[0-9a-fA-F]{5}', message)
|
||
# print(emoji_list)
|
||
for emoji_unicode in emoji_list:
|
||
chars = str(str.replace(emoji_unicode, zyf + '\\u', '\\U000')).encode('ASCII').decode('unicode-escape')
|
||
# print(chars)
|
||
message = message.replace(f'[{emoji_unicode}]', chars)
|
||
message = message.replace('[' + zyf + '\\u200d]', '\\u200d'.encode('ASCII').decode('unicode-escape'))
|
||
message = message.replace('[' + zyf + '\\ufe0f]', '\\ufe0f'.encode('ASCII').decode('unicode-escape'))
|
||
chars_list = re.findall(r'\[\w+]', message)
|
||
for chars in chars_list:
|
||
try:
|
||
message = message.replace(chars, WC_EMOTICON_CONVERSION[chars])
|
||
except KeyError:
|
||
pass
|
||
unicode_text_list = re.findall(zyf + zyf + r'\\u[0-9a-fA-F]{4}', message)
|
||
for unicode_text in unicode_text_list:
|
||
message = message.replace(f'[{unicode_text}]', unicode_text.replace(zyf, '', 1))
|
||
return message
|
||
else:
|
||
return message
|
||
|
||
|
||
def crypt_md5(message: Union[dict, str]) -> str:
|
||
"""鉴权算法"""
|
||
if isinstance(message, str):
|
||
message = json.loads(message)
|
||
crypt_text = f"{config.SECRET_KEY}{''.join(message.values())}{config.SECRET_KEY}"
|
||
jm = hashlib.md5()
|
||
jm.update(crypt_text.encode("utf8"))
|
||
crypt_text_2 = jm.hexdigest()
|
||
jm2 = hashlib.md5()
|
||
jm2.update(crypt_text_2.encode("utf8"))
|
||
return jm2.hexdigest()
|
||
|
||
|
||
def api_request(api_json: dict) -> dict:
|
||
r"""请求api接口，请勿主动调用
|
||
|
||
:param api_json: 请求参数
|
||
"""
|
||
|
||
if api_json.get('api') != 'K10002': # 添加日志不转emoji
|
||
api_json = emoji_to_unicode(json.dumps(api_json, ensure_ascii=False))
|
||
|
||
if isinstance(api_json, str): # str解析json
|
||
api_json = json.loads(api_json)
|
||
|
||
api_json.update({'sign': crypt_md5(api_json)}) # 增加鉴权
|
||
|
||
r = requests.post(config.wxbot_url, headers={'content-type': 'application/json'},
|
||
data=json.dumps(api_json, ensure_ascii=False).encode('utf-8')).json()
|
||
|
||
return r
|