Alicloud Observability Pts

interface:
  display_name: "Alibaba Cloud Observability PTS"
  short_description: "PTS API discovery and safe operation workflows"
  default_prompt: "Use $alicloud-observability-pts to complete this observability/pts task on Alibaba Cloud."

#!/usr/bin/env python3
"""Shared PTS client helpers for local scripts."""

from __future__ import annotations

import os
from typing import Tuple


def _get_env(*names: str) -> str | None:
    for name in names:
        value = os.getenv(name)
        if value:
            return value
    return None


def create_client(region_id: str):
    """Return (pts_client, pts_models_module).

    Imports PTS SDK lazily so scripts can provide a clear install hint.
    """
    try:
        from alibabacloud_pts20201020.client import Client as PtsClient
        from alibabacloud_pts20201020 import models as pts_models
        from alibabacloud_tea_openapi import models as open_api_models
    except ImportError as exc:  # pragma: no cover - compile-safe fallback
        raise RuntimeError(
            "Missing SDK dependencies. Install with: "
            "python -m pip install -U alibabacloud_pts20201020 alibabacloud_tea_openapi"
        ) from exc

    ak = _get_env("ALICLOUD_ACCESS_KEY_ID", "ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = _get_env("ALICLOUD_ACCESS_KEY_SECRET", "ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = _get_env("ALICLOUD_SECURITY_TOKEN", "ALIBABA_CLOUD_SECURITY_TOKEN")

    if not ak or not sk:
        raise RuntimeError(
            "Missing credentials: set ALICLOUD_ACCESS_KEY_ID and "
            "ALICLOUD_ACCESS_KEY_SECRET (or ALIBABA_CLOUD_* equivalents)."
        )

    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"pts.{region_id}.aliyuncs.com",
    )
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token

    return PtsClient(config), pts_models

#!/usr/bin/env python3
"""Fetch PTS OpenAPI metadata API list and save artifacts to output/."""

from __future__ import annotations

import argparse
import json
import os
import pathlib
import urllib.request

DEFAULT_PRODUCT_CODE = "PTS"
DEFAULT_VERSION = "2020-10-20"
OUTPUT_DIR = pathlib.Path("output/alicloud-observability-pts")


def fetch_json(url: str, timeout: int) -> dict:
    req = urllib.request.Request(url, headers={"User-Agent": "codex-skill"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


def parse_api_names(payload: dict) -> list[str]:
    raw_apis = payload.get("apis", {})
    if isinstance(raw_apis, dict):
        return sorted(raw_apis.keys())
    if isinstance(raw_apis, list):
        names = []
        for item in raw_apis:
            if isinstance(item, dict):
                name = item.get("name") or item.get("apiName")
                if name:
                    names.append(name)
            elif isinstance(item, str):
                names.append(item)
        return sorted(set(names))
    return []


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--product-code", default=DEFAULT_PRODUCT_CODE)
    parser.add_argument("--version", default=DEFAULT_VERSION)
    parser.add_argument("--output-dir", default=str(OUTPUT_DIR))
    args = parser.parse_args()

    timeout = int(os.getenv("OPENAPI_META_TIMEOUT", "20"))
    output_dir = pathlib.Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    url = (
        f"https://api.aliyun.com/meta/v1/products/{args.product_code}"
        f"/versions/{args.version}/api-docs.json"
    )
    payload = fetch_json(url, timeout)
    api_names = parse_api_names(payload)

    json_file = output_dir / f"{args.product_code}_{args.version}_api_docs.json"
    md_file = output_dir / f"{args.product_code}_{args.version}_api_list.md"

    json_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    lines = [
        f"# {args.product_code} {args.version} API List",
        "",
        f"- Source: {url}",
        f"- API count: {len(api_names)}",
        "",
    ]
    lines.extend([f"- `{name}`" for name in api_names])
    md_file.write_text("\n".join(lines) + "\n", encoding="utf-8")

    print(f"Saved: {json_file}")
    print(f"Saved: {md_file}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""List PTS scenes in a region."""

from __future__ import annotations

import argparse
import json
import os

try:
    from ._pts_client import create_client
except ImportError:
    from _pts_client import create_client


def _render_table(scenes: list[dict], region: str, page_number: int, page_size: int) -> str:
    header = f"{'SceneId':<20} {'SceneName':<36} {'Status':<16} {'CreateTime'}"
    sep = "-" * len(header)
    lines = [header, sep]
    for scene in scenes:
        lines.append(
            f"{str(scene.get('SceneId', '-')):<20} "
            f"{str(scene.get('SceneName', '-')):<36} "
            f"{str(scene.get('Status', '-')):<16} "
            f"{str(scene.get('CreateTime', '-'))}"
        )
    lines.append(sep)
    lines.append(
        f"Total page items: {len(scenes)} (region={region}, page_number={page_number}, page_size={page_size})"
    )
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="List PTS scenes")
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--keyword", help="Search by scene name or scene id")
    parser.add_argument("--page-number", type=int, default=1)
    parser.add_argument("--page-size", type=int, default=10)
    parser.add_argument("--json", action="store_true", help="Output response body JSON")
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    client, pts_models = create_client(args.region)
    req = pts_models.ListPtsSceneRequest(
        key_word=args.keyword,
        page_number=args.page_number,
        page_size=args.page_size,
    )
    resp = client.list_pts_scene(req)
    body = resp.body.to_map() if resp.body else {}
    scenes = body.get("SceneViewList") or []

    if args.json:
        output = json.dumps(body, ensure_ascii=False, indent=2)
    else:
        output = _render_table(scenes, args.region, args.page_number, args.page_size)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output + ("\n" if not output.endswith("\n") else ""))
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


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

#!/usr/bin/env python3
"""Start a PTS scene and optionally poll running status."""

from __future__ import annotations

import argparse
import json
import os
import time

try:
    from ._pts_client import create_client
except ImportError:
    from _pts_client import create_client


RUNNING_STATES = {"RUNNING", "DEBUGGING"}


def main() -> int:
    parser = argparse.ArgumentParser(description="Start a PTS scene")
    parser.add_argument("--scene-id", required=True)
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--wait", action="store_true", help="Poll running status after start")
    parser.add_argument("--poll-interval", type=float, default=3.0)
    parser.add_argument("--max-wait", type=int, default=120)
    parser.add_argument("--output", help="Write JSON output to file")
    args = parser.parse_args()

    client, pts_models = create_client(args.region)
    start_resp = client.start_pts_scene(pts_models.StartPtsSceneRequest(scene_id=args.scene_id))
    start_body = start_resp.body.to_map() if start_resp.body else {}

    result: dict[str, object] = {
        "region": args.region,
        "scene_id": args.scene_id,
        "start_response": start_body,
    }

    if args.wait:
        deadline = time.time() + args.max_wait
        polls: list[dict] = []
        final_status = None

        while time.time() < deadline:
            status_resp = client.get_pts_scene_running_status(
                pts_models.GetPtsSceneRunningStatusRequest(scene_id=args.scene_id)
            )
            status_body = status_resp.body.to_map() if status_resp.body else {}
            polls.append(status_body)
            status = str(status_body.get("Status", "")).upper()
            final_status = status
            if status in RUNNING_STATES:
                break
            time.sleep(args.poll_interval)

        result["status_polling"] = {
            "timed_out": final_status not in RUNNING_STATES,
            "final_status": final_status,
            "attempts": len(polls),
            "polls": polls,
        }

    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)
        print(f"Output written to {args.output}")
    else:
        print(output)

    if args.wait:
        sp = result.get("status_polling", {})
        if isinstance(sp, dict) and sp.get("timed_out"):
            return 2
    return 0


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

#!/usr/bin/env python3
"""Stop a PTS scene and optionally poll until it is not running."""

from __future__ import annotations

import argparse
import json
import os
import time

try:
    from ._pts_client import create_client
except ImportError:
    from _pts_client import create_client


RUNNING_STATES = {"RUNNING", "DEBUGGING", "WAITSTART"}


def main() -> int:
    parser = argparse.ArgumentParser(description="Stop a PTS scene")
    parser.add_argument("--scene-id", required=True)
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--wait", action="store_true", help="Poll status after stop")
    parser.add_argument("--poll-interval", type=float, default=3.0)
    parser.add_argument("--max-wait", type=int, default=120)
    parser.add_argument("--output", help="Write JSON output to file")
    args = parser.parse_args()

    client, pts_models = create_client(args.region)
    stop_resp = client.stop_pts_scene(pts_models.StopPtsSceneRequest(scene_id=args.scene_id))
    stop_body = stop_resp.body.to_map() if stop_resp.body else {}

    result: dict[str, object] = {
        "region": args.region,
        "scene_id": args.scene_id,
        "stop_response": stop_body,
    }

    if args.wait:
        deadline = time.time() + args.max_wait
        polls: list[dict] = []
        final_status = None

        while time.time() < deadline:
            status_resp = client.get_pts_scene_running_status(
                pts_models.GetPtsSceneRunningStatusRequest(scene_id=args.scene_id)
            )
            status_body = status_resp.body.to_map() if status_resp.body else {}
            polls.append(status_body)
            status = str(status_body.get("Status", "")).upper()
            final_status = status
            if status and status not in RUNNING_STATES:
                break
            time.sleep(args.poll_interval)

        result["status_polling"] = {
            "timed_out": final_status in RUNNING_STATES or final_status is None,
            "final_status": final_status,
            "attempts": len(polls),
            "polls": polls,
        }

    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)
        print(f"Output written to {args.output}")
    else:
        print(output)

    if args.wait:
        sp = result.get("status_polling", {})
        if isinstance(sp, dict) and sp.get("timed_out"):
            return 2
    return 0


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