
Huawei Cloud Monitoring Query
- 106 installs
- 19 repo stars
- Updated July 31, 2026
- huaweicloud/huaweicloud-skills
Query Huawei Cloud CES monitoring and enterprise-project resources: alarm rules, histories, templates, dashboards, and enterprise projects, read-only.
About
Read-only queries of Huawei Cloud monitoring (CES) and enterprise projects (EPS) including alarm rules, histories, templates, dashboards, notification masks, and project quotas via Python SDK scripts. A developer uses it to check alarm status and enterprise-project info.
- Covers alarm rules, histories, templates, and dashboards
- Read-only; runs through bundled Python scripts
Huawei Cloud Monitoring Query by the numbers
- 106 all-time installs (skills.sh)
- +14 installs in the week ending Aug 2, 2026 (Skillselion tracking)
- Ranked #554 of 1,042 Cloud & Infrastructure skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/huaweicloud/huaweicloud-skills --skill huawei-cloud-monitoring-queryAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 106 |
|---|---|
| repo stars | ★ 19 |
| Last updated | July 31, 2026 |
| Repository | huaweicloud/huaweicloud-skills ↗ |
What it does
Query Huawei Cloud CES monitoring and enterprise-project resources: alarm rules, histories, templates, dashboards, and enterprise projects, read-only.
Files
Huawei Cloud Resource Query
⚠️ Execution Method (Must Read): This skill executes queries via local Python scripts. Using hcloud, openstack, or other CLI tools or direct API calls is prohibited.
>
- Query scripts are located under the skill directoryscripts/<service_category>/(e.g.,scripts/as/list_scaling_groups.py)
- All scripts and environment check scripts are inside the skill package. You must use `skill action=exec` to execute them. Do not run them directly in a shell.
- For specific script paths and parameters, refer to references/<service>/guide.md- Do not attempt hcloud, openstack, curl IAM, or any other CLI/API methods. This skill does not depend on those tools.
- All paths are relative to the skill directory, which is the directory where this SKILL.md is located.
Overview
This skill is a standalone read-only query skill that uses local Python scripts to call the Huawei Cloud Python SDK to query Huawei Cloud resources, available specifications, and existing resource information.
This skill is applicable to the following scenarios:
1. Query available cloud resource specifications in a specific region 2. Query available images for a specific OS type 3. Query disk types and existing disk information 4. Query existing resources and their key attributes 5. Query resources that were not created via Terraform or other IaC tools 6. Prepare real parameters for automation configuration, resource verification, or environment inventory 7. Obtain reusable information such as resource IDs, names, specifications, images, networks, disks, etc.
This skill does NOT handle the following:
1. Creating resources 2. Modifying resources 3. Deleting resources 4. Guessing or fabricating information that has not been queried
---
Capability Scope
This skill provides query capabilities through categorized scripts in the scripts directory, and usage instructions through categorized guides in the references directory.
Capabilities provided by this skill include:
1. Query resource lists 2. Query individual resource details 3. Query selection information such as available specifications, images, and disk types 4. Query key identifiers and dependency relationships of existing resources
---
Usage Principles
Important: Script paths executed within this skill are all relative to the skill directory, which is the directory where this SKILL.md is located.
1. This skill only performs queries; it does not perform any write operations 2. Prioritize using explicitly specified region, project, AZ, resource name, resource ID, etc. provided by the user 3. Query results must be based on actual API responses; do not infer from experience 4. Returned results should prioritize retaining key fields for subsequent reuse 5. When the result set is large, narrow the scope first by region, name, id, status, tag, etc. 6. If there is no corresponding script or guide for the current resource type, clearly state that it is unsupported; do not return unreliable results 7. If the user has not provided necessary scope information and there are no default values in the environment, confirm the details before executing the query 8. Execute directly according to guide.md; do not read the script contents in the scripts directory 9. Cache results when the output is large 10. You must run -h to view usage before executing each script 11. Do not invent script names; execute according to the script names in guide.md. If a script name is not in guide.md, it means it is not supported
---
Prerequisites
Before use, you must run the environment check script to complete environment validation and dependency installation in one step:
- Linux / macOS:
skill action=exec: bash skill://scripts/check_env.sh - Windows:
skill action=exec: powershell -ExecutionPolicy Bypass -File skill://scripts/check_env.ps1
Windows note: Do not use && to chain commands (PowerShell 5.x does not support it); use semicolons if you need to change directories first.The script will check in order: Python >= 3.6 → install dependencies → validate SDK → validate credentials → validate service availability. If the environment check fails, fix the issues before proceeding with other scripts.
Environment Variables:
| Variable | Required | Description |
|---|---|---|
| HW_ACCESS_KEY | Yes | Huawei Cloud AK |
| HW_SECRET_KEY | Yes | Huawei Cloud SK |
| HW_REGION_NAME | No | Default cn-north-4 |
| HW_PROJECT_ID | No | Project ID (automatically obtained via IAM API if not set) |
| HW_SECURITY_TOKEN | No | Required when using temporary AK/SK |
Do not output the values of the above environment variables. For additional parameters required by other resource scripts (availability zone, enterprise project, etc.), see the corresponding guide.md.
---
Execution Flow
When this skill is invoked, you must follow these steps. Do not wait for the user to prompt again:
Step 1: Environment Preparation
Run the environment check script to ensure dependencies are installed and credentials are configured:
- Linux / macOS:
skill action=exec: bash skill://scripts/check_env.sh - Windows:
skill action=exec: powershell -ExecutionPolicy Bypass -File skill://scripts/check_env.ps1
If the environment check fails, fix the issues as prompted and re-run until it passes.
Step 2: Identify and Execute Query Scripts
1. Based on the user's query intent, read references/<service>/guide.md to determine the script path and parameters to execute 2. First run -h to view script usage:
- Linux / macOS:
skill action=exec: skill://.venv/bin/python3 skill://scripts/<service>/<script>.py -h - Windows:
skill action=exec: skill://.venv/Scripts/python3.exe skill://scripts/<service>/<script>.py -h
3. Assemble parameters based on user needs and execute the script:
- Linux / macOS:
skill action=exec: skill://.venv/bin/python3 skill://scripts/<service>/<script>.py <parameters> - Windows:
skill action=exec: skill://.venv/Scripts/python3.exe skill://scripts/<service>/<script>.py <parameters>
4. Format the results and return them to the user
Important:
- All scripts and environment check scripts are inside the skill package. You must use `skill action=exec` to execute them. Do not run them directly in a shell.
- The venv is automatically created by the check_env script. On Linux/macOS, Python is located at
.venv/bin/python3; on Windows, at.venv/Scripts/python3.exe - Do not use
python3directly to execute scripts - Do not read the source code of scripts in the scripts directory; just follow the instructions in guide.md
- Cache results when the output is large
- The
--project_idparameter is optional; if not provided, it is automatically obtained via IAM API based on region
---
Directory Structure
The directory conventions are as follows (all paths are relative to the skill directory):
1. scripts/<resource_category>/ contains the Python query scripts for corresponding resources. You do not need to read the script contents; just execute the scripts according to the usage instructions in guide.md 2. references/<resource_category>/guide.md contains the usage guide for corresponding resources 3. Each script is responsible for only one clear, single query action 4. Each resource category must maintain at least one guide.md to document script capabilities, parameters, and usage
---
Parameter Confirmation
Before executing query scripts, confirm the following parameters:
| Parameter | Required | Description |
|---|---|---|
| region | Yes | Huawei Cloud region, e.g., cn-north-4 |
| --project_id | No | Project ID; automatically obtained if not provided |
| --availability_zone | No | Availability zone; required for some resource queries |
For script-specific parameters, see references/<service>/guide.md.
---
Output Format
Query results are output in JSON format and contain the following common fields:
total: Total number of matched resourcesitems: Resource list, each resource contains key fields such as id, name, status, etc.- Specific fields vary by resource type; see individual guide.md for details
---
Verification Methods
1. Run the environment check script to confirm dependencies and credentials are available 2. Use the -h parameter to view script usage and confirm parameters are correct 3. Execute queries against known resources and compare with console data to verify result accuracy 4. Check whether the returned total count is reasonable
---
Best Practices
1. Narrow the query scope first (specify region, availability zone, etc.) to avoid returning too much data 2. Use --help to view the complete list of supported parameters for a script 3. Cache large query results locally to avoid repeated requests 4. When multiple resource information is needed, query in dependency order (e.g., query VPC first, then subnets) 5. When a script fails, check environment variables and network connectivity first
---
Reference Documentation
- Query script usage guides for each service:
references/<service>/guide.md
---
Notes
1. This skill only provides read-only query capabilities; it does not perform any write operations 2. Do not output the values of environment variables such as HW_ACCESS_KEY, HW_SECRET_KEY, etc. 3. All scripts must be executed via skill action=exec; do not run them directly in a shell 4. Do not guess script names; strictly execute according to the names in guide.md 5. You must run the environment check script before querying 6. When using temporary AK/SK, you need to set HW_SECURITY_TOKEN
CES Python Script Usage Guide
---
list_agent_dimension_info.py — List Agent Dimension Information
Purpose: List agent dimension information. Usage: python scripts/ces/list_agent_dimension_info.py -h
---
list_alarm_histories.py — List Alarm Histories
Purpose: List alarm histories. Usage: python scripts/ces/list_alarm_histories.py -h
---
list_alarm_rule_policies.py — List Alarm Policies Bound to an Alarm Rule
Purpose: List alarm policies bound to an alarm rule. Usage: python scripts/ces/list_alarm_rule_policies.py -h
---
list_alarm_rule_resources.py — List Resources Associated with an Alarm Rule
Purpose: List resources associated with an alarm rule. Usage: python scripts/ces/list_alarm_rule_resources.py -h
---
list_alarm_rules.py — List Alarm Rules
Purpose: List alarm rules. Usage: python scripts/ces/list_alarm_rules.py -h
---
list_alarm_template_association_alarms.py — List Alarm Rules Associated with an Alarm Template
Purpose: List alarm rules associated with an alarm template. Usage: python scripts/ces/list_alarm_template_association_alarms.py -h
---
list_alarm_templates.py — List Alarm Templates
Purpose: List alarm templates. Usage: python scripts/ces/list_alarm_templates.py -h
---
list_ces_target_project_tags.py — List CES Target Project Tags
Purpose: List CES target project tags. Usage: python scripts/ces/list_ces_target_project_tags.py -h
---
list_dashboard_infos.py — List Dashboards
Purpose: List dashboards. Usage: python scripts/ces/list_dashboard_infos.py -h
---
list_dashboard_widgets.py — List Dashboard Widgets
Purpose: List dashboard widgets. Usage: python scripts/ces/list_dashboard_widgets.py -h
---
list_notification_mask_resources.py — List Resources Associated with Notification Masking Rules
Purpose: List resources associated with notification masking rules. Usage: python scripts/ces/list_notification_mask_resources.py -h
---
list_notification_masks.py — List Notification Masking Rules
Purpose: List notification masking rules. Usage: python scripts/ces/list_notification_masks.py -h
---
list_one_click_alarm_rules.py — List One-Click Alarm Rules
Purpose: List one-click alarm rules. Usage: python scripts/ces/list_one_click_alarm_rules.py -h
---
list_one_click_alarms.py — List One-Click Alarms
Purpose: List one-click alarms. Usage: python scripts/ces/list_one_click_alarms.py -h
---
list_resource_groups.py — List Resource Groups
Purpose: List resource groups. Usage: python scripts/ces/list_resource_groups.py -h
---
list_resource_groups_services_resources.py — List Resource Group Service Resources
Purpose: List resource group service resources. Usage: python scripts/ces/list_resource_groups_services_resources.py -h
---
show_alarm_template.py — Show Alarm Template Details
Purpose: Show alarm template details. Usage: python scripts/ces/show_alarm_template.py -h
---
show_resource_group.py — Show Resource Group Details
Purpose: Show resource group details. Usage: python scripts/ces/show_resource_group.py -h
---
show_widget.py — Show Dashboard Widget Details
Purpose: Show dashboard widget details. Usage: python scripts/ces/show_widget.py -h
Enterprise Project EPS Python Script Usage Guide
EPS is a global service that uses GlobalCredentials for authentication. The --project_id parameter is not required.
---
Enterprise Projects
list_enterprise_projects.py — List Enterprise Projects
Purpose: List enterprise projects authorized for the current user, including ID, name, status, type, description, creation time, and update time. Supports filtering by ID/name/status/type, sorting, and pagination. Usage: python scripts/eps/list_enterprise_projects.py -h
---
show_enterprise_project.py — Show Enterprise Project Details
Purpose: Show details of a specified enterprise project, including ID, name, status, type, description, creation time, update time, and deletion marker. Usage: python scripts/eps/show_enterprise_project.py -h
---
show_enterprise_project_quota.py — Show Enterprise Project Quota
Purpose: Show quota information for enterprise projects, including usage, quota, and limit for each resource type. Usage: python scripts/eps/show_enterprise_project_quota.py -h
---
Cloud Services and Resources
list_providers.py — List Cloud Service Providers
Purpose: List cloud service providers supported by enterprise projects, including cloud service name, display name, and supported resource types. Supports filtering by cloud service name and pagination. Usage: python scripts/eps/list_providers.py -h
---
show_resource_bind_enterprise_project.py — List Resources Bound to an Enterprise Project
Purpose: List resources bound to a specified enterprise project, including resource ID, name, type, and owning project. Supports filtering by resource type and project ID, searching by resource name, and pagination. Usage: python scripts/eps/show_resource_bind_enterprise_project.py -h
---
show_associated_resources.py — Show Associated Resources
Purpose: Show associated enterprise project resource information for a specified resource, including resource ID, name, type, and associated resource list. Usage: python scripts/eps/show_associated_resources.py -h
---
list_resource_mapping.py — List Resource Type Mappings
Purpose: List resource type mappings, returning key-value pair mappings of resource types. Usage: python scripts/eps/list_resource_mapping.py -h
---
Migration Records
list_migration_records.py — List Migration Records
Purpose: List enterprise project migration records, including resource ID, name, type, operation type, source/target enterprise project, event time, and status. Supports filtering by time range and resource ID. Usage: python scripts/eps/list_migration_records.py -h
---
Service Configuration
show_ep_configs.py — Show Service Configuration
Purpose: Show enterprise project service configuration information. Usage: python scripts/eps/show_ep_configs.py -h
---
API Versions
list_api_versions.py — List API Versions
Purpose: List API versions for enterprise projects, including version ID, status, and update time. Usage: python scripts/eps/list_api_versions.py -h
---
show_api_version.py — Show API Version Details
Purpose: Show details of a specified enterprise project API version, including version ID, status, update time, and links. Usage: python scripts/eps/show_api_version.py -h
huaweicloudsdkcore>=3.1.0
huaweicloudsdkiam>=3.1.0
huaweicloudsdkces>=3.1.0
huaweicloudsdkeps>=3.1.0import 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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import ListAgentDimensionInfoRequest, CesClient
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询Agent维度信息列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--instance_id", type=str, required=True, help="ECS实例ID(36位),可通过 scripts/ecs/list_servers_details.py 获取")
parser.add_argument("--dim_name", type=str, required=True, help="维度名称,枚举值:mount_point(挂载点)/disk(磁盘)/proc(进程)/gpu(显卡)/raid(RAID控制器)")
parser.add_argument("--dim_value", type=str, help="维度值,32位字符串")
parser.add_argument("--offset", type=int, help="分页偏移量,最小值0,最大值2147483647,默认0")
parser.add_argument("--limit", type=int, help="分页大小,最小值1,最大值1000,默认1000")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAgentDimensionInfoRequest()
request.instance_id = args.instance_id
request.dim_name = args.dim_name
if args.dim_value:
request.dim_value = args.dim_value
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_agent_dimension_info(request)
dimensions = getattr(response, 'dimensions', []) or []
count = getattr(response, 'count', 0)
if not dimensions:
print("查询结果为空")
exit(0)
output = f"Agent维度信息列表(共{count}个):\n"
for dim in dimensions:
name = getattr(dim, 'name', '')
value = getattr(dim, 'value', '')
origin_value = getattr(dim, 'origin_value', '')
output += f"维度名称: {name}, 维度值: {value}, 实际维度信息: {origin_value}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_agent_dimension_info 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import ListAlarmHistoriesRequest, CesClient
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警历史列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_id", type=str, help="告警ID,多个用逗号分隔,可通过 list_alarm_rules.py 获取,最多50个")
parser.add_argument("--record_id", type=str, help="告警流水号,以ah开头,后跟22位字母或数字,长度24")
parser.add_argument("--name", type=str, help="告警规则名称(支持模糊匹配),最大128字符")
parser.add_argument("--status", type=str, help="告警状态,多个用逗号分隔:ok(已解决)/alarm(告警中)/invalid(已失效)/insufficient_data(数据不足)/ok_manual(已解决手动),最多3个")
parser.add_argument("--level", type=int, help="告警级别:1紧急/2重要/3次要/4提示")
parser.add_argument("--namespace", type=str, help="服务命名空间,格式service.item,长度3-32")
parser.add_argument("--resource_id", type=str, help="告警资源ID,多维度按字母升序逗号分隔,长度0-2048")
parser.add_argument("--from_time", type=str, help="查询起始更新时间,格式:2022-02-10T10:05:46+08:00。必须与to_time同时使用")
parser.add_argument("--to_time", type=str, help="查询截止更新时间,格式:2022-02-10T10:05:47+08:00。必须与from_time同时使用")
parser.add_argument("--alarm_type", type=str, help="告警类型:event(事件类型)/metric(指标类型)")
parser.add_argument("--create_time_from", type=str, help="查询起始创建时间,格式:2022-02-10T10:05:46+08:00。必须与create_time_to同时使用")
parser.add_argument("--create_time_to", type=str, help="查询截止创建时间,格式:2022-02-10T10:05:47+08:00。必须与create_time_from同时使用")
parser.add_argument("--offset", type=int, help="分页偏移量,最小0,最大1000000000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,最小1,最大100,默认100")
parser.add_argument("--order_by", type=str, help="排序字段,枚举值:first_alarm_time/update_time/alarm_level/record_id,默认update_time")
parser.add_argument("--mask_status", type=str, help="屏蔽状态,枚举值:UN_MASKED(活跃告警)/MASKED(屏蔽告警)")
args = parser.parse_args()
if (args.from_time is not None) != (args.to_time is not None):
parser.error("from_time 和 to_time 必须同时指定或同时不指定")
if (args.create_time_from is not None) != (args.create_time_to is not None):
parser.error("create_time_from 和 create_time_to 必须同时指定或同时不指定")
if args.region is not None:
Region = args.region
try:
request = ListAlarmHistoriesRequest()
if args.alarm_id:
request.alarm_id = args.alarm_id.split(',')
if args.record_id:
request.record_id = args.record_id
if args.name:
request.name = args.name
if args.status:
request.status = args.status.split(',')
if args.level is not None:
request.level = args.level
if args.namespace:
request.namespace = args.namespace
if args.resource_id:
request.resource_id = args.resource_id
if args.from_time:
request._from = args.from_time
if args.to_time:
request.to = args.to_time
if args.alarm_type:
request.alarm_type = args.alarm_type
if args.create_time_from:
request.create_time_from = args.create_time_from
if args.create_time_to:
request.create_time_to = args.create_time_to
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
if args.order_by:
request.order_by = args.order_by
if args.mask_status:
request.mask_status = args.mask_status
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_histories(request)
alarm_histories = getattr(response, 'alarm_histories', []) or []
count = getattr(response, 'count', 0)
if not alarm_histories:
print("查询结果为空")
exit(0)
output = f"告警历史列表(共{count}个):\n"
output += "流水号\t告警ID\t名称\t状态\t级别\t告警类型\t开始时间\t结束时间\n"
for history in alarm_histories:
record_id = getattr(history, 'record_id', '')
alarm_id = getattr(history, 'alarm_id', '')
name = getattr(history, 'name', '')
status = getattr(history, 'status', '')
level = getattr(history, 'level', '')
alarm_type = getattr(history, 'type', '')
begin_time = getattr(history, 'begin_time', '')
end_time = getattr(history, 'end_time', '')
output += f"{record_id}\t{alarm_id}\t{name}\t{status}\t{level}\t{alarm_type}\t{begin_time}\t{end_time}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_histories 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import ListAlarmRulePoliciesRequest, CesClient
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警规则绑定的告警策略列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_id", type=str, required=True, help="告警规则ID,以al开头,后跟22个数字或字母,长度24,可通过 list_alarm_rules.py 获取")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=50, help="分页大小,范围1-100,默认50")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAlarmRulePoliciesRequest()
request.alarm_id = args.alarm_id
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_rule_policies(request)
policies = getattr(response, 'policies', []) or []
count = getattr(response, 'count', 0)
if not policies:
print("查询结果为空")
exit(0)
output = f"告警规则 {args.alarm_id} 绑定的策略列表(共{count}个):\n"
output += "指标名称\t命名空间\t维度名称\t聚合方式\t比较运算符\t阈值\t单位\t触发次数\t告警级别\n"
for policy in policies:
metric_name = getattr(policy, 'metric_name', '')
namespace = getattr(policy, 'namespace', '')
dimension_name = getattr(policy, 'dimension_name', '')
filter = getattr(policy, 'filter', '')
comparison_operator = getattr(policy, 'comparison_operator', '')
value = getattr(policy, 'value', '')
unit = getattr(policy, 'unit', '')
policy_count = getattr(policy, 'count', '')
level = getattr(policy, 'level', '')
output += f"{metric_name}\t{namespace}\t{dimension_name}\t{filter}\t{comparison_operator}\t{value}\t{unit}\t{policy_count}\t{level}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_rule_policies 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListAlarmRuleResourcesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警规则关联的资源列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_id", type=str, required=True, help="告警规则ID,以al开头,后跟22位字母或数字,长度24,可通过 list_alarm_rules.py 获取")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=50, help="分页大小,范围1-100,默认50")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAlarmRuleResourcesRequest()
request.alarm_id = args.alarm_id
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_rule_resources(request)
resources = getattr(response, 'resources', []) or []
count = getattr(response, 'count', 0)
if not resources:
print("查询结果为空")
exit(0)
output = f"告警规则 {args.alarm_id} 关联资源列表(共{count}个):\n"
for idx, resource in enumerate(resources, 1):
output += f"资源 {idx}:\n"
if isinstance(resource, list):
for dim in resource:
name = getattr(dim, 'name', '')
value = getattr(dim, 'value', '')
output += f" {name}: {value}\n"
else:
output += f" {str(resource)}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_rule_resources 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListAlarmRulesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警规则列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_id", type=str, help="告警规则ID,以al开头,后跟22位字母或数字,长度24")
parser.add_argument("--name", type=str, help="告警规则名称(支持模糊匹配),长度1-128")
parser.add_argument("--namespace", type=str, help="服务命名空间,格式service.item")
parser.add_argument("--resource_id", type=str, help="告警资源ID,多维度按字母升序逗号分隔,长度0-700")
parser.add_argument("--enterprise_project_id", type=str, help="企业项目ID,长度36,也可为0或all_granted_eps,可通过 ../eps/list_enterprise_projects.py 获取")
parser.add_argument("--product_name", type=str, help="云产品名称,如SYS.ECS,instance_id,长度0-128")
parser.add_argument("--resource_level", type=str, help="资源层级,枚举值:product(云产品)/dimension(子维度)")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=50, help="分页大小,范围1-100,默认50")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAlarmRulesRequest()
if args.alarm_id:
request.alarm_id = args.alarm_id
if args.name:
request.name = args.name
if args.namespace:
request.namespace = args.namespace
if args.resource_id:
request.resource_id = args.resource_id
if args.enterprise_project_id:
request.enterprise_project_id = args.enterprise_project_id
if args.product_name:
request.product_name = args.product_name
if args.resource_level:
request.resource_level = args.resource_level
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_rules(request)
alarms = getattr(response, 'alarms', []) or []
count = getattr(response, 'count', 0)
if not alarms:
print("查询结果为空")
exit(0)
output = f"告警规则列表(共{count}个):\n"
output += "告警ID\t名称\t是否启用\t是否通知\t命名空间\t告警类型\n"
for alarm in alarms:
alarm_id = getattr(alarm, 'alarm_id', '')
name = getattr(alarm, 'name', '')
enabled = getattr(alarm, 'enabled', False)
notification_enabled = getattr(alarm, 'notification_enabled', False)
namespace = getattr(alarm, 'namespace', '')
alarm_type = getattr(alarm, 'type', '')
output += f"{alarm_id}\t{name}\t{'启用' if enabled else '禁用'}\t{'启用' if notification_enabled else '禁用'}\t{namespace}\t{alarm_type}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_rules 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListAlarmTemplateAssociationAlarmsRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警模板关联的告警规则列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_template_id", type=str, required=True, help="告警模板ID,以at开头,后跟字母、数字,长度2-64,可通过 list_alarm_templates.py 获取")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,范围1-100,默认100")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAlarmTemplateAssociationAlarmsRequest()
request.template_id = args.alarm_template_id
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_template_association_alarms(request)
alarms = getattr(response, 'alarms', []) or []
count = getattr(response, 'count', 0)
if not alarms:
print("查询结果为空")
exit(0)
output = f"告警模板 {args.alarm_template_id} 关联的告警规则列表(共{count}个):\n"
output += "告警ID\t名称\t描述\n"
for alarm in alarms:
alarm_id = getattr(alarm, 'alarm_id', '')
name = getattr(alarm, 'name', '')
description = getattr(alarm, 'description', '')
output += f"{alarm_id}\t{name}\t{description}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_template_association_alarms 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListAlarmTemplatesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警模板列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,范围1-100,默认100")
parser.add_argument("--namespace", type=str, help="服务命名空间,格式service.item,长度3-32")
parser.add_argument("--dim_name", type=str, help="资源维度名称,多维度用逗号分隔,每个维度最大长度32,总长度1-131")
parser.add_argument("--template_type", type=str, help="模板类型,枚举值:system/custom/system_event/custom_event/system_custom_event")
parser.add_argument("--template_name", type=str, help="告警模板名称,以字母或汉字开头,长度1-128,支持模糊匹配")
parser.add_argument("--product_name", type=str, help="产品层级跨维规则产品名称,如SYS.ECS,instance_id,长度0-128")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListAlarmTemplatesRequest()
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
if args.namespace:
request.namespace = args.namespace
if args.dim_name:
request.dim_name = args.dim_name
if args.template_type:
request.template_type = args.template_type
if args.template_name:
request.template_name = args.template_name
if args.product_name:
request.product_name = args.product_name
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_alarm_templates(request)
alarm_templates = getattr(response, 'alarm_templates', []) or []
count = getattr(response, 'count', 0)
if not alarm_templates:
print("查询结果为空")
exit(0)
output = f"告警模板列表(共{count}个):\n"
output += "模板ID\t名称\t类型\t创建时间\t描述\n"
for template in alarm_templates:
template_id = getattr(template, 'template_id', '')
template_name = getattr(template, 'template_name', '')
template_type = getattr(template, 'template_type', '')
create_time = getattr(template, 'create_time', '')
template_description = getattr(template, 'template_description', '')
output += f"{template_id}\t{template_name}\t{template_type}\t{create_time}\t{template_description}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_alarm_templates 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListCesTargetProjectTagsRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询CES目标项目标签列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--resource_type", type=str, default="CES-alarm", help="资源类型,固定为CES-alarm(告警规则),长度1-32")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListCesTargetProjectTagsRequest()
request.resource_type = args.resource_type
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_ces_target_project_tags(request)
tags = getattr(response, 'tags', []) or []
if not tags:
print("查询结果为空")
exit(0)
output = "CES目标项目标签列表:\n"
output += "键\t值列表\n"
for tag in tags:
key = getattr(tag, 'key', '')
values = getattr(tag, 'values', []) or []
output += f"{key}\t{','.join(str(v) for v in values)}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_ces_target_project_tags 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListDashboardInfosRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询仪表盘列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--enterprise_id", type=str, help="企业项目ID,长度36,也可为0(默认企业项目)或all_granted_eps(所有企业项目)")
parser.add_argument("--is_favorite", type=str, help="监控看板是否收藏,枚举值:true(收藏)/false(未收藏)。填此参数时enterprise_id必填")
parser.add_argument("--dashboard_name", type=str, help="监控看板名称(支持模糊匹配),长度1-128,只允许中文、英文、数字0-9、_和-")
parser.add_argument("--dashboard_id", type=str, help="监控看板ID,以db开头,包含22个字母和数字,长度24")
parser.add_argument("--dashboard_type", type=str, help="监控看板类型,枚举值:monitor_dashboard(监控大盘)/other(自定义看板)")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListDashboardInfosRequest()
if args.enterprise_id:
request.enterprise_id = args.enterprise_id
if args.is_favorite:
request.is_favorite = args.is_favorite.lower() == 'true'
if args.dashboard_name:
request.dashboard_name = args.dashboard_name
if args.dashboard_id:
request.dashboard_id = args.dashboard_id
if args.dashboard_type:
request.dashboard_type = args.dashboard_type
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_dashboard_infos(request)
dashboards = getattr(response, 'dashboards', []) or []
if not dashboards:
print("查询结果为空")
exit(0)
output = f"仪表盘列表(共{len(dashboards)}个):\n"
output += "仪表盘ID\t名称\t是否收藏\t企业项目ID\t创建时间\n"
for dashboard in dashboards:
dashboard_id = getattr(dashboard, 'dashboard_id', '')
dashboard_name = getattr(dashboard, 'dashboard_name', '')
is_favorite = getattr(dashboard, 'is_favorite', False)
enterprise_id = getattr(dashboard, 'enterprise_id', '')
create_time = getattr(dashboard, 'create_time', '')
output += f"{dashboard_id}\t{dashboard_name}\t{'是' if is_favorite else '否'}\t{enterprise_id}\t{create_time}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_dashboard_infos 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListDashboardWidgetsRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询仪表盘组件列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--dashboard_id", type=str, required=True, help="监控看板ID,以db开头,包含22个字母和数字,长度24,可通过 list_dashboard_infos.py 获取")
parser.add_argument("--group_id", type=str, help="视图分组ID,以dg开头包含22个字母和数字长度24,或default(不分组)")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListDashboardWidgetsRequest()
request.dashboard_id = args.dashboard_id
if args.group_id:
request.group_id = args.group_id
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_dashboard_widgets(request)
widgets = getattr(response, 'widgets', []) or []
if not widgets:
print("查询结果为空")
exit(0)
output = f"监控看板 {args.dashboard_id} 的视图列表:\n"
output += "视图ID\t标题\t图表类型\t分组ID\t单位\n"
for widget in widgets:
widget_id = getattr(widget, 'widget_id', '')
title = getattr(widget, 'title', '')
view = getattr(widget, 'view', '')
group_id = getattr(widget, 'group_id', '')
unit = getattr(widget, 'unit', '')
output += f"{widget_id}\t{title}\t{view}\t{group_id}\t{unit}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_dashboard_widgets 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListNotificationMaskResourcesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询通知屏蔽规则关联的资源列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--notification_mask_id", type=str, required=True, help="屏蔽规则ID,只能包含字母和数字,长度1-64,可通过 list_notification_masks.py 获取")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,范围1-100,默认100")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListNotificationMaskResourcesRequest()
request.notification_mask_id = args.notification_mask_id
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_notification_mask_resources(request)
resources = getattr(response, 'resources', []) or []
count = getattr(response, 'count', 0)
if not resources:
print("查询结果为空")
exit(0)
output = f"屏蔽规则 {args.notification_mask_id} 关联的资源列表(共{count}个):\n"
for idx, resource in enumerate(resources, 1):
namespace = getattr(resource, 'namespace', '')
dimensions = getattr(resource, 'dimensions', []) or []
output += f"资源 {idx} (命名空间: {namespace}):\n"
for dim in dimensions:
name = getattr(dim, 'name', '')
value = getattr(dim, 'value', '')
output += f" {name}: {value}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_notification_mask_resources 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListNotificationMasksRequest, ListNotificationMaskRequestBody, ListRelationType
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询通知屏蔽规则列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,范围1-100,默认100")
parser.add_argument("--sort_key", type=str, help="排序关键字,与sort_dir同时使用,枚举值:create_time(按创建时间)/update_time(按修改时间)")
parser.add_argument("--sort_dir", type=str, help="排序顺序,与sort_key同时使用,枚举值:DESC(降序)/ASC(升序)")
parser.add_argument("--mask_id", type=str, help="屏蔽规则ID,以nm开头,后跟0-62位字母或数字")
parser.add_argument("--mask_name", type=str, help="屏蔽规则名称,字母/数字/汉字/-/_,长度1-64")
parser.add_argument("--mask_status", type=str, help="屏蔽状态,枚举值:MASK_EFFECTIVE(已生效)/MASK_INEFFECTIVE(未生效)")
parser.add_argument("--relation_type", type=str, default="ALARM_RULE", help="关联类型,枚举值:ALARM_RULE/DEFAULT/RESOURCE/RESOURCE_POLICY_ALARM/RESOURCE_POLICY_NOTIFICATION,默认ALARM_RULE")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListNotificationMasksRequest()
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
if args.sort_key:
request.sort_key = args.sort_key
if args.sort_dir:
request.sort_dir = args.sort_dir
body = ListNotificationMaskRequestBody()
relation_type_map = {
"ALARM_RULE": ListRelationType.ALARM_RULE,
"DEFAULT": ListRelationType.DEFAULT,
"RESOURCE": ListRelationType.RESOURCE,
"RESOURCE_POLICY_ALARM": ListRelationType.RESOURCE_POLICY_ALARM,
"RESOURCE_POLICY_NOTIFICATION": ListRelationType.RESOURCE_POLICY_NOTIFICATION,
}
body.relation_type = relation_type_map.get(args.relation_type, ListRelationType.ALARM_RULE)
if args.mask_id:
body.mask_id = args.mask_id
if args.mask_name:
body.mask_name = args.mask_name
if args.mask_status:
body.mask_status = args.mask_status
request.body = body
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_notification_masks(request)
masks = getattr(response, 'notification_masks', []) or []
count = getattr(response, 'count', 0)
if not masks:
print("查询结果为空")
exit(0)
output = f"通知屏蔽规则列表(共{count}个):\n"
output += "屏蔽规则ID\t名称\t状态\t屏蔽类型\t起始日期\t起始时间\t截止日期\t截止时间\n"
for mask in masks:
notification_mask_id = getattr(mask, 'notification_mask_id', '')
mask_name = getattr(mask, 'mask_name', '')
mask_status = getattr(mask, 'mask_status', '')
mask_type = getattr(mask, 'mask_type', '')
start_date = getattr(mask, 'start_date', '')
start_time = getattr(mask, 'start_time', '')
end_date = getattr(mask, 'end_date', '')
end_time = getattr(mask, 'end_time', '')
output += f"{notification_mask_id}\t{mask_name}\t{mask_status}\t{mask_type}\t{start_date}\t{start_time}\t{end_date}\t{end_time}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_notification_masks 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListOneClickAlarmRulesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询一键告警规则列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--one_click_alarm_id", type=str, required=True, help="一键告警ID,以服务名+OneClickAlarm命名,可通过 list_one_click_alarms.py 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListOneClickAlarmRulesRequest()
request.one_click_alarm_id = args.one_click_alarm_id
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_one_click_alarm_rules(request)
alarms = getattr(response, 'alarms', []) or []
if not alarms:
print("查询结果为空")
exit(0)
output = f"一键告警 {args.one_click_alarm_id} 的告警规则列表:\n"
output += "告警ID\t名称\t是否启用\n"
for alarm in alarms:
alarm_id = getattr(alarm, 'alarm_id', '')
name = getattr(alarm, 'name', '')
enabled = getattr(alarm, 'enabled', False)
output += f"{alarm_id}\t{name}\t{'启用' if enabled else '禁用'}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_one_click_alarm_rules 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListOneClickAlarmsRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询一键告警列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListOneClickAlarmsRequest()
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_one_click_alarms(request)
one_click_alarms = getattr(response, 'one_click_alarms', []) or []
if not one_click_alarms:
print("查询结果为空")
exit(0)
output = "一键告警列表:\n"
output += "一键告警ID\t命名空间\t描述\t是否启用\n"
for alarm in one_click_alarms:
one_click_alarm_id = getattr(alarm, 'one_click_alarm_id', '')
namespace = getattr(alarm, 'namespace', '')
description = getattr(alarm, 'description', '')
enabled = getattr(alarm, 'enabled', False)
output += f"{one_click_alarm_id}\t{namespace}\t{description}\t{'启用' if enabled else '禁用'}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_one_click_alarms 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListResourceGroupsServicesResourcesRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询资源分组服务资源列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--group_id", type=str, required=True, help="资源分组ID,可通过 list_resource_groups.py 获取")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=str, help="分页大小,范围1-1000,默认1000")
parser.add_argument("--service", type=str, help="服务命名空间,格式service.item")
parser.add_argument("--dim_name", type=str, help="资源维度名称,如instance_id")
parser.add_argument("--status", type=str, help="资源状态,枚举值:HEALTHY/UNHEALTHY/NOT_EXIST")
parser.add_argument("--dim_value", type=str, help="资源维度值")
parser.add_argument("--tag", type=str, help="标签,格式key=value")
parser.add_argument("--extend_relation_id", type=str, help="关联编号")
parser.add_argument("--product_name", type=str, help="云产品名称,如SYS.ECS,instance_id")
parser.add_argument("--resource_name", type=str, help="资源名称")
parser.add_argument("--event_status", type=str, help="事件状态,枚举值:HEALTHY/UNHEALTHY/NOT_EXIST")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListResourceGroupsServicesResourcesRequest()
request.group_id = args.group_id
if args.offset is not None:
request.offset = args.offset
if args.limit:
request.limit = args.limit
if args.service:
request.service = args.service
if args.dim_name:
request.dim_name = args.dim_name
if args.status:
request.status = args.status
if args.dim_value:
request.dim_value = args.dim_value
if args.tag:
request.tag = args.tag
if args.extend_relation_id:
request.extend_relation_id = args.extend_relation_id
if args.product_name:
request.product_name = args.product_name
if args.resource_name:
request.resource_name = args.resource_name
if args.event_status:
request.event_status = args.event_status
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_resource_groups_services_resources(request)
resources = getattr(response, 'resources', []) or []
count = getattr(response, 'count', 0)
if not resources:
print("查询结果为空")
exit(0)
output = f"资源分组 {args.group_id} 的服务资源列表(共{count}个):\n"
for idx, resource in enumerate(resources, 1):
status = getattr(resource, 'status', '')
resource_name = getattr(resource, 'resource_name', '')
dimensions = getattr(resource, 'dimensions', []) or []
output += f"资源 {idx} (状态: {status}, 名称: {resource_name}):\n"
for dim in dimensions:
name = getattr(dim, 'name', '')
value = getattr(dim, 'value', '')
output += f" {name}: {value}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_resource_groups_services_resources 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ListResourceGroupsRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询资源分组列表")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--offset", type=int, help="分页偏移量,范围0-10000,默认0")
parser.add_argument("--limit", type=int, default=100, help="分页大小,范围1-100,默认100")
parser.add_argument("--group_name", type=str, help="资源分组名称,长度1-128")
parser.add_argument("--group_id", type=str, help="资源分组ID,以rg开头,长度24")
parser.add_argument("--enterprise_project_id", type=str, help="企业项目ID,长度36,也可为0或all_granted_eps,可通过 ../eps/list_enterprise_projects.py 获取")
parser.add_argument("--type", type=str, help="资源分组类型,枚举值:ECS/ELB/VPC/EVS/AS等")
parser.add_argument("--status", type=str, help="资源状态,枚举值:HEALTHY/UNHEALTHY/NOT_EXIST")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ListResourceGroupsRequest()
if args.offset is not None:
request.offset = args.offset
if args.limit is not None:
request.limit = args.limit
if args.group_name:
request.group_name = args.group_name
if args.group_id:
request.group_id = args.group_id
if args.enterprise_project_id:
request.enterprise_project_id = args.enterprise_project_id
if args.type:
request.type = args.type
if args.status:
request.status = args.status
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.list_resource_groups(request)
groups = getattr(response, 'resource_groups', []) or []
count = getattr(response, 'count', 0)
if not groups:
print("查询结果为空")
exit(0)
output = f"资源分组列表(共{count}个):\n"
output += "分组ID\t名称\t类型\t状态\n"
for group in groups:
group_id = getattr(group, 'group_id', '')
group_name = getattr(group, 'group_name', '')
type = getattr(group, 'type', '')
status = getattr(group, 'status', '')
output += f"{group_id}\t{group_name}\t{type}\t{status}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.list_resource_groups 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ShowAlarmTemplateRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询告警模板详情")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--alarm_template_id", type=str, required=True, help="告警模板ID,以at开头,可通过 list_alarm_templates.py 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ShowAlarmTemplateRequest()
request.template_id = args.alarm_template_id
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.show_alarm_template(request)
template_id = getattr(response, 'template_id', '')
template_name = getattr(response, 'template_name', '')
template_type = getattr(response, 'template_type', '')
create_time = getattr(response, 'create_time', '')
template_description = getattr(response, 'template_description', '')
policies = getattr(response, 'policies', []) or []
output = "告警模板详情:\n"
output += f"模板ID: {template_id}\n"
output += f"名称: {template_name}\n"
output += f"类型: {template_type}\n"
output += f"创建时间: {create_time}\n"
output += f"描述: {template_description}\n"
if policies:
output += "策略列表:\n"
for idx, policy in enumerate(policies, 1):
metric_name = getattr(policy, 'metric_name', '')
alarm_level = getattr(policy, 'alarm_level', '')
namespace = getattr(policy, 'namespace', '')
output += f" {idx}. {namespace} - {metric_name} - 级别: {alarm_level}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.show_alarm_template 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ShowResourceGroupRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询资源分组详情")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--group_id", type=str, required=True, help="资源分组ID,可通过 list_resource_groups.py 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ShowResourceGroupRequest()
request.group_id = args.group_id
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.show_resource_group(request)
group_id = getattr(response, 'group_id', '')
group_name = getattr(response, 'group_name', '')
type = getattr(response, 'type', '')
status = getattr(response, 'status', '')
create_time = getattr(response, 'create_time', '')
output = "资源分组详情:\n"
output += f"分组ID: {group_id}\n"
output += f"名称: {group_name}\n"
output += f"类型: {type}\n"
output += f"状态: {status}\n"
output += f"创建时间: {create_time}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.show_resource_group 查询失败: {e}")
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 huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkces.v2 import CesClient
from huaweicloudsdkces.v2.model import ShowWidgetRequest
from huaweicloudsdkces.v2.region.ces_region import CesRegion
AK, SK, Region, SecurityToken = load_credentials()
parser = argparse.ArgumentParser(description="查询仪表盘组件详情")
parser.add_argument("--region", type=str, default="cn-north-4", help="区域,默认 cn-north-4")
parser.add_argument("--widget_id", type=str, required=True, help="监控视图ID,以wg开头,长度24,可通过 list_dashboard_widgets.py 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
try:
request = ShowWidgetRequest()
request.widget_id = args.widget_id
http_config = build_http_config()
client = CesClient.new_builder().with_http_config(http_config).with_credentials(
BasicCredentials(AK, SK) if not SecurityToken else BasicCredentials(AK, SK).with_security_token(SecurityToken)).with_region(CesRegion.value_of(Region)).build()
if not client:
print("无法获取 Ces 客户端")
exit(-1)
response = client.show_widget(request)
widget_id = getattr(response, 'widget_id', '')
title = getattr(response, 'title', '')
view = getattr(response, 'view', '')
unit = getattr(response, 'unit', '')
output = "监控视图详情:\n"
output += f"视图ID: {widget_id}\n"
output += f"标题: {title}\n"
output += f"图表类型: {view}\n"
output += f"单位: {unit}\n"
print(output.strip())
exit(0)
except Exception as e:
print(f"ces.show_widget 查询失败: {e}")
exit(1)
<#
华为云资源查询 - 环境检查前置脚本 (Windows PowerShell)
#>
$ErrorActionPreference = "Stop"
Write-Output ""
Write-Output "=================================================="
Write-Output " 华为云资源查询 - 环境检查"
Write-Output "=================================================="
# 步骤 1:检查 Python 是否存在
Write-Output ""
Write-Output "[1/3] 检查 Python 环境"
$pythonPath = $null
foreach ($cmd in @("python3", "python")) {
$candidate = Get-Command $cmd -ErrorAction SilentlyContinue
if ($candidate) {
$testVer = (& $candidate.Path --version 2>&1 | Out-String).Trim()
if ($testVer -match "^Python \d+") {
$pythonPath = $candidate
break
}
}
}
if (-not $pythonPath) {
Write-Output "FAIL: Python 未安装或不可用"
Write-Output "环境未就绪,请先安装 Python 3.6+"
exit 1
}
Write-Output "OK: 找到 Python: $($pythonPath.Path)"
# 步骤 2:检查 Python 版本
Write-Output ""
Write-Output "[2/3] 检查 Python 版本"
$pythonVersion = (& $pythonPath.Path --version 2>&1 | Out-String).Trim()
if (-not $pythonVersion) {
Write-Output "FAIL: 无法获取 Python 版本"
exit 1
}
$versionMatch = [regex]::Match($pythonVersion, 'Python (\d+)\.(\d+)')
if (-not $versionMatch.Success) {
Write-Output "FAIL: 无法解析 Python 版本"
exit 1
}
$major = [int]$versionMatch.Groups[1].Value
$minor = [int]$versionMatch.Groups[2].Value
$versionStr = "$major.$minor"
Write-Output " 当前版本: $versionStr"
if ($major -lt 3 -or ($major -eq 3 -and $minor -lt 6)) {
Write-Output "FAIL: Python $versionStr `< 3.6"
Write-Output "环境未就绪,请升级 Python 到 3.6+"
exit 1
}
Write-Output "OK: Python $versionStr >= 3.6"
# 步骤 3:调用 Python 详细检查
Write-Output ""
Write-Output "[3/3] 执行详细检查"
Write-Output "--------------------------------------------------"
$scriptDir = Split-Path $MyInvocation.MyCommand.Definition -Parent
$pythonScript = Join-Path $scriptDir "ensure_env.py"
if (-not (Test-Path $pythonScript)) {
Write-Output "FAIL: 未找到检查脚本: $pythonScript"
exit 1
}
try {
& $pythonPath.Path $pythonScript
$exitCode = $LASTEXITCODE
} catch {
Write-Output "FAIL: Python 脚本执行异常: $_"
exit 1
}
Write-Output ""
Write-Output "=================================================="
if ($exitCode -eq 0) {
Write-Output "✓ 环境检查通过"
exit 0
} else {
Write-Output "✗ 环境未就绪,请按提示修复"
exit $exitCode
}
#!/bin/bash
# 华为云资源查询 - 环境检查前置脚本
# 步骤 1:检查 Python 是否存在
echo ""
echo "[1/3] 检查 Python 环境"
PYTHON_CMD=""
# 优先使用项目 .venv 中的 Python
if [ -n "${BASH_SOURCE:-}" ]; then
SCRIPT_PATH="${BASH_SOURCE[0]}"
else
SCRIPT_PATH="$0"
fi
SCRIPT_DIR=$(cd "$(dirname "$SCRIPT_PATH")" >/dev/null 2>&1 && pwd)
PROJECT_ROOT=$(cd "${SCRIPT_DIR}/.." >/dev/null 2>&1 && pwd)
VENV_PYTHON="${PROJECT_ROOT}/.venv/bin/python3"
if [ -f "$VENV_PYTHON" ]; then
PYTHON_CMD="$VENV_PYTHON"
else
for cmd in python3 python; do
if command -v "$cmd" &>/dev/null; then
test_ver=$("$cmd" --version 2>&1)
if echo "$test_ver" | grep -q "^Python [0-9]"; then
PYTHON_CMD="$cmd"
break
fi
fi
done
fi
if [ -z "$PYTHON_CMD" ]; then
echo "FAIL: 未检测到可用的 Python"
echo ""
echo "请先安装 Python 3.6+,根据当前系统参考:"
echo " macOS : brew install python"
echo " Ubuntu : sudo apt update && sudo apt install -y python3 python3-pip"
echo " CentOS : sudo yum install -y python3"
exit 1
fi
echo "OK: 找到 Python: ${PYTHON_CMD}"
# 步骤 2:检查 Python 版本
echo ""
echo "[2/3] 检查 Python 版本"
PYTHON_VERSION=$(${PYTHON_CMD} --version 2>&1 | awk '{print $2}')
if [ -z "${PYTHON_VERSION}" ]; then
echo "FAIL: 无法获取 Python 版本"
exit 1
fi
echo " 当前版本: ${PYTHON_VERSION}"
MAJOR=$(echo "${PYTHON_VERSION}" | cut -d'.' -f1)
MINOR=$(echo "${PYTHON_VERSION}" | cut -d'.' -f2)
if [ -z "${MAJOR}" ] || [ -z "${MINOR}" ]; then
echo "FAIL: 无法解析 Python 版本: ${PYTHON_VERSION}"
exit 1
fi
if [ "$MAJOR" -lt 3 ] || { [ "$MAJOR" -eq 3 ] && [ "$MINOR" -lt 6 ]; }; then
echo "FAIL: Python 版本过低 ($PYTHON_VERSION),需要 >= 3.6"
echo ""
echo "请升级 Python 后重试。"
exit 1
fi
echo "OK: Python ${PYTHON_VERSION} >= 3.6"
# 步骤 3:调用 Python 详细检查
echo ""
echo "[3/3] 执行详细检查"
echo "--------------------------------------------------"
PYTHON_SCRIPT="${SCRIPT_DIR}/ensure_env.py"
if [ ! -f "$PYTHON_SCRIPT" ]; then
echo "FAIL: 未找到检查脚本"
echo " 预期路径: $PYTHON_SCRIPT"
echo " 当前目录: $(pwd)"
exit 1
fi
"${PYTHON_CMD}" "${PYTHON_SCRIPT}"
EXIT_CODE=$?
echo ""
echo "=================================================="
if [ "${EXIT_CODE}" -eq 0 ]; then
echo "✓ 环境检查通过"
exit 0
else
echo "✗ 环境未就绪,请按提示修复"
exit 1
fi
import os
import urllib3
from urllib.parse import urlparse
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcore.http.http_config import HttpConfig
# 抑制因 ignore_ssl_verification 产生的 InsecureRequestWarning
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def load_credentials():
"""从环境变量加载华为云凭据
支持永久 AK/SK 和临时 AK/SK(含 SecurityToken)。
设置 HW_SECURITY_TOKEN 环境变量即可启用临时凭证模式。
"""
ak = os.getenv("HW_ACCESS_KEY", "")
sk = os.getenv("HW_SECRET_KEY", "")
region = os.getenv("HW_REGION_NAME", "cn-north-4")
security_token = os.getenv("HW_SECURITY_TOKEN", "")
if not ak or not sk:
print("未配置 AK/SK,请设置环境变量 HW_ACCESS_KEY 和 HW_SECRET_KEY")
exit(-1)
return ak, sk, region, security_token
def _get_proxy_url():
"""获取代理 URL,优先级: HTTPS_PROXY > HTTP_PROXY"""
proxy_url = os.getenv("HTTPS_PROXY", "")
if proxy_url:
return proxy_url
proxy_url = os.getenv("HTTP_PROXY", "")
if proxy_url:
return proxy_url
return ""
def build_http_config():
"""构建 HTTP 配置,代理支持环境变量
代理 URL 来源(优先级从高到低):
1. HTTPS_PROXY
2. HTTP_PROXY
代理 URL 格式:
- http://host:port
- http://user:pass@host:port
"""
http_config = HttpConfig.get_default_config()
http_config.ignore_ssl_verification = True
http_config.timeout = (30, 60)
http_config.retry_times = 3
proxy_url = _get_proxy_url()
if proxy_url:
parsed = urlparse(proxy_url)
http_config.proxy_protocol = parsed.scheme or "http"
http_config.proxy_host = parsed.hostname or ""
http_config.proxy_port = parsed.port or 8080
http_config.proxy_user = parsed.username or ""
http_config.proxy_password = parsed.password or ""
return http_config
#!/usr/bin/env python3
"""
华为云资源查询 - 环境确保脚本
一键完成所有环境准备并验证可用性,执行后即可直接运行查询脚本:
1. 校验必填环境变量 HW_ACCESS_KEY / HW_SECRET_KEY
2. 检查 Python 版本是否满足最低要求,若不满足则尝试自动安装
3. 安装 Python 依赖(支持 --upgrade)
4. 校验华为云 SDK 是否可导入
5. 校验凭据是否有效(真正调用 IAM API)
6. 校验项目级服务是否可用(真正调用 ECS API)
只有所有步骤都通过(包括 API 调用),才说明环境真正准备好了。
"""
import os
import sys
import subprocess
import platform
import tempfile
import argparse
import urllib.request
sys.path.insert(0, os.path.dirname(__file__))
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
# ── 虚拟环境管理 ──────────────────────────────────────────────────────
def _is_in_venv():
"""判断当前是否已在虚拟环境中"""
return sys.prefix != sys.base_prefix
def _venv_path():
"""返回项目根目录下的 .venv 路径"""
return os.path.join(get_project_root(), ".venv")
def _ensure_venv_and_reexec():
"""若不在 venv 中,创建 .venv 并用 venv Python 重新执行自身。
返回 True 表示已 reexec(不会返回到调用者),
返回 False 表示已在 venv 中,无需 reexec。
"""
if _is_in_venv():
return False
venv_dir = _venv_path()
venv_python = os.path.join(venv_dir, "bin", "python3")
if not os.path.isabs(venv_python):
venv_python = os.path.abspath(venv_python)
# 创建 venv(如不存在)
if not os.path.isfile(venv_python):
print("\n PEP 668: 系统 Python 禁止直接 pip install,正在创建虚拟环境...")
rc, out, err = run_cmd([sys.executable, "-m", "venv", venv_dir], timeout=60)
if rc != 0:
fail(f"创建虚拟环境失败: {err}")
print(" 请手动执行: python3 -m venv .venv && source .venv/bin/activate")
return False
ok(f"虚拟环境已创建: {venv_dir}")
# 用 venv Python 重新执行当前脚本
print(f" 使用虚拟环境 Python: {venv_python}")
os.execv(venv_python, [venv_python] + sys.argv)
def info(msg):
print(f" {msg}")
def ok(msg):
print(f" OK: {msg}")
def fail(msg):
print(f" FAIL: {msg}")
def get_project_id(region):
"""通过 IAM KeystoneListProjects 接口获取指定区域的项目 ID
IAM 是全局服务,不区分 region,但项目与区域一一对应。
查询项目列表后,按 region 名称匹配对应的项目 ID。
:param region: 区域名称,如 cn-north-4
:return: 项目 ID 字符串
"""
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
ak = os.getenv("HW_ACCESS_KEY", "")
sk = os.getenv("HW_SECRET_KEY", "")
security_token = os.getenv("HW_SECURITY_TOKEN", "")
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()
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}")
exit(-1)
# ── 工具函数 ──────────────────────────────────────────────────────────
def run_cmd(cmd, timeout=None, **kwargs):
"""运行命令,返回 (returncode, stdout, stderr)
Args:
cmd: 命令列表
timeout: 超时秒数,超时后终止子进程并返回失败
"""
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, **kwargs)
return result.returncode, result.stdout.strip(), result.stderr.strip()
except subprocess.TimeoutExpired:
return -1, "", f"命令超时({timeout}s): {' '.join(cmd)}"
except FileNotFoundError:
return -1, "", f"命令未找到: {cmd[0]}"
except Exception as e:
return -1, "", str(e)
def get_project_root():
"""获取项目根目录(scripts 的上级目录)"""
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# ── 步骤 1:校验环境变量 ─────────────────────────────────────────────
REQUIRED_ENV = ["HW_ACCESS_KEY", "HW_SECRET_KEY"]
def check_env_vars():
"""校验必填环境变量,只检查是否设置,绝不输出值"""
print("\n[1/5] 校验全局环境变量")
all_ok = True
for var in REQUIRED_ENV:
if os.getenv(var):
print(f" {var} 已设置")
else:
print(f" {var} 未设置")
all_ok = False
if not all_ok:
print("\n 必需的全局环境变量未设置,请配置 HW_ACCESS_KEY 和 HW_SECRET_KEY")
return False
print(" 全局环境变量校验通过")
return True
# ── 步骤 2:检查并安装 Python ────────────────────────────────────────
MIN_PYTHON = (3, 6)
def print_manual_python_help():
print()
info("请安装 Python 3.6+ 后重试,参考:")
print(" Windows : winget install Python.Python.3.11")
print(" Ubuntu : sudo apt update && sudo apt install -y python3 python3-pip python3-venv")
print(" CentOS : sudo yum install -y python3 python3-pip")
print(" Fedora : sudo dnf install -y python3 python3-pip")
print(" macOS : brew install python@3.11")
def check_python():
"""检查 Python 版本,若不满足则尝试自动安装"""
print("\n[2/5] 检查 Python 版本")
version = sys.version_info[:2]
version_str = f"{version[0]}.{version[1]}"
info(f"当前 Python: {version_str}")
if version >= MIN_PYTHON:
ok(f"Python {version_str} >= {MIN_PYTHON[0]}.{MIN_PYTHON[1]}")
return True
fail(f"Python {version_str} < {MIN_PYTHON[0]}.{MIN_PYTHON[1]}")
print_manual_python_help()
return False
def install_python():
"""尝试自动安装 Python 3.11"""
system = platform.system()
if system == "Windows":
rc, out, err = run_cmd(["winget", "install", "Python.Python.3.11", "--accept-source-agreements", "--accept-package-agreements"])
if rc == 0:
return True
print(f" winget 安装失败: {err}")
return False
elif system == "Linux":
# 尝试 apt(Ubuntu/Debian)
rc, _, err = run_cmd(["sudo", "apt", "update"])
rc2, _, err2 = run_cmd(["sudo", "apt", "install", "-y", "python3", "python3-pip", "python3-venv"])
if rc == 0 and rc2 == 0:
return True
# 尝试 yum(CentOS/RHEL/EulerOS 旧版)
rc, _, err = run_cmd(["sudo", "yum", "install", "-y", "python3", "python3-pip"])
if rc == 0:
return True
# 尝试 dnf(Fedora/EulerOS 新版)
rc, _, err = run_cmd(["sudo", "dnf", "install", "-y", "python3", "python3-pip"])
if rc == 0:
return True
print(" apt/yum/dnf 安装失败")
return False
elif system == "Darwin":
rc, _, err = run_cmd(["brew", "install", "python@3.11"])
if rc == 0:
return True
print(f" brew 安装失败: {err}")
return False
return False
# ── 步骤 3:安装依赖 ─────────────────────────────────────────────────
# 新增:支持 --upgrade 参数,用于升级已安装的包
UPGRADE = False # 全局标记,由命令行参数设置
def _check_deps_already_installed():
"""快速检查核心依赖是否已安装,避免不必要的 pip 调用"""
core_modules = ["huaweicloudsdkcore", "huaweicloudsdkecs", "huaweicloudsdkiam"]
for mod in core_modules:
try:
__import__(mod)
except ImportError:
return False
return True
def _ensure_pip():
"""确保 pip 可用,若不可用则尝试 ensurepip 引导安装"""
rc, _, _ = run_cmd([sys.executable, "-m", "pip", "--version"])
if rc == 0:
return True
print(" pip 不可用,尝试使用 ensurepip 引导安装...")
rc, _, err = run_cmd([sys.executable, "-m", "ensurepip", "--upgrade"])
if rc == 0:
print(" ensurepip 安装 pip 成功")
return True
print(f" ensurepip 也失败: {err}")
get_pip_path = os.path.join(tempfile.gettempdir(), "get-pip.py")
urls = [
"https://mirrors.huaweicloud.com/repository/pypi/simple/get-pip.py",
"https://bootstrap.pypa.io/get-pip.py",
]
ctx = ssl._create_unverified_context()
for url in urls:
info(f"尝试下载 get-pip.py: {url}")
try:
urllib.request.urlretrieve(url, get_pip_path, context=ctx)
except Exception as e:
print(f" 下载失败: {e}")
continue
rc, out, err = run_cmd([sys.executable, get_pip_path], timeout=120)
if rc == 0:
ok("get-pip.py 安装 pip 成功")
return True
print(" get-pip.py 执行失败")
for line in (err or out).split('\n')[-5:]:
print(f" {line}")
fail("无法安装 pip,请手动安装后重试")
return False
def _find_fastest_mirror():
"""轻量探测可用镜像源,返回第一个可达的 URL"""
mirrors = [
"https://repo.huaweicloud.com/repository/pypi/simple",
"https://pypi.tuna.tsinghua.edu.cn/simple",
"https://mirrors.aliyun.com/pypi/simple",
]
for url in mirrors:
try:
req = urllib.request.Request(url, method="HEAD")
resp = urllib.request.urlopen(req, timeout=5)
if resp.status == 200:
print(f" 可用镜像:{url}")
return url
except Exception:
continue
print(" 无可用镜像源,将尝试官方源")
return None
def install_dependencies():
"""安装 requirements.txt 中的依赖,优先使用国内镜像源"""
print("\n[3/5] 安装 Python 依赖")
root = get_project_root()
req_file = os.path.join(root, "requirements.txt")
if not os.path.exists(req_file):
print(f" 未找到 {req_file}")
return False
# 检查依赖是否已满足(非 --upgrade 模式可跳过安装)
if not UPGRADE and _check_deps_already_installed():
print(" 核心依赖已安装,跳过 pip 安装(使用 --upgrade 可强制重新安装)")
return True
# 确保 pip 可用
if not _ensure_pip():
return False
mirror_url = _find_fastest_mirror()
pip_timeout = "30"
pip_cmd = [sys.executable, "-m", "pip", "install", "-r", req_file]
# 新增:--upgrade 模式
if UPGRADE:
pip_cmd.append("--upgrade")
if mirror_url:
host = mirror_url.split("//")[1].split("/")[0]
rc, out, err = run_cmd(
pip_cmd + ["-i", mirror_url, "--trusted-host", host,
"--timeout", pip_timeout, "--retries", "2"],
timeout=120,
)
if rc == 0:
print(" 依赖安装成功")
return True
print(f" 镜像源安装失败,尝试官方源:{err[-200:]}")
# 兜底:官方源
rc, _, err = run_cmd(
pip_cmd + ["--timeout", pip_timeout, "--retries", "2"],
timeout=120,
)
if rc == 0:
print(" 依赖安装成功(官方源)")
return True
print(f" 依赖安装失败:{err[-300:]}")
return False
# ── 步骤 4:校验 SDK 可导入 ─────────────────────────────────────────
SDK_MODULES = [
"huaweicloudsdkcore",
"huaweicloudsdkiam",
"huaweicloudsdkces",
"huaweicloudsdkeps",
]
def check_sdk():
"""校验华为云 SDK 是否可正常导入"""
print("\n[4/5] 校验华为云 SDK")
all_ok = True
for mod in SDK_MODULES:
try:
__import__(mod)
print(f" {mod}")
except ImportError:
print(f" {mod} 不可导入")
all_ok = False
if all_ok:
print(" SDK 校验通过")
else:
print(" 部分 SDK 不可导入,请运行: pip install -r requirements.txt")
return all_ok
# ── 步骤 5:校验凭据有效性(真正调用 IAM API)──────────────────────
def check_credentials():
"""通过调用 IAM API 验证 AK/SK 凭据是否有效"""
print("\n[5/5] 校验凭据有效性(调用 IAM API)")
try:
from config import load_credentials, build_http_config
from huaweicloudsdkcore.auth.credentials import GlobalCredentials
from huaweicloudsdkiam.v5 import IamClient
from huaweicloudsdkiam.v5.region.iam_region import IamRegion
from huaweicloudsdkiam.v5.model import ListUsersV5Request
AK, SK, Region, SecurityToken = load_credentials()
http_config = build_http_config()
credentials = (
GlobalCredentials(AK, SK)
if not SecurityToken
else GlobalCredentials(AK, SK).with_security_token(SecurityToken)
)
client = (
IamClient.new_builder()
.with_http_config(http_config)
.with_credentials(credentials)
.with_region(IamRegion.value_of("cn-north-1"))
.build()
)
# 调用 IAM API:列出用户(限制 1 条,仅验证凭据有效)
request = ListUsersV5Request()
request.limit = 1
response = client.list_users_v5(request)
user_count = len(response.users) if response.users else 0
print(f" IAM 凭据有效,成功获取用户列表({user_count} 个用户)")
return True
except Exception as e:
print(f" IAM API 调用失败: {e}")
print(" 可能原因: AK/SK 无效、已过期、或网络不通")
return False
# ── 主流程 ──────────────────────────────────────────────────────────
TOTAL_STEPS = 5
def skip_step(step_num, label):
print(f"\n[{step_num}/{TOTAL_STEPS}] {label}")
print(" 跳过(前置步骤未通过)")
def parse_args():
parser = argparse.ArgumentParser(description="华为云资源查询 - 环境确保脚本")
parser.add_argument("--upgrade", action="store_true",
help="强制升级所有依赖到最新版本")
return parser.parse_args()
def main():
global UPGRADE
args = parse_args()
UPGRADE = args.upgrade
print("=" * 50)
print(" 华为云资源查询 - 环境确保")
print("=" * 50)
# 若系统 Python 受 PEP 668 保护,自动创建 venv 并 reexec
_ensure_venv_and_reexec()
results = []
# 步骤 1:环境变量
results.append(check_env_vars())
# 步骤 2:Python 版本
results.append(check_python())
# 步骤 3:依赖安装
if all(results):
results.append(install_dependencies())
else:
skip_step(3, "安装 Python 依赖")
results.append(False)
# 步骤 4:SDK 校验
if all(results):
results.append(check_sdk())
else:
skip_step(4, "校验华为云 SDK")
results.append(False)
# 步骤 5:凭据有效性
if all(results):
results.append(check_credentials())
else:
skip_step(5, "校验凭据有效性(调用 IAM API)")
results.append(False)
# 汇总
print()
print("=" * 50)
if all(results):
print(" 环境确保完成,所有检查项均通过(含 API 验证)")
print(" 可以正常使用查询脚本")
else:
print(" 环境确保未完成,请按上述提示修复后重新运行")
print("=" * 50)
return 0 if all(results) else 1
if __name__ == "__main__":
sys.exit(main())
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ListApiVersionsRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 常量
PAGE_SIZE = 50 # 每页展示条数
FETCH_SIZE = PAGE_SIZE + 1 # 多查1条用于判断是否还有更多
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目API版本列表")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--marker", type=str, help="分页标记,从上次查询结果的 next_marker 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 渲染
def render(versions, total_count=None, has_more=False, next_marker=None):
"""
渲染API版本列表
:param versions: 版本列表(最多 PAGE_SIZE 条)
:param total_count: 总数,None 表示未知
:param has_more: 是否还有更多数据
:param next_marker: 下一页的 marker
"""
if not versions:
print("没有找到API版本列表")
return
output = f"id\tstatus\tupdated\n"
for v in versions:
v_id = getattr(v, 'id', '')
status = getattr(v, 'status', '')
updated = getattr(v, 'updated', '')
output += f"{v_id}\t{status}\t{updated}\n"
# 汇总信息
showing_count = len(versions)
if total_count is not None:
output += f"\n共 {total_count} 条,当前返回 {showing_count} 条"
if has_more and next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页"
elif has_more:
output += f"\n当前返回 {showing_count} 条,还有更多数据"
if next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页"
else:
output += f"\n共 {showing_count} 条"
print(output)
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
# API 不支持分页(无 marker/limit/offset),一次返回所有数据,本地 marker 翻页
request = ListApiVersionsRequest()
response = client.list_api_versions(request)
versions = response.versions
if not versions:
print(f"没有找到API版本列表")
exit(0)
# 本地 marker 翻页:找到 marker 对应的位置,从该位置之后开始展示
start_idx = 0
if args.marker:
for i, item in enumerate(versions):
if str(getattr(item, 'id', '')) == args.marker:
start_idx = i + 1
break
remaining_items = versions[start_idx:]
if not remaining_items:
print("没有更多数据")
exit(0)
# 判断是否还有更多数据
has_more = len(remaining_items) > PAGE_SIZE
next_marker = None
if has_more:
next_marker = str(getattr(remaining_items[PAGE_SIZE - 1], 'id', ''))
display_items = remaining_items[:PAGE_SIZE]
# 渲染结果
render(display_items, total_count=len(versions), has_more=has_more, next_marker=next_marker)
except Exception as e:
print(f"eps.list_api_versions 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ListEnterpriseProjectRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 常量
PAGE_SIZE = 50 # 每页展示条数
FETCH_SIZE = PAGE_SIZE + 1 # 多查1条用于判断是否还有更多
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目列表")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--id", type=str, help="企业项目ID,0表示默认企业项目")
parser.add_argument("--name", type=str, help="企业项目名称,支持模糊搜索")
parser.add_argument("--status", type=int, choices=[1, 2], help="企业项目状态: 1(启用)/2(停用)")
parser.add_argument("--type", type=str, choices=["prod", "poc"], help="项目类型: prod(商用项目)/poc(测试项目)")
parser.add_argument("--sort_key", type=str, help="排序关键字,支持updated_at等,默认created_at")
parser.add_argument("--sort_dir", type=str, choices=["asc", "desc"], help="排序方向: asc(升序)/desc(降序),默认desc")
parser.add_argument("--marker", type=str, help="分页标记,从上次查询结果的 next_marker 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 渲染
def render(eps_list, total_count=None, has_more=False, next_marker=None):
"""
渲染企业项目列表
:param eps_list: 企业项目列表(最多 PAGE_SIZE 条)
:param total_count: 总数,None 表示未知
:param has_more: 是否还有更多数据
:param next_marker: 下一页的 marker
"""
if not eps_list:
print("没有找到企业项目")
return
output = f"id\tname\tstatus\ttype\tdescription\tcreated_at\tupdated_at\n"
for ep in eps_list:
ep_id = getattr(ep, 'id', '')
name = getattr(ep, 'name', '')
status = getattr(ep, 'status', '')
status_str = {1: '启用', 2: '停用'}.get(status, str(status))
ep_type = getattr(ep, 'type', '')
description = getattr(ep, 'description', '')
created_at = getattr(ep, 'created_at', '')
updated_at = getattr(ep, 'updated_at', '')
output += f"{ep_id}\t{name}\t{status_str}\t{ep_type}\t{description}\t{created_at}\t{updated_at}\n"
# 汇总信息
showing_count = len(eps_list)
if total_count is not None:
output += f"\n共 {total_count} 个企业项目,当前返回 {showing_count} 条"
if has_more and next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --id / --name / --status 等参数缩小查询范围"
elif has_more:
output += f"\n当前返回 {showing_count} 条,还有更多数据"
if next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --id / --name / --status 等参数缩小查询范围"
else:
output += f"\n共 {showing_count} 个企业项目"
print(output)
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
# API 使用 offset/limit 分页(无 marker),用 --marker 传递 offset 值实现翻页
request = ListEnterpriseProjectRequest()
request.limit = FETCH_SIZE
# marker 的值就是 offset(已跳过的条数)
request.offset = int(args.marker) if args.marker else 0
if args.id:
request.id = args.id
if args.name:
request.name = args.name
if args.status is not None:
request.status = args.status
if args.type:
request.type = args.type
if args.sort_key:
request.sort_key = args.sort_key
if args.sort_dir:
request.sort_dir = args.sort_dir
# 只做一次查询
response = client.list_enterprise_project(request)
total_count = getattr(response, 'total_count', None)
eps_list = response.enterprise_projects
if not eps_list:
print(f"没有找到企业项目 (区域: {Region})")
exit(0)
# 判断是否还有更多数据,计算 next_marker
next_marker = None
if total_count is not None:
has_more = total_count > (request.offset + PAGE_SIZE)
else:
# 无 total_count 时,通过多查的第 FETCH_SIZE 条判断
has_more = len(eps_list) > PAGE_SIZE
if has_more and len(eps_list) > PAGE_SIZE:
# 多查的1条存在,说明还有更多,next_marker 为当前 offset + PAGE_SIZE
next_marker = str(request.offset + PAGE_SIZE)
# 只展示前 PAGE_SIZE 条
display_eps = eps_list[:PAGE_SIZE]
# 渲染结果
render(display_eps, total_count=total_count, has_more=has_more, next_marker=next_marker)
except Exception as e:
print(f"eps.list_enterprise_projects 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ListMigrationRecordRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 常量
PAGE_SIZE = 50 # 每页展示条数
FETCH_SIZE = PAGE_SIZE + 1 # 多查1条用于判断是否还有更多
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目迁移记录列表")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--start_time", type=str, help="开始时间,格式: yyyy-MM-ddThh:mm:ssZ")
parser.add_argument("--end_time", type=str, help="结束时间,格式: yyyy-MM-ddThh:mm:ssZ")
parser.add_argument("--resource_id", type=str, help="资源ID,用于精确过滤")
parser.add_argument("--marker", type=str, help="分页标记,从上次查询结果的 next_marker 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 渲染
def render(records, total_count=None, has_more=False, next_marker=None):
"""
渲染迁移记录列表
:param records: 迁移记录列表(最多 PAGE_SIZE 条)
:param total_count: 总数,None 表示未知
:param has_more: 是否还有更多数据
:param next_marker: 下一页的 marker
"""
if not records:
print("没有找到迁移记录")
return
output = f"resource_id\tresource_name\tresource_type\toperate_type\torigin_ep_name\ttarget_ep_name\tevent_time\tstatus\n"
for r in records:
resource_id = getattr(r, 'resource_id', '')
resource_name = getattr(r, 'resource_name', '')
resource_type = getattr(r, 'resource_type', '')
operate_type = getattr(r, 'operate_type', '')
origin_ep_name = getattr(r, 'origin_ep_name', '')
target_ep_name = getattr(r, 'target_ep_name', '')
event_time = getattr(r, 'event_time', '')
associated = getattr(r, 'associated', None)
status = '成功' if associated is True else ('失败' if associated is False else str(associated))
output += f"{resource_id}\t{resource_name}\t{resource_type}\t{operate_type}\t{origin_ep_name}\t{target_ep_name}\t{event_time}\t{status}\n"
# 汇总信息
showing_count = len(records)
if total_count is not None:
output += f"\n共 {total_count} 条迁移记录,当前返回 {showing_count} 条"
if has_more and next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --start_time / --end_time / --resource_id 等参数缩小查询范围"
elif has_more:
output += f"\n当前返回 {showing_count} 条,还有更多数据"
if next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --start_time / --end_time / --resource_id 等参数缩小查询范围"
else:
output += f"\n共 {showing_count} 条迁移记录"
print(output)
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
# API 使用 offset/limit 分页(无 marker),用 --marker 传递 offset 值实现翻页
request = ListMigrationRecordRequest()
request.limit = FETCH_SIZE
# marker 的值就是 offset(已跳过的条数)
request.offset = str(int(args.marker) if args.marker else 0)
if args.start_time:
request.start_time = args.start_time
if args.end_time:
request.end_time = args.end_time
if args.resource_id:
request.resource_id = args.resource_id
# 只做一次查询
response = client.list_migration_record(request)
records = response.records
if not records:
print(f"没有找到迁移记录 (区域: {Region})")
exit(0)
# 判断是否还有更多数据,计算 next_marker
# Response 无 total_count/page_info,通过多查的第 FETCH_SIZE 条判断
next_marker = None
has_more = len(records) > PAGE_SIZE
if has_more:
# 多查的1条存在,说明还有更多,next_marker 为当前 offset + PAGE_SIZE
next_marker = str(int(args.marker) if args.marker else 0) + PAGE_SIZE
next_marker = str(next_marker)
# 只展示前 PAGE_SIZE 条
display_records = records[:PAGE_SIZE]
# 渲染结果
render(display_records, has_more=has_more, next_marker=next_marker)
except Exception as e:
print(f"eps.list_migration_records 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ListProvidersRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 常量
PAGE_SIZE = 50 # 每页展示条数
FETCH_SIZE = PAGE_SIZE + 1 # 多查1条用于判断是否还有更多
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目支持的云服务提供商列表")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--locale", type=str, help="指定显示语言,如 zh-cn/en-us")
parser.add_argument("--provider", type=str, help="云服务名称,用于精确过滤")
parser.add_argument("--marker", type=str, help="分页标记,从上次查询结果的 next_marker 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 渲染
def render(providers, total_count=None, has_more=False, next_marker=None):
"""
渲染云服务提供商列表
:param providers: 提供商列表(最多 PAGE_SIZE 条)
:param total_count: 总数,None 表示未知
:param has_more: 是否还有更多数据
:param next_marker: 下一页的 marker
"""
if not providers:
print("没有找到云服务提供商")
return
output = f"provider\tdisplay_name\tresource_types\n"
for p in providers:
provider = getattr(p, 'provider', '')
display_name = getattr(p, 'provider_i18n_display_name', '')
resource_types = getattr(p, 'resource_types', []) or []
type_str = ';'.join([getattr(rt, 'type', '') for rt in resource_types])
output += f"{provider}\t{display_name}\t{type_str}\n"
# 汇总信息
showing_count = len(providers)
if total_count is not None:
output += f"\n共 {total_count} 个云服务提供商,当前返回 {showing_count} 条"
if has_more and next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --provider 参数精确过滤"
elif has_more:
output += f"\n当前返回 {showing_count} 条,还有更多数据"
if next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页,或使用 --provider 参数精确过滤"
else:
output += f"\n共 {showing_count} 个云服务提供商"
print(output)
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
# API 使用 offset/limit 分页(无 marker),用 --marker 传递 offset 值实现翻页
request = ListProvidersRequest()
request.limit = FETCH_SIZE
# marker 的值就是 offset(已跳过的条数)
request.offset = int(args.marker) if args.marker else 0
if args.locale:
request.locale = args.locale
if args.provider:
request.provider = args.provider
# 只做一次查询
response = client.list_providers(request)
total_count = getattr(response, 'total_count', None)
providers = response.providers
if not providers:
print(f"没有找到云服务提供商 (区域: {Region})")
exit(0)
# 判断是否还有更多数据,计算 next_marker
next_marker = None
if total_count is not None:
has_more = total_count > (request.offset + PAGE_SIZE)
else:
# 无 total_count 时,通过多查的第 FETCH_SIZE 条判断
has_more = len(providers) > PAGE_SIZE
if has_more and len(providers) > PAGE_SIZE:
# 多查的1条存在,说明还有更多,next_marker 为当前 offset + PAGE_SIZE
next_marker = str(request.offset + PAGE_SIZE)
# 只展示前 PAGE_SIZE 条
display_providers = providers[:PAGE_SIZE]
# 渲染结果
render(display_providers, total_count=total_count, has_more=has_more, next_marker=next_marker)
except Exception as e:
print(f"eps.list_providers 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ListResourceMappingRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 常量
PAGE_SIZE = 50 # 每页展示条数
FETCH_SIZE = PAGE_SIZE + 1 # 多查1条用于判断是否还有更多
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询资源类型映射")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--marker", type=str, help="分页标记,从上次查询结果的 next_marker 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 渲染
def render(items, total_count=None, has_more=False, next_marker=None):
"""
渲染资源类型映射
:param items: 映射项列表(最多 PAGE_SIZE 条)
:param total_count: 总数,None 表示未知
:param has_more: 是否还有更多数据
:param next_marker: 下一页的 marker
"""
if not items:
print("没有找到资源类型映射")
return
output = f"key\tvalue\n"
for k, v in items:
output += f"{k}\t{v}\n"
# 汇总信息
showing_count = len(items)
if total_count is not None:
output += f"\n共 {total_count} 条,当前返回 {showing_count} 条"
if has_more and next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页"
elif has_more:
output += f"\n当前返回 {showing_count} 条,还有更多数据"
if next_marker:
output += f"\n可使用 --marker={next_marker} 继续查询下一页"
else:
output += f"\n共 {showing_count} 条"
print(output)
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
# API 不支持分页(无 marker/limit/offset),一次返回所有数据,本地 marker 翻页
request = ListResourceMappingRequest()
response = client.list_resource_mapping(request)
resource_mapping = getattr(response, 'resource_mapping', None)
if not resource_mapping:
print(f"没有找到资源类型映射")
exit(0)
# 将字典转为排序后的列表,便于分页
all_items = sorted(resource_mapping.items())
# 本地 marker 翻页:找到 marker 对应的位置,从该位置之后开始展示
start_idx = 0
if args.marker:
for i, (k, v) in enumerate(all_items):
if str(k) == args.marker:
start_idx = i + 1
break
remaining_items = all_items[start_idx:]
if not remaining_items:
print("没有更多数据")
exit(0)
# 判断是否还有更多数据
has_more = len(remaining_items) > PAGE_SIZE
next_marker = None
if has_more:
next_marker = str(remaining_items[PAGE_SIZE - 1][0])
display_items = remaining_items[:PAGE_SIZE]
# 渲染结果
render(display_items, total_count=len(all_items), has_more=has_more, next_marker=next_marker)
except Exception as e:
print(f"eps.list_resource_mapping 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ShowApiVersionRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目API版本详情")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--api_version", type=str, required=True, help="版本号,如 v1.0")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
request = ShowApiVersionRequest()
request.api_version = args.api_version
response = client.show_api_version(request)
version = response.version
if not version:
print(f"没有找到API版本 (版本号: {args.api_version})")
exit(0)
v_id = getattr(version, 'id', '')
status = getattr(version, 'status', '')
updated = getattr(version, 'updated', '')
links = getattr(version, 'links', []) or []
output = f"id:\t{v_id}\n"
output += f"status:\t{status}\n"
output += f"updated:\t{updated}\n"
if links:
link_str = '; '.join([f"{getattr(l, 'rel', '')}:{getattr(l, 'href', '')}" for l in links])
output += f"links:\t{link_str}"
print(output)
except Exception as e:
print(f"eps.show_api_version 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ShowAssociatedResourcesRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询资源关联的企业项目资源")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--project_id", type=str, required=True, help="项目ID,可通过 ../iam/get_project_id.py 获取")
parser.add_argument("--region_id", type=str, required=True, help="区域ID,如 cn-north-4")
parser.add_argument("--resource_id", type=str, required=True, help="资源ID")
parser.add_argument("--resource_type", type=str, required=True, help="资源类型,如 ecs, vpc, eip 等,可通过 list_providers.py 查询支持的类型")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
request = ShowAssociatedResourcesRequest()
request.project_id = args.project_id
request.region_id = args.region_id
request.resource_id = args.resource_id
request.resource_type = args.resource_type
response = client.show_associated_resources(request)
res_id = getattr(response, 'id', '')
res_name = getattr(response, 'name', '')
res_type = getattr(response, 'type', '')
associated = getattr(response, 'associated_resources', []) or []
errors = getattr(response, 'errors', []) or []
output = f"id:\t{res_id}\n"
output += f"name:\t{res_name}\n"
output += f"type:\t{res_type}\n"
if associated:
output += f"\nassociated_resources:\n"
for a in associated:
a_id = getattr(a, 'id', '')
a_name = getattr(a, 'name', '')
a_type = getattr(a, 'type', '')
output += f" {a_id}\t{a_name}\t{a_type}\n"
if errors:
output += f"\nerrors:\n"
for e in errors:
e_code = getattr(e, 'code', '')
e_msg = getattr(e, 'message', '')
output += f" {e_code}\t{e_msg}\n"
print(output.strip())
except Exception as e:
print(f"eps.show_associated_resources 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ShowEnterpriseProjectQuotaRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目配额")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
request = ShowEnterpriseProjectQuotaRequest()
response = client.show_enterprise_project_quota(request)
quotas = getattr(response, 'quotas', None)
if not quotas:
print(f"没有找到企业项目配额信息")
exit(0)
resources = getattr(quotas, 'resources', []) or []
if not resources:
print(f"没有找到企业项目配额资源")
exit(0)
output = f"type\tused\tquota\tlimit\n"
for r in resources:
r_type = getattr(r, 'type', '')
used = getattr(r, 'used', 0)
quota = getattr(r, 'quota', 0)
limit = getattr(r, 'limit', 0)
output += f"{r_type}\t{used}\t{quota}\t{limit}\n"
print(output.strip())
except Exception as e:
print(f"eps.show_enterprise_project_quota 查询失败: {e}")
exit(1)
import argparse
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 GlobalCredentials
from huaweicloudsdkeps.v1 import EpsClient
from huaweicloudsdkeps.v1.model import ShowEnterpriseProjectRequest
from huaweicloudsdkeps.v1.region.eps_region import EpsRegion
# 初始化凭据
AK, SK, Region, SecurityToken = load_credentials()
# 参数
parser = argparse.ArgumentParser(description="查询企业项目详情")
parser.add_argument("--region", type=str, help="区域,默认 cn-north-4")
parser.add_argument("--enterprise_project_id", type=str, required=True, help="企业项目ID,可通过 list_enterprise_projects.py 获取")
args = parser.parse_args()
if args.region is not None:
Region = args.region
# 使用 sdk
try:
http_config = build_http_config()
client = EpsClient.new_builder().with_http_config(http_config).with_credentials(
GlobalCredentials(AK, SK) if not SecurityToken else GlobalCredentials(AK, SK).with_security_token(SecurityToken)).with_region(EpsRegion.value_of(Region)).build()
if not client:
print(f"无法获取 EPS 客户端")
exit(-1)
request = ShowEnterpriseProjectRequest()
request.enterprise_project_id = args.enterprise_project_id
response = client.show_enterprise_project(request)
ep = response.enterprise_project
if not ep:
print(f"没有找到企业项目 (ID: {args.enterprise_project_id})")
exit(0)
ep_id = getattr(ep, 'id', '')
name = getattr(ep, 'name', '')
status = getattr(ep, 'status', '')
status_str = {1: '启用', 2: '停用'}.get(status, str(status))
ep_type = getattr(ep, 'type', '')
description = getattr(ep, 'description', '')
created_at = getattr(ep, 'created_at', '')
updated_at = getattr(ep, 'updated_at', '')
delete_flag = getattr(ep, 'delete_flag', False)
output = f"id:\t{ep_id}\n"
output += f"name:\t{name}\n"
output += f"status:\t{status_str}\n"
output += f"type:\t{ep_type}\n"
output += f"description:\t{description}\n"
output += f"created_at:\t{created_at}\n"
output += f"updated_at:\t{updated_at}\n"
output += f"delete_flag:\t{delete_flag}"
print(output)
except Exception as e:
print(f"eps.show_enterprise_project 查询失败: {e}")
exit(1)