Aliyun Ecs Manage

interface:
  display_name: "Alibaba Cloud Compute ECS"
  short_description: "ECS lifecycle and inventory workflows"
  default_prompt: "Use $aliyun-ecs-manage to complete this compute/ecs task on Alibaba Cloud."

#!/usr/bin/env python3
"""List ECS instances across all regions.

Outputs TSV by default. Use --json for JSON output.
"""

from __future__ import annotations

import argparse
import json
import os
from typing import Iterable

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str) -> Iterable[ecs_models.DescribeInstancesResponseBodyInstancesInstance]:
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def to_record(region_id: str, inst) -> dict:
    return {
        "region_id": region_id,
        "instance_id": inst.instance_id,
        "instance_name": inst.instance_name,
        "status": inst.status,
        "instance_type": inst.instance_type,
        "cpu": inst.cpu,
        "memory_gib": inst.memory,
        "zone_id": inst.zone_id,
        "vpc_id": (inst.vpc_attributes.vpc_id if inst.vpc_attributes else None),
        "vswitch_id": (inst.vpc_attributes.v_switch_id if inst.vpc_attributes else None),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", action="store_true", help="Output JSON array")
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    records = []
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            records.append(to_record(region_id, inst))

    if args.json:
        output = json.dumps(records, ensure_ascii=False, indent=2)
    else:
        lines = [
            "region_id\tinstance_id\tinstance_name\tstatus\tinstance_type\tcpu\tmemory_gib\tzone_id\tvpc_id\tvswitch_id"
        ]
        for r in records:
            lines.append(
                "\t".join(
                    [
                        str(r.get("region_id") or ""),
                        str(r.get("instance_id") or ""),
                        str(r.get("instance_name") or ""),
                        str(r.get("status") or ""),
                        str(r.get("instance_type") or ""),
                        str(r.get("cpu") or ""),
                        str(r.get("memory_gib") or ""),
                        str(r.get("zone_id") or ""),
                        str(r.get("vpc_id") or ""),
                        str(r.get("vswitch_id") or ""),
                    ]
                )
            )
        output = "\n".join(lines)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Query ECS resource usage metrics for one instance via CloudMonitor (CMS).

Default namespace is `acs_ecs_dashboard`, with 5-minute period and 1-hour window.
Outputs JSON by default; use --summary-only for compact output.
"""

from __future__ import annotations

import argparse
import json
import os
from datetime import datetime, timedelta, timezone
from statistics import fmean
from typing import Any

from alibabacloud_cms20190101 import models as cms_models
from alibabacloud_cms20190101.client import Client as CmsClient
from alibabacloud_tea_openapi import models as open_api_models


DEFAULT_METRICS = [
    "CPUUtilization",
    "memory_usedutilization",
    "InternetInRate",
    "InternetOutRate",
    "IntranetInRate",
    "IntranetOutRate",
]


def create_client(region_id: str) -> CmsClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint="metrics.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return CmsClient(config)


def parse_datapoints(raw: str) -> list[dict[str, Any]]:
    try:
        data = json.loads(raw or "[]")
    except json.JSONDecodeError:
        return []
    if isinstance(data, dict):
        data = [data]
    if not isinstance(data, list):
        return []

    out = [x for x in data if isinstance(x, dict)]
    out.sort(key=lambda x: int(x.get("timestamp") or x.get("Timestamp") or 0))
    return out


def pick_value(point: dict[str, Any]) -> float | None:
    for key in ("Average", "Value", "Maximum", "Minimum"):
        val = point.get(key)
        if val is None:
            continue
        try:
            return float(val)
        except (TypeError, ValueError):
            continue
    return None


def summarize(points: list[dict[str, Any]]) -> dict[str, Any]:
    values = [v for v in (pick_value(p) for p in points) if v is not None]
    if not values:
        return {"points": 0}
    return {
        "points": len(values),
        "avg": round(fmean(values), 3),
        "max": round(max(values), 3),
        "min": round(min(values), 3),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--instance-id", required=True)
    parser.add_argument("--region-id", default=os.getenv("ALIBABACLOUD_REGION_ID") or os.getenv("ALIBABA_CLOUD_REGION_ID") or os.getenv("ALICLOUD_REGION_ID") or "cn-hangzhou")
    parser.add_argument("--hours", type=float, default=1.0, help="Lookback window in hours")
    parser.add_argument("--period", type=int, default=300, help="Metric period in seconds")
    parser.add_argument("--namespace", default="acs_ecs_dashboard")
    parser.add_argument(
        "--metrics",
        default=",".join(DEFAULT_METRICS),
        help="Comma-separated metric names",
    )
    parser.add_argument("--summary-only", action="store_true", help="Output summary only")
    parser.add_argument("--output", help="Write JSON output to file")
    args = parser.parse_args()

    end = datetime.now(timezone.utc).replace(microsecond=0)
    start = end - timedelta(hours=args.hours)
    start_time = start.isoformat().replace("+00:00", "Z")
    end_time = end.isoformat().replace("+00:00", "Z")

    metrics = [m.strip() for m in args.metrics.split(",") if m.strip()]
    client = create_client(args.region_id)

    result: dict[str, Any] = {
        "instance_id": args.instance_id,
        "region_id": args.region_id,
        "namespace": args.namespace,
        "window_utc": {
            "start": start_time,
            "end": end_time,
            "period_sec": args.period,
        },
        "metrics": {},
    }

    for metric in metrics:
        req = cms_models.DescribeMetricListRequest(
            namespace=args.namespace,
            metric_name=metric,
            dimensions=json.dumps({"instanceId": args.instance_id}),
            start_time=start_time,
            end_time=end_time,
            period=str(args.period),
        )

        try:
            resp = client.describe_metric_list(req)
            points = parse_datapoints(resp.body.datapoints or "[]")
            payload: dict[str, Any] = {"summary": summarize(points)}
            if not args.summary_only:
                payload["datapoints"] = points
            result["metrics"][metric] = payload
        except Exception as exc:  # noqa: BLE001
            result["metrics"][metric] = {"error": str(exc)}

    output = json.dumps(result, ensure_ascii=False, indent=2)
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Run a remote command on one ECS instance via Cloud Assistant (RunCommand).

This utility submits a command, polls invocation results, decodes stdout,
and outputs JSON for reproducible evidence.
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import time
from typing import Any

from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_tea_openapi import models as open_api_models


FINAL_STATES = {"Finished", "Failed", "Stopped", "Terminated", "Timeout"}


def create_client(region_id: str) -> EcsClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return EcsClient(config)


def safe_b64decode(content: str | None) -> str:
    if not content:
        return ""
    try:
        return base64.b64decode(content).decode("utf-8", errors="replace")
    except Exception:  # noqa: BLE001
        return ""


def to_result_payload(item: Any) -> dict[str, Any]:
    return {
        "invoke_record_status": getattr(item, "invoke_record_status", None),
        "exit_code": getattr(item, "exit_code", None),
        "finished_time": getattr(item, "finished_time", None),
        "start_time": getattr(item, "start_time", None),
        "instance_id": getattr(item, "instance_id", None),
        "stdout": safe_b64decode(getattr(item, "output", None)),
        "stderr": getattr(item, "error_info", None),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--instance-id", required=True)
    parser.add_argument("--command", required=True, help="Shell command text, e.g. 'ps -ef'")
    parser.add_argument("--region-id", default=os.getenv("ALIBABACLOUD_REGION_ID") or os.getenv("ALIBABA_CLOUD_REGION_ID") or os.getenv("ALICLOUD_REGION_ID") or "cn-hangzhou")
    parser.add_argument("--name", default="codex-remote-command")
    parser.add_argument("--command-type", default="RunShellScript", choices=["RunShellScript", "RunPowerShellScript"])
    parser.add_argument("--timeout", type=int, default=120, help="RunCommand timeout seconds")
    parser.add_argument("--poll-interval", type=float, default=2.0)
    parser.add_argument("--max-wait", type=int, default=180, help="Max polling wait seconds")
    parser.add_argument("--output", help="Write JSON output to file")
    args = parser.parse_args()

    client = create_client(args.region_id)

    run_req = ecs_models.RunCommandRequest(
        region_id=args.region_id,
        type=args.command_type,
        command_content=args.command,
        instance_id=[args.instance_id],
        name=args.name,
        timeout=args.timeout,
    )
    run_resp = client.run_command(run_req)
    invoke_id = run_resp.body.invoke_id

    poll_req = ecs_models.DescribeInvocationResultsRequest(
        region_id=args.region_id,
        invoke_id=invoke_id,
        instance_id=args.instance_id,
    )

    deadline = time.time() + args.max_wait
    final_payload: dict[str, Any] | None = None
    while time.time() < deadline:
        poll_resp = client.describe_invocation_results(poll_req)
        invocation = getattr(poll_resp.body, "invocation", None)
        inv_results = getattr(invocation, "invocation_results", None) if invocation else None
        items = getattr(inv_results, "invocation_result", None) if inv_results else None
        if items:
            item = items[0]
            status = getattr(item, "invoke_record_status", "")
            if status in FINAL_STATES:
                final_payload = to_result_payload(item)
                break
        time.sleep(args.poll_interval)

    result = {
        "region_id": args.region_id,
        "instance_id": args.instance_id,
        "invoke_id": invoke_id,
        "command": args.command,
        "command_type": args.command_type,
        "timed_out": final_payload is None,
        "result": final_payload,
    }

    output = json.dumps(result, ensure_ascii=False, indent=2)
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    if final_payload is None:
        return 2
    status = final_payload.get("invoke_record_status")
    return 0 if status == "Finished" else 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance specs across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            spec = f"{inst.instance_type} | {inst.cpu} vCPU | {inst.memory} GiB"
            counter[spec] += 1

    lines = []
    for spec, cnt in counter.most_common():
        lines.append(f"{cnt}\t{spec}")

    output = "\n".join(lines) if lines else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance counts by instance type across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            counter[inst.instance_type] += 1

    lines = ["instance_type\tcount"]
    for itype, cnt in sorted(counter.items()):
        lines.append(f"{itype}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance counts by region (and optionally by status)."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--by-status", action="store_true", help="Break down by instance status")
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            key = f"{region_id}\t{inst.status}" if args.by_status else region_id
            counter[key] += 1

    lines = []
    if args.by_status:
        lines.append("region_id\tstatus\tcount")
        for key, cnt in sorted(counter.items()):
            region_id, status = key.split("\t", 1)
            lines.append(f"{region_id}\t{status}\t{cnt}")
    else:
        lines.append("region_id\tcount")
        for region_id, cnt in sorted(counter.items()):
            lines.append(f"{region_id}\t{cnt}")

    output = "\n".join(lines) if lines else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance counts by security group across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            group_ids = []
            if inst.security_group_ids and inst.security_group_ids.security_group_id:
                group_ids = inst.security_group_ids.security_group_id
            if not group_ids:
                counter["(none)"] += 1
                continue
            for sg in group_ids:
                counter[sg] += 1

    lines = ["security_group_id\tcount"]
    for sg_id, cnt in sorted(counter.items()):
        lines.append(f"{sg_id}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance counts by status across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            counter[inst.status] += 1

    lines = ["status\tcount"]
    for status, cnt in sorted(counter.items()):
        lines.append(f"{status}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Summarize ECS instance counts by VPC across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Ecs20140526Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"ecs.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALIBABACLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALIBABACLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALIBABACLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN") or os.getenv("ALICLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return Ecs20140526Client(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.describe_regions(ecs_models.DescribeRegionsRequest())
    return [r.region_id for r in resp.body.regions.region]


def iter_instances(client: Ecs20140526Client, region_id: str):
    page_number = 1
    page_size = 100
    while True:
        resp = client.describe_instances(ecs_models.DescribeInstancesRequest(
            region_id=region_id,
            page_number=page_number,
            page_size=page_size,
        ))
        for inst in resp.body.instances.instance:
            yield inst
        total = resp.body.total_count
        if page_number * page_size >= total:
            break
        page_number += 1


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        client = create_client(region_id)
        for inst in iter_instances(client, region_id):
            vpc_id = None
            if inst.vpc_attributes:
                vpc_id = inst.vpc_attributes.vpc_id
            key = vpc_id or "(classic)"
            counter[key] += 1

    lines = ["vpc_id\tcount"]
    for vpc_id, cnt in sorted(counter.items()):
        lines.append(f"{vpc_id}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No ECS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())