kboss/b/capital/ecs_describe_security_groups.dspy
2025-07-16 14:27:17 +08:00

41 lines
1.8 KiB
Plaintext

async def ecs_describe_security_groups(ns={}):
import aiohttp
NETWORK_URL = 'http://api.capitalonline.net/sg/v1'
action = 'DescribeSecurityGroups'
method = "GET"
param = {}
url = get_signature(action, method, NETWORK_URL, param=param)
body = {
'RegionCode': ns.get('RegionCode'),
'SecurityGroupId': ns.get('SecurityGroupId'),
'SecurityGroupName': ns.get('SecurityGroupName'),
'PageNumber': int(ns.get('PageNumber')) if ns.get('PageNumber') else 0,
'PageSize': int(ns.get('PageSize')) if ns.get('PageSize') else 20,
}
async with aiohttp.ClientSession() as session:
async with session.request(method, url, json=body) as response:
resp = await response.text()
result = json.loads(resp)
if result['Code'] == 'Success':
result['status'] = True
result.pop('Code')
result['data'] = result.pop('Data')
if result['data']['TotalCount'] > 0:
s_list = []
for s_detail in result['data']['SecurityGroupSet']:
s_split_res = s_detail['SecurityGroupName'].split('0_____0')
s_name = s_split_res[0]
if ns['orgid'] == s_name:
s_detail['SecurityGroupName'] = s_split_res[1]
s_list.append(s_detail)
result['data']['SecurityGroupSet'] = s_list
result['data']['TotalCount'] = len(s_list)
result['msg'] = result.pop('Message')
else:
result['status'] = False
result['msg'] = result.pop('Message')
return result
ret = await ecs_describe_security_groups(params_kw)
return ret