Skip to content

Commit

Permalink
完全支持walle-q剩余接口
Browse files Browse the repository at this point in the history 
+ 群踢人
+ 群禁言
+ 群管设置
+ 群名设置
+ 退群
+ 陌生人信息
+ 好友列表
+ 群成员信息
+ 群成员列表
+ 群信息
+ 群列表
  • Loading branch information
@lunzhiPenxil
lunzhiPenxil committed Mar 25, 2023
1 parent d3858ec commit 98c64ac
Show file tree
Hide file tree
Showing 2 changed files with 223 additions and 12 deletions.
32 changes: 21 additions & 11 deletions OlivOS/API.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -930,7 +930,8 @@ def send_like(self, user_id: 'str|int', times: int = 1, flag_log: bool = True, r
def __set_group_kick(self, group_id, user_id, host_id, rehect_add_request, flag_log=True):
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
OlivOS.onebotV12SDK.event_action.set_group_kick(self, group_id, user_id)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
OlivOS.onebotSDK.event_action.set_group_kick(self, group_id, user_id, rehect_add_request)
Expand All @@ -948,7 +949,8 @@ def set_group_kick(self, group_id: 'str|int', user_id: 'str|int', host_id: 'str|
def __set_group_ban(self, group_id, user_id, host_id, duration, flag_log=True):
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
OlivOS.onebotV12SDK.event_action.set_group_ban(self, group_id, user_id, duration)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
OlivOS.onebotSDK.event_action.set_group_ban(self, group_id, user_id, duration)
Expand Down Expand Up @@ -1004,7 +1006,8 @@ def set_group_whole_ban(self, group_id: 'str|int', enable: bool, host_id: 'str|i
def __set_group_admin(self, group_id, user_id, enable, host_id, flag_log=True):
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
OlivOS.onebotV12SDK.event_action.set_group_admin(self, group_id, user_id, enable)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
OlivOS.onebotSDK.event_action.set_group_admin(self, group_id, user_id, enable)
Expand Down Expand Up @@ -1058,7 +1061,8 @@ def set_group_card(self, group_id: 'str|int', user_id: 'str|int', card, host_id:
def __set_group_name(self, group_id, group_name, host_id, flag_log=True):
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
OlivOS.onebotV12SDK.event_action.set_group_name(self, group_id, group_name)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
OlivOS.onebotSDK.event_action.set_group_name(self, group_id, group_name)
Expand All @@ -1076,7 +1080,8 @@ def set_group_name(self, group_id: 'str|int', group_name: str, host_id: 'str|int
def __set_group_leave(self, group_id, host_id, is_dismiss, flag_log=True):
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
OlivOS.onebotV12SDK.event_action.set_group_leave(self, group_id)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
OlivOS.onebotSDK.event_action.set_group_leave(self, group_id, is_dismiss)
Expand Down Expand Up @@ -1200,7 +1205,7 @@ def __get_stranger_info(self, user_id, no_cache, flag_log=True):
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
res_data = OlivOS.onebotV12SDK.event_action.get_stranger_info(self, user_id)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
res_data = OlivOS.onebotSDK.event_action.get_stranger_info(self, user_id, no_cache)
elif self.platform['sdk'] == 'telegram_poll':
Expand All @@ -1223,7 +1228,7 @@ def __get_friend_list(self, flag_log=True):
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
res_data = OlivOS.onebotV12SDK.event_action.get_friend_list(self)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
res_data = OlivOS.onebotSDK.event_action.get_friend_list(self)
elif self.platform['sdk'] == 'telegram_poll':
Expand All @@ -1243,7 +1248,8 @@ def __get_group_info(self, group_id, host_id, no_cache, flag_log=True):
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
res_data = OlivOS.onebotV12SDK.event_action.get_group_info(self, group_id)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
res_data = OlivOS.onebotSDK.event_action.get_group_info(self, group_id, no_cache)
Expand All @@ -1265,7 +1271,7 @@ def __get_group_list(self, flag_log=True):
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
res_data = OlivOS.onebotV12SDK.event_action.get_group_list(self)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
res_data = OlivOS.onebotSDK.event_action.get_group_list(self)
elif self.platform['sdk'] == 'telegram_poll':
Expand All @@ -1285,7 +1291,10 @@ def __get_group_member_info(self, group_id, user_id, host_id, no_cache, flag_log
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
res_data = OlivOS.onebotV12SDK.event_action.get_group_member_info(self, group_id, user_id)
else:
pass
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
res_data = OlivOS.onebotSDK.event_action.get_group_member_info(self, group_id, user_id, no_cache)
Expand All @@ -1311,7 +1320,8 @@ def __get_group_member_list(self, group_id, host_id, flag_log=True):
res_data = None
if self.platform['sdk'] == 'onebot':
if self.platform['model'] in OlivOS.onebotV12LinkServerAPI.gCheckList:
pass
if host_id is None:
res_data = OlivOS.onebotV12SDK.event_action.get_group_member_list(self, group_id)
elif self.platform['model'] in OlivOS.flaskServerAPI.gCheckList:
if host_id is None:
res_data = OlivOS.onebotSDK.event_action.get_group_member_list(self, group_id)
Expand Down
203 changes: 202 additions & 1 deletion OlivOS/onebotV12SDK.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -334,7 +334,7 @@ def get_Event_from_SDK(target_event:OlivOS.API.Event):
target_event.active = True
target_event.plugin_info['func_type'] = 'heartbeat'
target_event.data = target_event.heartbeat(
target_event.sdk_event.json['interval']
int(target_event.sdk_event.json['interval']) * 1000
)
elif 'detail_type' in target_event.sdk_event.json \
and 'connect' == target_event.sdk_event.json['detail_type']:
Expand Down Expand Up @@ -730,6 +730,63 @@ def delete_msg(target_event, message_id):
this_msg.params.message_id = str(message_id)
this_msg.do_api(bot_hash, control_queue)

def set_group_kick(target_event, group_id, user_id):
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.kick_group_member()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
this_msg.do_api(bot_hash, control_queue)

def set_group_leave(target_event, group_id):
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.leave_group()
this_msg.params.group_id = str(group_id)
this_msg.do_api(bot_hash, control_queue)

def set_group_name(target_event, group_id, group_name):
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.set_group_name()
this_msg.params.group_id = str(group_id)
this_msg.params.group_name = str(group_name)
this_msg.do_api(bot_hash, control_queue)

def set_group_ban(target_event, group_id, user_id, duration=1800):
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
if 0 == duration:
this_msg = api.ban_group_member()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
this_msg.params.duration = str(duration)
this_msg.do_api(bot_hash, control_queue)
else:
this_msg = api.unban_group_member()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
this_msg.do_api(bot_hash, control_queue)

def set_group_admin(target_event, group_id, user_id, enable):
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
if True is enable:
this_msg = api.set_group_admin()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
this_msg.do_api(bot_hash, control_queue)
else:
this_msg = api.unset_group_admin()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
this_msg.do_api(bot_hash, control_queue)

def get_login_info(target_event:OlivOS.API.Event):
res_data = OlivOS.contentAPI.api_result_data_template.get_group_info()
raw_obj = None
Expand All @@ -748,6 +805,150 @@ def get_login_info(target_event:OlivOS.API.Event):
res_data['data']['id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
return res_data

def get_stranger_info(target_event:OlivOS.API.Event, user_id):
res_data = OlivOS.contentAPI.api_result_data_template.get_stranger_info()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_user_info()
this_msg.params.user_id = str(user_id)
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) is dict:
res_data['active'] = True
res_data['data']['name'] = init_api_do_mapping_for_dict(raw_obj, ['user_name'], str)
res_data['data']['id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
return res_data

def get_friend_list(target_event:OlivOS.API.Event):
res_data = OlivOS.contentAPI.api_result_data_template.get_friend_list()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_friend_list()
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) == list:
res_data['active'] = True
for raw_obj_this in raw_obj:
tmp_res_data_this = OlivOS.contentAPI.api_result_data_template.get_user_info_strip()
tmp_res_data_this['name'] = init_api_do_mapping_for_dict(raw_obj_this, ['user_name'], str)
tmp_res_data_this['id'] = init_api_do_mapping_for_dict(raw_obj_this, ['user_id'], str)
res_data['data'].append(tmp_res_data_this)
return res_data

def get_group_info(target_event:OlivOS.API.Event, group_id):
res_data = OlivOS.contentAPI.api_result_data_template.get_group_info()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_group_info()
this_msg.params.group_id = str(group_id)
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) is dict:
res_data['active'] = True
res_data['data']['name'] = init_api_do_mapping_for_dict(raw_obj, ['group_name'], str)
res_data['data']['id'] = init_api_do_mapping_for_dict(raw_obj, ['group_id'], str)
res_data['data']['memo'] = ''
res_data['data']['member_count'] = 0
res_data['data']['max_member_count'] = 0
return res_data

def get_group_list(target_event:OlivOS.API.Event):
res_data = OlivOS.contentAPI.api_result_data_template.get_group_list()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_group_list()
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) == list:
res_data['active'] = True
for raw_obj_this in raw_obj:
tmp_res_data_this = OlivOS.contentAPI.api_result_data_template.get_user_info_strip()
tmp_res_data_this['name'] = init_api_do_mapping_for_dict(raw_obj_this, ['group_name'], str)
tmp_res_data_this['id'] = init_api_do_mapping_for_dict(raw_obj_this, ['group_id'], str)
tmp_res_data_this['memo'] = ''
tmp_res_data_this['member_count'] = 0
tmp_res_data_this['max_member_count'] = 0
res_data['data'].append(tmp_res_data_this)
return res_data

def get_group_member_info(target_event:OlivOS.API.Event, group_id, user_id):
res_data = OlivOS.contentAPI.api_result_data_template.get_group_member_info()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_group_member_info()
this_msg.params.group_id = str(group_id)
this_msg.params.user_id = str(user_id)
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) is dict:
res_data['active'] = True
res_data['data']['name'] = init_api_do_mapping_for_dict(raw_obj, ['user_name'], str)
res_data['data']['id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
res_data['data']['user_id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
res_data['data']['group_id'] = this_msg.params.group_id
res_data['data']['times']['join_time'] = 0
res_data['data']['times']['last_sent_time'] = 0
res_data['data']['times']['shut_up_timestamp'] = 0
res_data['data']['role'] = 'member'
res_data['data']['card'] = init_api_do_mapping_for_dict(raw_obj, ['user_remark'], str)
res_data['data']['title'] = init_api_do_mapping_for_dict(raw_obj, ['user_displayname'], str)
return res_data

def get_group_member_list(target_event:OlivOS.API.Event, group_id):
res_data = OlivOS.contentAPI.api_result_data_template.get_group_member_list()
raw_obj = None
if target_event.bot_info != None:
bot_hash = target_event.bot_info.hash
control_queue = target_event.plugin_info['control_queue']
this_msg = api.get_group_member_list()
this_msg.params.group_id = str(group_id)
waitForResReady(this_msg.echo)
this_msg.do_api(bot_hash, control_queue)
res_raw = waitForRes(this_msg.echo)
raw_obj = init_api_json(res_raw)
if raw_obj is not None:
if type(raw_obj) == list:
res_data['active'] = True
for raw_obj_this in raw_obj:
tmp_res_data_this = OlivOS.contentAPI.api_result_data_template.get_user_info_strip()
tmp_res_data_this['data']['name'] = init_api_do_mapping_for_dict(raw_obj, ['user_name'], str)
tmp_res_data_this['data']['id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
tmp_res_data_this['data']['user_id'] = init_api_do_mapping_for_dict(raw_obj, ['user_id'], str)
tmp_res_data_this['data']['group_id'] = this_msg.params.group_id
tmp_res_data_this['data']['times']['join_time'] = 0
tmp_res_data_this['data']['times']['last_sent_time'] = 0
tmp_res_data_this['data']['times']['shut_up_timestamp'] = 0
tmp_res_data_this['data']['role'] = 'member'
tmp_res_data_this['data']['card'] = init_api_do_mapping_for_dict(raw_obj, ['user_remark'], str)
tmp_res_data_this['data']['title'] = init_api_do_mapping_for_dict(raw_obj, ['user_displayname'], str)
res_data['data'].append(tmp_res_data_this)
return res_data

def set_friend_add_request(target_event, flag, approve:bool, remark):
global gTmpReg
if target_event.bot_info != None:
Expand Down

0 comments on commit 98c64ac

Please sign in to comment.