Huawei Cloud Computing Query

huaweicloudsdkcore>=3.1.0
huaweicloudsdkecs>=3.1.0
huaweicloudsdkiam>=3.1.0
huaweicloudsdkims>=3.1.0
huaweicloudsdkbms>=3.1.0
huaweicloudsdkas>=3.1.0

"""AS 模块公共工具函数"""
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from config import load_credentials, build_http_config
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkiam.v3 import IamClient
from huaweicloudsdkiam.v3.model import KeystoneListProjectsRequest
from huaweicloudsdkiam.v3.region.iam_region import IamRegion


def resolve_project_id(region, project_id=None):
    """解析项目 ID

    优先使用传入的 project_id，否则从环境变量 HW_PROJECT_ID 获取，
    最后通过 IAM KeystoneListProjects 接口按 region 自动获取。

    :param region: 区域名称，如 cn-north-4
    :param project_id: 可选的项目 ID，若提供则直接返回
    :return: 项目 ID 字符串
    """
    if project_id:
        return project_id

    env_pid = os.getenv("HW_PROJECT_ID", "")
    if env_pid:
        return env_pid

    ak = os.getenv("HW_ACCESS_KEY", "")
    sk = os.getenv("HW_SECRET_KEY", "")
    security_token = os.getenv("HW_SECURITY_TOKEN", "")

    if not ak or not sk:
        print("未配置 AK/SK，请设置环境变量 HW_ACCESS_KEY 和 HW_SECRET_KEY")
        exit(-1)

    try:
        http_config = build_http_config()
        credentials = BasicCredentials(ak, sk)
        if security_token:
            credentials = credentials.with_security_token(security_token)
        client = (IamClient.new_builder()
                  .with_http_config(http_config)
                  .with_credentials(credentials)
                  .with_region(IamRegion.value_of(region))
                  .build())
        request = KeystoneListProjectsRequest()
        request.name = region
        response = client.keystone_list_projects(request)
        projects = response.projects
        if not projects:
            print(f"未找到可访问的项目 (区域: {region})")
            exit(0)
        for project in projects:
            if getattr(project, 'name', '') == region:
                return project.id
        return projects[0].id
    except Exception as e:
        print(f"自动获取项目 ID 失败: {e}")
        print("请通过 --project_id 参数指定，或设置 HW_PROJECT_ID 环境变量")
        exit(-1)


def _build_vpc_client(region, project_id):
    """构建 VPC 客户端"""
    try:
        from huaweicloudsdkvpc.v3 import VpcClient
        from huaweicloudsdkvpc.v3.region.vpc_region import VpcRegion
    except ImportError:
        return None

    ak = os.getenv("HW_ACCESS_KEY", "")
    sk = os.getenv("HW_SECRET_KEY", "")
    security_token = os.getenv("HW_SECURITY_TOKEN", "")
    credentials = BasicCredentials(ak, sk, project_id)
    if security_token:
        credentials = credentials.with_security_token(security_token)
    http_config = build_http_config()
    return VpcClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(VpcRegion.value_of(region)).build()


def resolve_vpc_info(region, project_id, vpc_ids):
    """批量解析 VPC 名称和 CIDR

    :param region: 区域名称
    :param project_id: 项目 ID
    :param vpc_ids: VPC ID 列表
    :return: dict {vpc_id: {name, cidr}}
    """
    if not vpc_ids:
        return {}
    client = _build_vpc_client(region, project_id)
    if not client:
        return {vid: {"name": "", "cidr": ""} for vid in vpc_ids}
    try:
        from huaweicloudsdkvpc.v3 import ListVpcsRequest
        result = {}
        for vid in vpc_ids:
            req = ListVpcsRequest()
            req.id = [vid]
            resp = client.list_vpcs(req)
            vpcs = getattr(resp, 'vpcs', []) or []
            if vpcs:
                result[vid] = {"name": getattr(vpcs[0], 'name', ''), "cidr": getattr(vpcs[0], 'cidr', '')}
            else:
                result[vid] = {"name": "", "cidr": ""}
        return result
    except Exception:
        return {vid: {"name": "", "cidr": ""} for vid in vpc_ids}


def resolve_subnet_info(region, project_id, subnet_ids, vpc_id=None):
    """批量解析子网名称、CIDR、可用区等信息

    :param region: 区域名称
    :param project_id: 项目 ID
    :param subnet_ids: 子网 ID 列表
    :param vpc_id: 可选的 VPC ID，用于缩小查询范围
    :return: dict {subnet_id: {name, cidr, gateway_ip, availability_zone, available_ip_address_count}}
    """
    if not subnet_ids:
        return {}
    try:
        from huaweicloudsdkvpc.v2 import VpcClient as VpcClientV2
        from huaweicloudsdkvpc.v2.region.vpc_region import VpcRegion as VpcRegionV2
        from huaweicloudsdkvpc.v2 import ListSubnetsRequest
    except ImportError:
        return {sid: {"name": "", "cidr": "", "gateway_ip": "", "availability_zone": "", "available_ip_address_count": ""} for sid in subnet_ids}

    ak = os.getenv("HW_ACCESS_KEY", "")
    sk = os.getenv("HW_SECRET_KEY", "")
    security_token = os.getenv("HW_SECURITY_TOKEN", "")
    credentials = BasicCredentials(ak, sk, project_id)
    if security_token:
        credentials = credentials.with_security_token(security_token)
    http_config = build_http_config()

    try:
        client = VpcClientV2.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(VpcRegionV2.value_of(region)).build()
        result = {}
        req = ListSubnetsRequest()
        if vpc_id:
            req.vpc_id = vpc_id
        resp = client.list_subnets(req)
        subnets = getattr(resp, 'subnets', []) or []
        subnet_map = {getattr(s, 'id', ''): s for s in subnets}
        for sid in subnet_ids:
            s = subnet_map.get(sid)
            if s:
                result[sid] = {
                    "name": getattr(s, 'name', ''),
                    "cidr": getattr(s, 'cidr', ''),
                    "gateway_ip": getattr(s, 'gateway_ip', ''),
                    "availability_zone": getattr(s, 'availability_zone', ''),
                    "available_ip_address_count": getattr(s, 'available_ip_address_count', ''),
                }
            else:
                result[sid] = {"name": "", "cidr": "", "gateway_ip": "", "availability_zone": "", "available_ip_address_count": ""}
        return result
    except Exception:
        return {sid: {"name": "", "cidr": "", "gateway_ip": "", "availability_zone": "", "available_ip_address_count": ""} for sid in subnet_ids}


# 资源类型对应的列表查询脚本提示
_RESOURCE_LIST_HINTS = {
    "scalingGroup": "list_scaling_groups.py",
    "scalingConfiguration": "list_scaling_configs.py",
    "scalingPolicy": "list_scaling_policies.py",
    "lifecycleHook": "list_life_cycle_hooks.py",
    "warmPool": "list_warm_pool_instances.py",
}


def handle_not_found_error(error, resource_type, resource_id):
    """处理 404 资源不存在错误，输出友好提示

    :param error: 异常对象
    :param resource_type: 资源类型英文名，如 scalingGroup
    :param resource_id: 请求的资源 ID
    """
    error_str = str(error)
    is_404 = "404" in error_str or "could not be found" in error_str or "not exist" in error_str
    if is_404:
        list_script = _RESOURCE_LIST_HINTS.get(resource_type, "")
        hint = f"资源 [{resource_type}] 不存在 (ID: {resource_id})，请检查 ID 是否正确。"
        if list_script:
            hint += f"\n提示：可先运行 {list_script} 查看可用资源列表。"
        print(hint)
    else:
        print(f"查询失败: {error}")
    exit(1)

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListAllScalingV2PoliciesRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询全量弹性伸缩策略列表(v2)')
parser.add_argument('--scaling_resource_id', type=str, default=None,
                    help='伸缩组ID')
parser.add_argument('--scaling_resource_type', type=str, default=None,
                    choices=['SCALING_GROUP', 'BANDWIDTH'],
                    help='伸缩资源类型：SCALING_GROUP-伸缩组；BANDWIDTH-带宽')
parser.add_argument('--scaling_policy_name', type=str, default=None,
                    help='伸缩策略名称')
parser.add_argument('--scaling_policy_type', type=str, default=None,
                    choices=['ALARM', 'SCHEDULED', 'RECURRENCE'],
                    help='策略类型：ALARM-告警策略；SCHEDULED-定时策略；RECURRENCE-周期策略')
parser.add_argument('--scaling_policy_id', type=str, default=None,
                    help='伸缩策略ID')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询记录数，默认20，最大100')
parser.add_argument('--sort_by', type=str, default=None,
                    choices=['POLICY_NAME', 'TRIGGER_CONDITION', 'CREATE_TIME'],
                    help='排序方法：POLICY_NAME-策略名称；TRIGGER_CONDITION-触发条件；CREATE_TIME-创建时间')
parser.add_argument('--order', type=str, default=None,
                    choices=['ASC', 'DESC'],
                    help='排序顺序：ASC-升序；DESC-降序')
parser.add_argument('--enterprise_project_id', type=str, default=None,
                    help='企业项目ID，传入all_granted_eps表示查询所有授权企业项目')
parser.add_argument('--alarm_id', type=str, default=None,
                    help='告警ID，即告警规则的ID')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(policies):
    if not policies:
        print(f"没有找到伸缩策略 (区域: {Region})")
        return
    header = "scaling_policy_id\tscaling_policy_name\tpolicy_status\tscaling_policy_type\tscaling_resource_id\talarm_id\tcool_down_time\tcreate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(policies))
    for p in policies[start:end]:
        print(f"{getattr(p, 'scaling_policy_id', '')}\t{getattr(p, 'scaling_policy_name', '')}\t{getattr(p, 'policy_status', '')}\t{getattr(p, 'scaling_policy_type', '')}\t{getattr(p, 'scaling_resource_id', '')}\t{getattr(p, 'alarm_id', '')}\t{getattr(p, 'cool_down_time', 0)}\t{getattr(p, 'create_time', '')}")
    if len(policies) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_policies = []
        start = args.start_number
        while True:
            request = ListAllScalingV2PoliciesRequest()
            if args.scaling_resource_id:
                request.scaling_resource_id = args.scaling_resource_id
            if args.scaling_resource_type:
                request.scaling_resource_type = args.scaling_resource_type
            if args.scaling_policy_name:
                request.scaling_policy_name = args.scaling_policy_name
            if args.scaling_policy_type:
                request.scaling_policy_type = args.scaling_policy_type
            if args.scaling_policy_id:
                request.scaling_policy_id = args.scaling_policy_id
            if args.sort_by:
                request.sort_by = args.sort_by
            if args.order:
                request.order = args.order
            if args.enterprise_project_id:
                request.enterprise_project_id = args.enterprise_project_id
            if args.alarm_id:
                request.alarm_id = args.alarm_id
            request.start_number = start
            request.limit = args.limit

            response = client.list_all_scaling_v2_policies(request)
            policies = getattr(response, 'scaling_policies', []) or []
            all_policies.extend(policies)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_policies) >= total:
                break
            start = len(all_policies)

        render(all_policies)
    except Exception as e:
        print(f"as.list_all_scaling_v2_policies 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListApiVersionsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()


parser = argparse.ArgumentParser(description='查询AS API版本列表')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')


def main():
    global Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        request = ListApiVersionsRequest()
        response = client.list_api_versions(request)

        versions = getattr(response, 'versions', []) or []
        if not versions:
            print(f"没有找到API版本 (区域: {Region})")
            return

        header = "id\tstatus\tmin_version\tversion\tupdate"
        print(header)
        for v in versions:
            print(f"{getattr(v, 'id', '')}\t{getattr(v, 'status', '')}\t{getattr(v, 'min_version', '')}\t{getattr(v, 'version', '')}\t{getattr(v, 'update', '')}")

    except Exception as e:
        print(f"as.list_api_versions 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListGroupScheduledTasksRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询伸缩组计划任务列表')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--limit', type=int, default=100,
                    help='查询的记录条数，默认20，最大100')
parser.add_argument('--marker', type=str, default=None,
                    help='查询的分页marker')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(tasks):
    if not tasks:
        print(f"没有找到计划任务 (区域: {Region})")
        return
    header = "task_id\tname\tcreate_time\tupdate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(tasks))
    for t in tasks[start:end]:
        print(f"{getattr(t, 'task_id', '')}\t{getattr(t, 'name', '')}\t{getattr(t, 'create_time', '')}\t{getattr(t, 'update_time', '')}")
    if len(tasks) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_tasks = []
        marker = args.marker

        while True:
            request = ListGroupScheduledTasksRequest()
            request.scaling_group_id = args.scaling_group_id
            request.limit = args.limit
            if marker:
                request.marker = marker

            response = client.list_group_scheduled_tasks(request)
            tasks = getattr(response, 'scheduled_tasks', []) or []
            all_tasks.extend(tasks)

            page_info = getattr(response, 'page_info', None)
            next_marker = getattr(page_info, 'next_marker', '') if page_info else ''
            if not next_marker:
                break
            marker = next_marker

        render(all_tasks)
    except Exception as e:
        print(f"as.list_group_scheduled_tasks 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListHookInstancesRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()


parser = argparse.ArgumentParser(description='查询伸缩组挂钩实例列表')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--instance_id', type=str, default=None,
                    help='伸缩实例ID')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')


def main():
    global Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region


    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        request = ListHookInstancesRequest()
        request.scaling_group_id = args.scaling_group_id
        if args.instance_id:
            request.instance_id = args.instance_id

        response = client.list_hook_instances(request)
        infos = getattr(response, 'instance_hanging_info', []) or []

        if not infos:
            print(f"没有找到挂钩实例 (区域: {Region})")
            return

        header = "instance_id\tlifecycle_hook_name\tlifecycle_action_key\tlifecycle_hook_status\ttimeout\tdefault_result"
        print(header)
        for i in infos:
            print(f"{getattr(i, 'instance_id', '')}\t{getattr(i, 'lifecycle_hook_name', '')}\t{getattr(i, 'lifecycle_action_key', '')}\t{getattr(i, 'lifecycle_hook_status', '')}\t{getattr(i, 'timeout', '')}\t{getattr(i, 'default_result', '')}")

    except Exception as e:
        print(f"as.list_hook_instances 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListLifeCycleHooksRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()


parser = argparse.ArgumentParser(description='查询生命周期挂钩列表')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')


def main():
    global Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        request = ListLifeCycleHooksRequest()
        request.scaling_group_id = args.scaling_group_id

        response = client.list_life_cycle_hooks(request)
        hooks = getattr(response, 'lifecycle_hooks', []) or []

        if not hooks:
            print(f"没有找到生命周期挂钩 (区域: {Region})")
            return

        header = "lifecycle_hook_name\tlifecycle_hook_type\tdefault_result\tdefault_timeout\tnotification_topic_urn\tnotification_topic_name\tcreate_time"
        print(header)
        for h in hooks:
            print(f"{getattr(h, 'lifecycle_hook_name', '')}\t{getattr(h, 'lifecycle_hook_type', '')}\t{getattr(h, 'default_result', '')}\t{getattr(h, 'default_timeout', 0)}\t{getattr(h, 'notification_topic_urn', '')}\t{getattr(h, 'notification_topic_name', '')}\t{getattr(h, 'create_time', '')}")

    except Exception as e:
        print(f"as.list_life_cycle_hooks 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListResourceInstancesRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()


parser = argparse.ArgumentParser(description='按标签查询资源实例')
parser.add_argument('--resource_type', type=str, required=True,
                    help='资源类型，枚举类：SCALING_GROUP_TAG')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')


def main():
    global Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region


    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        request = ListResourceInstancesRequest()
        request.resource_type = args.resource_type

        response = client.list_resource_instances(request)
        resources = getattr(response, 'resources', []) or []
        total_count = getattr(response, 'total_count', 0)

        if not resources:
            print(f"没有找到资源实例 (区域: {Region})")
            return

        print(f"total_count:\t{total_count}")
        header = "resource_id\tresource_name\tresource_detail"
        print(header)
        for r in resources:
            print(f"{getattr(r, 'resource_id', '')}\t{getattr(r, 'resource_name', '')}\t{getattr(r, 'resource_detail', '')}")
            tags = getattr(r, 'tags', []) or []
            if tags:
                tag_str = ', '.join(f"{getattr(t, 'key', '')}={getattr(t, 'value', '')}" for t in tags)
                print(f"  tags: {tag_str}")

    except Exception as e:
        print(f"as.list_resource_instances 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingActivityLogsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩活动日志列表(v1)')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--start_time', type=str, default=None,
                    help='查询的起始时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--end_time', type=str, default=None,
                    help='查询的截止时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询记录数，默认20，最大100')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(logs):
    if not logs:
        print(f"没有找到伸缩活动日志 (区域: {Region})")
        return
    header = "id\tstatus\tstart_time\tend_time\tscaling_value\tinstance_value\tdesire_value\tdescription"
    print(header)
    start = Offset
    end = min(Offset + 50, len(logs))
    for l in logs[start:end]:
        print(f"{getattr(l, 'id', '')}\t{getattr(l, 'status', '')}\t{getattr(l, 'start_time', '')}\t{getattr(l, 'end_time', '')}\t{getattr(l, 'scaling_value', 0)}\t{getattr(l, 'instance_value', 0)}\t{getattr(l, 'desire_value', 0)}\t{getattr(l, 'description', '')}")
    if len(logs) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_logs = []
        start = args.start_number
        while True:
            request = ListScalingActivityLogsRequest()
            request.scaling_group_id = args.scaling_group_id
            if args.start_time:
                request.start_time = args.start_time
            if args.end_time:
                request.end_time = args.end_time
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_activity_logs(request)
            logs = getattr(response, 'scaling_activity_log', []) or []
            all_logs.extend(logs)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_logs) >= total:
                break
            start = len(all_logs)

        render(all_logs)
    except Exception as e:
        print(f"as.list_scaling_activity_logs 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingActivityV2LogsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩活动日志列表(v2)')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--log_id', type=str, default=None,
                    help='伸缩活动日志ID')
parser.add_argument('--start_time', type=str, default=None,
                    help='查询的起始时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--end_time', type=str, default=None,
                    help='查询的截止时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--type', type=str, default=None,
                    help='查询的伸缩活动类型（逗号分隔）：NORMAL-普通；MANNUAL_REMOVE-手动移除；MANNUAL_DELETE-手动移除并删除；MANNUAL_ADD-手动加入；ELB_CHECK_DELETE-ELB检查移除并删除；AUDIT_CHECK_DELETE-审计移除并删除；DIFF-期望实例数与实际不一致；MODIFY_ELB-LB迁移；ENTER_STANDBY-转入备用；EXIT_STANDBY-移出备用')
parser.add_argument('--status', type=str, default=None,
                    choices=['SUCCESS', 'FAIL', 'DOING'],
                    help='查询的伸缩活动状态：SUCCESS-成功；FAIL-失败；DOING-伸缩中')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询记录数，默认20，最大100')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(logs):
    if not logs:
        print(f"没有找到伸缩活动日志 (区域: {Region})")
        return
    header = "id\tstatus\ttype\tstart_time\tend_time\tscaling_value\tinstance_value\tdesire_value\tdescription"
    print(header)
    start = Offset
    end = min(Offset + 50, len(logs))
    for l in logs[start:end]:
        print(f"{getattr(l, 'id', '')}\t{getattr(l, 'status', '')}\t{getattr(l, 'type', '')}\t{getattr(l, 'start_time', '')}\t{getattr(l, 'end_time', '')}\t{getattr(l, 'scaling_value', 0)}\t{getattr(l, 'instance_value', 0)}\t{getattr(l, 'desire_value', 0)}\t{getattr(l, 'description', '')}")
    if len(logs) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_logs = []
        start = args.start_number
        while True:
            request = ListScalingActivityV2LogsRequest()
            request.scaling_group_id = args.scaling_group_id
            if args.log_id:
                request.log_id = args.log_id
            if args.start_time:
                request.start_time = args.start_time
            if args.end_time:
                request.end_time = args.end_time
            if args.type:
                request.type = args.type
            if args.status:
                request.status = args.status
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_activity_v2_logs(request)
            logs = getattr(response, 'scaling_activity_log', []) or []
            all_logs.extend(logs)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_logs) >= total:
                break
            start = len(all_logs)

        render(all_logs)
    except Exception as e:
        print(f"as.list_scaling_activity_v2_logs 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingConfigsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩配置列表')
parser.add_argument('--scaling_configuration_name', type=str, default=None,
                    help='伸缩配置名称')
parser.add_argument('--image_id', type=str, default=None,
                    help='镜像ID，同imageRef')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询的记录条数，默认为20')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(configs):
    if not configs:
        print(f"没有找到伸缩配置 (区域: {Region})")
        return
    header = "scaling_configuration_id\tscaling_configuration_name\tscaling_group_id\tcreate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(configs))
    for c in configs[start:end]:
        print(f"{getattr(c, 'scaling_configuration_id', '')}\t{getattr(c, 'scaling_configuration_name', '')}\t{getattr(c, 'scaling_group_id', '')}\t{getattr(c, 'create_time', '')}")
    if len(configs) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_configs = []
        start = args.start_number
        while True:
            request = ListScalingConfigsRequest()
            if args.scaling_configuration_name:
                request.scaling_configuration_name = args.scaling_configuration_name
            if args.image_id:
                request.image_id = args.image_id
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_configs(request)
            configs = getattr(response, 'scaling_configurations', []) or []
            all_configs.extend(configs)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_configs) >= total:
                break
            start = len(all_configs)

        render(all_configs)
    except Exception as e:
        print(f"as.list_scaling_configs 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id, resolve_vpc_info

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingGroupsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩组列表')
parser.add_argument('--scaling_group_name', type=str, default=None,
                    help='伸缩组名称')
parser.add_argument('--scaling_configuration_id', type=str, default=None,
                    help='伸缩配置ID，通过查询弹性伸缩配置列表接口获取')
parser.add_argument('--scaling_group_status', type=str, default=None,
                    choices=['INSERVICE', 'PAUSED', 'ERROR', 'DELETING', 'FREEZED'],
                    help='伸缩组状态：INSERVICE-正常；PAUSED-停用；ERROR-异常；DELETING-删除中；FREEZED-已冻结')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询的记录条数，默认为20，最大100')
parser.add_argument('--enterprise_project_id', type=str, default=None,
                    help='企业项目ID，传入all_granted_eps表示查询所有授权企业项目下的伸缩组')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(groups, vpc_info=None):
    if not groups:
        print(f"没有找到伸缩组 (区域: {Region})")
        return
    header = "scaling_group_id\tscaling_group_name\tstatus\tscaling_configuration_id\tcurrent_instance_number\tdesire_instance_number\tmin_instance_number\tmax_instance_number\tvpc_id\tvpc_name\tcreate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(groups))
    for g in groups[start:end]:
        vid = getattr(g, 'vpc_id', '')
        vname = (vpc_info or {}).get(vid, {}).get('name', '')
        print(f"{getattr(g, 'scaling_group_id', '')}\t{getattr(g, 'scaling_group_name', '')}\t{getattr(g, 'scaling_group_status', '')}\t{getattr(g, 'scaling_configuration_id', '')}\t{getattr(g, 'current_instance_number', 0)}\t{getattr(g, 'desire_instance_number', 0)}\t{getattr(g, 'min_instance_number', 0)}\t{getattr(g, 'max_instance_number', 0)}\t{vid}\t{vname}\t{getattr(g, 'create_time', '')}")
    if len(groups) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Offset, Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_groups = []
        start = args.start_number
        while True:
            request = ListScalingGroupsRequest()
            if args.scaling_group_name:
                request.scaling_group_name = args.scaling_group_name
            if args.scaling_configuration_id:
                request.scaling_configuration_id = args.scaling_configuration_id
            if args.scaling_group_status:
                request.scaling_group_status = args.scaling_group_status
            if args.enterprise_project_id:
                request.enterprise_project_id = args.enterprise_project_id
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_groups(request)
            groups = getattr(response, 'scaling_groups', []) or []
            all_groups.extend(groups)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_groups) >= total:
                break
            start = len(all_groups)

        vpc_ids = list(set(getattr(g, 'vpc_id', '') for g in all_groups if getattr(g, 'vpc_id', '')))
        vpc_info = resolve_vpc_info(Region, project_id, vpc_ids) if vpc_ids else {}
        render(all_groups, vpc_info)
    except Exception as e:
        print(f"as.list_scaling_groups 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingInstancesRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩组实例列表')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--life_cycle_state', type=str, default=None,
                    choices=['INSERVICE', 'PENDING', 'REMOVING', 'PENDING_WAIT', 'REMOVING_WAIT'],
                    help='实例生命周期状态：INSERVICE-正在使用；PENDING-正在加入；REMOVING-正在移出；PENDING_WAIT-正在加入等待；REMOVING_WAIT-正在移出等待')
parser.add_argument('--health_status', type=str, default=None,
                    choices=['INITIALIZING', 'NORMAL', 'ERROR'],
                    help='实例健康状态：INITIALIZING-初始化；NORMAL-正常；ERROR-异常')
parser.add_argument('--protect_from_scaling_down', type=str, default=None,
                    choices=['true', 'false'],
                    help='实例保护状态：true-已设置实例保护；false-未设置实例保护')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询的记录条数，默认为20')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(instances):
    if not instances:
        print(f"没有找到伸缩组实例 (区域: {Region})")
        return
    header = "instance_id\tinstance_name\tlife_cycle_state\thealth_status\tscaling_configuration_id\tprotect_from_scaling_down\tcreate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(instances))
    for i in instances[start:end]:
        print(f"{getattr(i, 'instance_id', '')}\t{getattr(i, 'instance_name', '')}\t{getattr(i, 'life_cycle_state', '')}\t{getattr(i, 'health_status', '')}\t{getattr(i, 'scaling_configuration_id', '')}\t{getattr(i, 'protect_from_scaling_down', False)}\t{getattr(i, 'create_time', '')}")
    if len(instances) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_instances = []
        start = args.start_number
        while True:
            request = ListScalingInstancesRequest()
            request.scaling_group_id = args.scaling_group_id
            if args.life_cycle_state:
                request.life_cycle_state = args.life_cycle_state
            if args.health_status:
                request.health_status = args.health_status
            if args.protect_from_scaling_down:
                request.protect_from_scaling_down = args.protect_from_scaling_down
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_instances(request)
            instances = getattr(response, 'scaling_group_instances', []) or []
            all_instances.extend(instances)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_instances) >= total:
                break
            start = len(all_instances)

        render(all_instances)
    except Exception as e:
        print(f"as.list_scaling_instances 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingNotificationsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()


parser = argparse.ArgumentParser(description='查询弹性伸缩组通知列表')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')


def main():
    global Region
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        request = ListScalingNotificationsRequest()
        request.scaling_group_id = args.scaling_group_id

        response = client.list_scaling_notifications(request)
        topics = getattr(response, 'topics', []) or []

        if not topics:
            print(f"没有找到通知 (区域: {Region})")
            return

        header = "topic_urn\ttopic_name\ttopic_scene"
        print(header)
        for t in topics:
            scene = ', '.join(getattr(t, 'topic_scene', []) or [])
            print(f"{getattr(t, 'topic_urn', '')}\t{getattr(t, 'topic_name', '')}\t{scene}")

    except Exception as e:
        print(f"as.list_scaling_notifications 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingPoliciesRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩策略列表(v1)')
parser.add_argument('--scaling_group_id', type=str, required=True,
                    help='伸缩组ID，可通过 list_scaling_groups.py 获取')
parser.add_argument('--scaling_policy_name', type=str, default=None,
                    help='伸缩策略名称')
parser.add_argument('--scaling_policy_type', type=str, default=None,
                    choices=['ALARM', 'SCHEDULED', 'RECURRENCE'],
                    help='策略类型：ALARM-告警策略；SCHEDULED-定时策略；RECURRENCE-周期策略')
parser.add_argument('--scaling_policy_id', type=str, default=None,
                    help='伸缩策略ID')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询记录数，默认20，最大100')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(policies):
    if not policies:
        print(f"没有找到伸缩策略 (区域: {Region})")
        return
    header = "scaling_policy_id\tscaling_policy_name\tpolicy_status\tscaling_policy_type\talarm_id\tcool_down_time\tcreate_time"
    print(header)
    start = Offset
    end = min(Offset + 50, len(policies))
    for p in policies[start:end]:
        print(f"{getattr(p, 'scaling_policy_id', '')}\t{getattr(p, 'scaling_policy_name', '')}\t{getattr(p, 'policy_status', '')}\t{getattr(p, 'scaling_policy_type', '')}\t{getattr(p, 'alarm_id', '')}\t{getattr(p, 'cool_down_time', 0)}\t{getattr(p, 'create_time', '')}")
    if len(policies) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_policies = []
        start = args.start_number
        while True:
            request = ListScalingPoliciesRequest()
            request.scaling_group_id = args.scaling_group_id
            if args.scaling_policy_name:
                request.scaling_policy_name = args.scaling_policy_name
            if args.scaling_policy_type:
                request.scaling_policy_type = args.scaling_policy_type
            if args.scaling_policy_id:
                request.scaling_policy_id = args.scaling_policy_id
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_policies(request)
            policies = getattr(response, 'scaling_policies', []) or []
            all_policies.extend(policies)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_policies) >= total:
                break
            start = len(all_policies)

        render(all_policies)
    except Exception as e:
        print(f"as.list_scaling_policies 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()

import argparse
import sys
import os

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from config import load_credentials, build_http_config
from _common import resolve_project_id

from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkas.v1 import AsClient, ListScalingPolicyExecuteLogsRequest
from huaweicloudsdkas.v1.region.as_region import AsRegion

# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
Offset = 0

parser = argparse.ArgumentParser(description='查询弹性伸缩策略执行日志列表')
parser.add_argument('--scaling_policy_id', type=str, required=True,
                    help='伸缩策略ID，可通过 list_scaling_policies.py 获取')
parser.add_argument('--log_id', type=str, default=None,
                    help='日志ID')
parser.add_argument('--scaling_resource_type', type=str, default=None,
                    choices=['SCALING_GROUP', 'BANDWIDTH'],
                    help='伸缩资源类型：SCALING_GROUP-伸缩组；BANDWIDTH-带宽')
parser.add_argument('--scaling_resource_id', type=str, default=None,
                    help='伸缩资源ID')
parser.add_argument('--execute_type', type=str, default=None,
                    choices=['SCHEDULED', 'RECURRENCE', 'ALARM', 'MANUAL'],
                    help='策略执行类型：SCHEDULED-自动触发(定时)；RECURRENCE-自动触发(周期)；ALARM-自动触发(告警)；MANUAL-手动触发')
parser.add_argument('--start_time', type=str, default=None,
                    help='查询的起始时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--end_time', type=str, default=None,
                    help='查询的截止时间，格式为yyyy-MM-ddThh:mm:ssZ')
parser.add_argument('--start_number', type=int, default=0,
                    help='查询的起始行号，默认为0')
parser.add_argument('--limit', type=int, default=100,
                    help='查询记录数，默认20，最大100')
parser.add_argument('--project_id', type=str, default=None,
                    help='项目ID')
parser.add_argument('--region', type=str,
                    help='区域，默认从环境变量获取')
parser.add_argument('--offset', type=int,
                    help='客户端分页偏移量，从0开始')


def render(logs):
    if not logs:
        print(f"没有找到策略执行日志 (区域: {Region})")
        return
    header = "id\tstatus\texecute_type\texecute_time\told_value\tdesire_value\tlimit_value\ttype\tfailed_reason"
    print(header)
    start = Offset
    end = min(Offset + 50, len(logs))
    for l in logs[start:end]:
        print(f"{getattr(l, 'id', '')}\t{getattr(l, 'status', '')}\t{getattr(l, 'execute_type', '')}\t{getattr(l, 'execute_time', '')}\t{getattr(l, 'old_value', '')}\t{getattr(l, 'desire_value', '')}\t{getattr(l, 'limit_value', '')}\t{getattr(l, 'type', '')}\t{getattr(l, 'failed_reason', '')}")
    if len(logs) > end:
        print(f"可以使用 --offset={end} 参数继续获取后续数据")


def main():
    global Region, Offset
    args = parser.parse_args()

    if args.region is not None:
        Region = args.region

    if args.offset is not None:
        Offset = args.offset

    if Offset < 0:
        Offset = 0

    # 使用 sdk
    try:
        project_id = resolve_project_id(Region, args.project_id)
        credentials = BasicCredentials(AK, SK, project_id)
        if SecurityToken:
            credentials = credentials.with_security_token(SecurityToken)
        http_config = build_http_config()

        client = AsClient.new_builder().with_http_config(http_config).with_credentials(credentials).with_region(AsRegion.value_of(Region)).build()

        all_logs = []
        start = args.start_number
        while True:
            request = ListScalingPolicyExecuteLogsRequest()
            request.scaling_policy_id = args.scaling_policy_id
            if args.log_id:
                request.log_id = args.log_id
            if args.scaling_resource_type:
                request.scaling_resource_type = args.scaling_resource_type
            if args.scaling_resource_id:
                request.scaling_resource_id = args.scaling_resource_id
            if args.execute_type:
                request.execute_type = args.execute_type
            if args.start_time:
                request.start_time = args.start_time
            if args.end_time:
                request.end_time = args.end_time
            request.start_number = start
            request.limit = args.limit

            response = client.list_scaling_policy_execute_logs(request)
            logs = getattr(response, 'scaling_policy_execute_log', []) or []
            all_logs.extend(logs)

            total = getattr(response, 'total_number', 0)
            if total <= 0 or len(all_logs) >= total:
                break
            start = len(all_logs)

        render(all_logs)
    except Exception as e:
        print(f"as.list_scaling_policy_execute_logs 查询失败: {e}")
        exit(1)


if __name__ == '__main__':
    main()