Alicloud Platform Devops

interface:
  display_name: "Alibaba Cloud Platform DevOps"
  short_description: "Yunxiao OpenAPI discovery and workflow guidance"
  default_prompt: "Use $alicloud-platform-devops to complete this platform/devops task on Alibaba Cloud."

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

from __future__ import annotations

import os


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 (devops_client, devops_models_module)."""
    try:
        from alibabacloud_devops20210625.client import Client as DevopsClient
        from alibabacloud_devops20210625 import models as devops_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_devops20210625 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)
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token

    return DevopsClient(config), devops_models

#!/usr/bin/env python3
"""Fetch DevOps 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 = "devops"
DEFAULT_VERSION = "2021-06-25"
OUTPUT_DIR = pathlib.Path("output/alicloud-platform-devops")


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 DevOps pipelines by organization."""

from __future__ import annotations

import argparse
import json
import os

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


def _to_table(pipelines: list[dict], org_id: str) -> str:
    header = f"{'PipelineId':<14} {'PipelineName':<42} {'GroupId':<12} {'CreatorAccountId'}"
    sep = "-" * len(header)
    lines = [header, sep]
    for p in pipelines:
        lines.append(
            f"{str(p.get('pipelineId', '-')):<14} "
            f"{str(p.get('pipelineName', '-')):<42} "
            f"{str(p.get('groupId', '-')):<12} "
            f"{str(p.get('creatorAccountId', '-'))}"
        )
    lines.append(sep)
    lines.append(f"Total: {len(pipelines)} pipeline(s) in organization {org_id}")
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="List DevOps pipelines")
    parser.add_argument("--organization-id", required=True)
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--pipeline-name")
    parser.add_argument("--status-list", help="comma-separated status values")
    parser.add_argument("--max-results", type=int, default=20)
    parser.add_argument("--next-token")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--output")
    args = parser.parse_args()

    client, devops_models = create_client(args.region)
    req = devops_models.ListPipelinesRequest(
        pipeline_name=args.pipeline_name,
        status_list=args.status_list,
        max_results=args.max_results,
        next_token=args.next_token,
    )
    resp = client.list_pipelines(args.organization_id, req)
    body = resp.body.to_map() if resp.body else {}
    pipelines = body.get("pipelines") or []

    output = json.dumps(body, ensure_ascii=False, indent=2) if args.json else _to_table(pipelines, args.organization_id)
    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
"""List DevOps projects by organization."""

from __future__ import annotations

import argparse
import json
import os

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


def _to_table(projects: list[dict], org_id: str) -> str:
    header = f"{'Identifier':<26} {'Name':<34} {'Type':<18} {'Creator'}"
    sep = "-" * len(header)
    lines = [header, sep]
    for p in projects:
        lines.append(
            f"{str(p.get('identifier', '-')):<26} "
            f"{str(p.get('name', '-')):<34} "
            f"{str(p.get('typeIdentifier', '-')):<18} "
            f"{str(p.get('creator', '-'))}"
        )
    lines.append(sep)
    lines.append(f"Total: {len(projects)} project(s) in organization {org_id}")
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="List DevOps projects")
    parser.add_argument("--organization-id", required=True)
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--category", help="Project category (required by API in most cases)")
    parser.add_argument("--scope")
    parser.add_argument("--max-results", type=int, default=20)
    parser.add_argument("--next-token")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--output")
    args = parser.parse_args()

    client, devops_models = create_client(args.region)
    req = devops_models.ListProjectsRequest(
        category=args.category or "BizOps",
        scope=args.scope,
        max_results=args.max_results,
        next_token=args.next_token,
    )
    resp = client.list_projects(args.organization_id, req)
    body = resp.body.to_map() if resp.body else {}
    projects = body.get("projects") or []

    output = json.dumps(body, ensure_ascii=False, indent=2) if args.json else _to_table(projects, args.organization_id)
    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
"""List DevOps code repositories by organization."""

from __future__ import annotations

import argparse
import json
import os

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


def _to_table(repositories: list[dict], org_id: str) -> str:
    header = f"{'Id':<12} {'Name':<32} {'PathWithNamespace':<42} {'Visibility':<10} {'UpdatedAt'}"
    sep = "-" * len(header)
    lines = [header, sep]
    for r in repositories:
        lines.append(
            f"{str(r.get('Id', '-')):<12} "
            f"{str(r.get('name', '-')):<32} "
            f"{str(r.get('pathWithNamespace', '-')):<42} "
            f"{str(r.get('visibilityLevel', '-')):<10} "
            f"{str(r.get('updatedAt', '-'))}"
        )
    lines.append(sep)
    lines.append(f"Total: {len(repositories)} repository item(s) in organization {org_id}")
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description="List DevOps repositories")
    parser.add_argument("--organization-id", required=True)
    parser.add_argument("--region", default=os.getenv("ALICLOUD_REGION_ID", "cn-hangzhou"))
    parser.add_argument("--page", type=int, default=1)
    parser.add_argument("--per-page", type=int, default=20)
    parser.add_argument("--search")
    parser.add_argument("--sort", choices=["asc", "desc"])
    parser.add_argument("--order-by")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--output")
    args = parser.parse_args()

    client, devops_models = create_client(args.region)
    req = devops_models.ListRepositoriesRequest(
        organization_id=args.organization_id,
        page=args.page,
        per_page=args.per_page,
        search=args.search,
        sort=args.sort,
        order_by=args.order_by,
    )
    resp = client.list_repositories(req)
    body = resp.body.to_map() if resp.body else {}
    repositories = body.get("result") or []

    output = json.dumps(body, ensure_ascii=False, indent=2) if args.json else _to_table(repositories, args.organization_id)
    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())