Huawei Cloud Sac New Api

{
  "region": "cn-north-4",
  "access_key": "<YOUR_ACCESS_KEY>",
  "secret_key": "<YOUR_SECRET_KEY>",
  "ecs_password": "<YOUR_ECS_PASSWORD>"
}

#!/usr/bin/env python
"""Download a Terraform template file (.tf or .tf.json) to local directory."""

from __future__ import annotations

import argparse
import shutil
import sys
import urllib.parse
import urllib.request
from pathlib import Path


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Download Terraform template file from URL.")
    parser.add_argument("--url", required=True, help="Template URL (expected .tf or .tf.json)")
    parser.add_argument("--out-dir", required=True, help="Output directory (Terraform working directory)")
    parser.add_argument("--filename", default="", help="Optional target filename")
    parser.add_argument("--timeout", type=int, default=120, help="Download timeout in seconds")
    return parser.parse_args()


def guess_filename(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    name = Path(parsed.path).name.strip()
    if not name:
        return "main.tf"
    return name


def ensure_tf_name(name: str, url: str) -> str:
    if name.lower().endswith(".tf") or name.lower().endswith(".tf.json"):
        return name
    # Keep strict behavior to avoid silently saving non-template assets.
    raise RuntimeError(f"URL does not look like a Terraform template file: {url}")


def main() -> int:
    args = parse_args()
    out_dir = Path(args.out_dir).expanduser().resolve()
    out_dir.mkdir(parents=True, exist_ok=True)

    target_name = args.filename.strip() or guess_filename(args.url)
    target_name = ensure_tf_name(target_name, args.url)
    target = out_dir / target_name

    req = urllib.request.Request(
        args.url,
        headers={"User-Agent": "Mozilla/5.0 (compatible; find-and-deploy-sac/1.0)"},
    )

    try:
        with urllib.request.urlopen(req, timeout=args.timeout) as resp, target.open("wb") as f:
            shutil.copyfileobj(resp, f)
    except Exception as exc:  # noqa: BLE001
        print(f"Download failed: {exc}", file=sys.stderr)
        return 1

    print(f"Downloaded template: {target}")
    return 0


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

#!/usr/bin/env python
"""Extract SAC detail page price and deploy links via playwright-cli."""

import argparse
import json
import re
import sys
import time
from pathlib import Path

# Ensure sibling modules are importable regardless of CWD.
sys.path.insert(0, str(Path(__file__).resolve().parent))

from playwright_utils import build_pw_command, extract_marked_json, run_pw, run_pw_code

MARKER = "__SAC_JSON__"
TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "extract_sac_deploy_info.js"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Extract deploy links and estimated price from SAC detail page."
    )
    parser.add_argument("--url", required=True, help="SAC detail page URL")
    parser.add_argument("--out", default="", help="Optional output JSON path")
    parser.add_argument("--timeout-ms", type=int, default=60000)
    parser.add_argument("--session", default=f"sac-detail-{int(time.time() * 1000)}")
    parser.add_argument("--headed", action="store_true")
    return parser.parse_args()


def score_link(text: str, url: str) -> int:
    """Score a link by how likely it is a Terraform deploy/template download link.

    Chinese keywords (部署/模板/下载) match Huawei Cloud solution page UI text,
    which is primarily in Simplified Chinese.
    """
    signal = f"{text} {url}".lower()
    score = 0
    if "部署" in signal:  # Chinese: "deploy"
        score += 3
    if "terraform" in signal:
        score += 4
    if "template" in signal or "模板" in signal:  # Chinese: "template"
        score += 3
    if "download" in signal or "下载" in signal:  # Chinese: "download"
        score += 3
    if re.search(r"\.tf(\.json)?($|[?#])", url, re.IGNORECASE):
        score += 6
    if "aos" in signal or "stack" in signal:
        score += 1
    return score


def score_price_text(text: str) -> int:
    """Score a text candidate by how likely it contains pricing information.

    Chinese keywords (元/人民币/费用/成本规划 etc.) match Huawei Cloud
    solution page pricing text, which uses Simplified Chinese currency terms.
    """
    value = text or ""
    score = 0
    # Chinese: 元=CNY, 人民币=RMB, 至/到=range "to"
    if re.search(r"[0-9][0-9,.]*(?:\s*(?:~|-|至|到)\s*[0-9][0-9,.]*)?\s*(?:元|人民币|USD|美元)", value, re.IGNORECASE):
        score += 10
    if re.search(r"[0-9][0-9,.]*\s*\+\s*[^ ]*费用", value):  # 费用=cost/fee
        score += 4
    # Chinese: 资源和成本规划="resource and cost planning", 成本规划="cost planning"
    if "资源和成本规划" in value or "成本规划" in value:
        score += 1
    if re.search(r"表\d+", value):
        score += 1
    return score


def filter_out_reserved_price_lines(price_candidates: list[str]) -> list[str]:
    """Remove reserved/subscription pricing lines (包年包月=annual/monthly subscription)."""
    filtered = []
    for item in price_candidates:
        value = item or ""
        if re.search(r"包年包月|年付|月付|包月", value):  # Chinese: subscription terms
            continue
        filtered.append(value)
    return filtered


def filter_doc_fallback_on_demand_candidates(price_candidates: list[str]) -> list[str]:
    """When falling back to a cost documentation page, keep only on-demand pricing.

    Chinese: 按需计费/按需=on-demand/pay-as-you-go, 包年包月=subscription.
    """
    if not price_candidates:
        return []

    amount_re = re.compile(r"[0-9][0-9,]*(?:\.[0-9]+)?\s*(?:元|人民币|USD|美元)", re.IGNORECASE)
    state = "unknown"  # unknown | on_demand | reserved
    kept: list[str] = []

    for item in price_candidates:
        value = item or ""
        if re.search(r"按需计费|按需", value):
            state = "on_demand"
            kept.append(value)
            continue
        if re.search(r"包年包月|年付|月付|包月", value):
            state = "reserved"
            continue

        if amount_re.search(value):
            if state == "reserved":
                continue
            kept.append(value)
            continue

        # Keep non-amount metadata lines only before entering reserved section.
        if state != "reserved":
            kept.append(value)

    return filter_out_reserved_price_lines(kept)


def derive_hourly_price_text(
    estimated_price_text: str,
    price_candidates: list[str],
    hours_per_month: int = 730,
) -> str:
    """Convert monthly price to hourly estimate (÷730 hours/month).

    Only applies when on-demand pricing context (按需) is detected.
    """
    if not estimated_price_text or hours_per_month <= 0:
        return ""

    text = estimated_price_text.strip()
    if re.search(r"/\s*小时|每小时", text):
        return text

    on_demand_context = any("按需" in (item or "") for item in price_candidates)
    if not on_demand_context:
        return ""

    amount_match = re.search(
        r"([0-9][0-9,]*(?:\.[0-9]+)?)\s*(?:~|-|至|到)?\s*([0-9][0-9,]*(?:\.[0-9]+)?)?\s*(元|人民币|USD|美元)",
        text,
        re.IGNORECASE,
    )
    if not amount_match:
        return ""

    def _to_num(s: str) -> float:
        return float(s.replace(",", ""))

    lower = _to_num(amount_match.group(1))
    upper = _to_num(amount_match.group(2)) if amount_match.group(2) else None

    unit = "元/小时"
    if (amount_match.group(3) or "").lower() in ("usd", "美元"):
        unit = "USD/小时"

    suffix = text[amount_match.end() :].strip()
    if suffix.startswith("+"):
        suffix = suffix[1:].strip()
    suffix_text = f" + {suffix}（未折算）" if suffix else ""

    if upper is not None:
        hourly = f"{lower / hours_per_month:.4f}~{upper / hours_per_month:.4f}{unit}"
    else:
        hourly = f"{lower / hours_per_month:.4f}{unit}"

    return f"{hourly}{suffix_text}（按{hours_per_month}小时/月估算）"


def build_extract_script(timeout_ms: int) -> str:
    if not TEMPLATE_PATH.exists():
        raise RuntimeError(f"Missing JS template: {TEMPLATE_PATH}")
    script = TEMPLATE_PATH.read_text(encoding="utf-8")
    script = script.replace("__TIMEOUT__", str(timeout_ms))
    script = script.replace("__MARKER__", MARKER)
    return script


def main() -> int:
    args = parse_args()
    if args.timeout_ms <= 0:
        args.timeout_ms = 60000

    base_cmd = build_pw_command()

    open_args = ["open", args.url]
    if args.headed:
        open_args.append("--headed")
    run_pw(base_cmd, args.session, open_args)

    try:
        script = build_extract_script(args.timeout_ms)
        scrape = run_pw_code(base_cmd, args.session, script)
        extracted = extract_marked_json(f"{scrape.stdout}\n{scrape.stderr}", MARKER)
        price_candidates = extracted.get("price_text_candidates", [])
        price_source_url = extracted.get("price_source_url", extracted.get("page_url", ""))
        is_doc_fallback_price = bool(price_source_url and price_source_url != args.url)
        if is_doc_fallback_price:
            price_candidates = filter_doc_fallback_on_demand_candidates(price_candidates)
        best_price_text = ""
        if price_candidates:
            ranked_prices = sorted(price_candidates, key=score_price_text, reverse=True)
            best_price_text = ranked_prices[0]
        hourly_price_text = derive_hourly_price_text(best_price_text, price_candidates, 730)
        if is_doc_fallback_price and hourly_price_text:
            # For doc-page fallback, default to on-demand hourly display and hide monthly list prices.
            best_price_text = hourly_price_text

        deploy_links = extracted.get("deploy_links", [])
        deploy_links_ranked = sorted(
            [{**item, "score": score_link(item.get("text", ""), item.get("url", ""))} for item in deploy_links],
            key=lambda x: x.get("score", 0),
            reverse=True,
        )

        template_candidates = [
            item
            for item in deploy_links_ranked
            if re.search(r"terraform|template|模板|download|下载|\.tf(\.json)?($|[?#])", f"{item.get('text', '')} {item.get('url', '')}", re.IGNORECASE)
        ]
        tf_file_candidates = [
            item for item in deploy_links_ranked if re.search(r"\.tf(\.json)?($|[?#])", item.get("url", ""), re.IGNORECASE)
        ]

        result = {
            "selected_url": args.url,
            "session": args.session,
            "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "title": extracted.get("title", ""),
            "estimated_price_text": best_price_text,
            "price_text_candidates": price_candidates,
            "price_source_url": price_source_url,
            "deploy_links": deploy_links_ranked,
            "template_download_candidates": template_candidates,
            "tf_template_file_candidates": tf_file_candidates,
            "primary_tf_template_url": (tf_file_candidates[0]["url"] if tf_file_candidates else ""),
        }

        if args.out:
            out_path = Path(args.out).expanduser().resolve()
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

        print(f"Title: {result.get('title', '')}")
        if result.get("estimated_price_text"):
            print(f"Estimated Price: {result['estimated_price_text']}")
        else:
            print("Estimated Price: (not found)")
        print(f"Deploy links: {len(result.get('deploy_links', []))}")
        print(f"TF template files: {len(result.get('tf_template_file_candidates', []))}")
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0
    finally:
        run_pw(base_cmd, args.session, ["close"], allow_failure=True)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:  # noqa: BLE001
        print(f"extract_sac_deploy_info failed: {exc}", file=sys.stderr)
        raise SystemExit(1)

#!/usr/bin/env python
"""List Terraform variables and defaults from .tf / .tf.json files."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

# Ensure sibling modules are importable regardless of CWD.
sys.path.insert(0, str(Path(__file__).resolve().parent))

from sensitive_mask import mask_line


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="List Terraform variable defaults.")
    parser.add_argument("directory", help="Terraform template directory")
    return parser.parse_args()


def parse_tf_json_variables(path: Path) -> list[tuple[str, str]]:
    items: list[tuple[str, str]] = []
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:  # noqa: BLE001
        return items

    variable_block = data.get("variable")
    if not isinstance(variable_block, dict):
        return items

    for name, cfg in variable_block.items():
        default = "<NO_DEFAULT>"
        if isinstance(cfg, dict) and "default" in cfg:
            value = cfg["default"]
            if isinstance(value, str):
                default = value
            else:
                default = json.dumps(value, ensure_ascii=False)
        items.append((str(name), default))
    return items


def parse_tf_hcl_variables(path: Path) -> list[tuple[str, str]]:
    text = path.read_text(encoding="utf-8", errors="replace")
    items: list[tuple[str, str]] = []

    for m in re.finditer(r'variable\s+"([^"]+)"\s*\{', text):
        name = m.group(1)
        block_start = m.end() - 1  # points to "{"
        depth = 0
        end = -1
        for i in range(block_start, len(text)):
            ch = text[i]
            if ch == "{":
                depth += 1
            elif ch == "}":
                depth -= 1
                if depth == 0:
                    end = i
                    break
        if end < 0:
            continue

        block = text[block_start + 1 : end]
        default = "<NO_DEFAULT>"
        default_match = re.search(r"(?m)^\s*default\s*=\s*(.+?)\s*$", block)
        if default_match:
            default = default_match.group(1).strip()
        items.append((name, default))

    return items


def main() -> int:
    args = parse_args()
    root = Path(args.directory).expanduser().resolve()
    if not root.exists() or not root.is_dir():
        print(f"Directory does not exist: {root}", file=sys.stderr)
        return 1

    pairs: list[tuple[str, str, str]] = []
    for p in sorted(root.rglob("*.tf.json")):
        for name, default in parse_tf_json_variables(p):
            pairs.append((name, default, str(p)))
    for p in sorted(root.rglob("*.tf")):
        for name, default in parse_tf_hcl_variables(p):
            pairs.append((name, default, str(p)))

    if not pairs:
        print("No Terraform variables found.")
        return 2

    for name, default, src in pairs:
        # Mask sensitive values in output.
        output_line = mask_line(f"{name}={default}")
        print(output_line)
        print(f"  source: {src}")
    return 0


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

#!/usr/bin/env python
"""Normalize Terraform provider sources for HuaweiCloud SAC templates."""

import argparse
import json
import os
import re
import sys
from pathlib import Path
from typing import Any

# Ensure sibling modules are importable regardless of CWD.
sys.path.insert(0, str(Path(__file__).resolve().parent))

from sensitive_mask import mask_dict

HUAWEICLOUD_PATTERNS = [
    re.compile(r'source\s*=\s*"[^"]*huaweicloud[^"]*"', re.IGNORECASE),
]

KUBERNETES_PATTERNS = [
    re.compile(r'source\s*=\s*"[^"]*kubernetes[^"]*"', re.IGNORECASE),
]


def normalize_source_text(content: str) -> tuple[str, bool]:
    updated = content

    for pattern in HUAWEICLOUD_PATTERNS:
        updated = pattern.sub('source = "huaweicloud/huaweicloud"', updated)

    for pattern in KUBERNETES_PATTERNS:
        updated = pattern.sub('source = "hashicorp/kubernetes"', updated)

    return updated, updated != content


def find_block_end(text: str, open_brace_idx: int) -> int:
    depth = 0
    for i in range(open_brace_idx, len(text)):
        ch = text[i]
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return i
    return -1


def patch_huaweicloud_provider_block(block_body: str) -> tuple[str, bool]:
    changed = False
    updated = block_body

    replacements = {
        "access_key": "var.access_key",
        "secret_key": "var.secret_key",
        "region": "var.region",
    }

    for key, expr in replacements.items():
        pattern = re.compile(rf"(?m)^(\s*{key}\s*=\s*)(.+?)\s*$")
        match = pattern.search(updated)
        if match:
            current_value = match.group(2).strip()
            if current_value != expr:
                updated = pattern.sub(rf"\1{expr}", updated, count=1)
                changed = True
        else:
            # Default 2-space indentation for provider body attributes.
            line = f"  {key} = {expr}"
            if updated.strip():
                updated = updated.rstrip() + "\n" + line + "\n"
            else:
                updated = line + "\n"
            changed = True

    return updated, changed


def ensure_hcl_variables(content: str) -> tuple[str, bool]:
    changed = False
    updated = content.rstrip() + "\n"
    required_vars = ["access_key", "secret_key", "region"]

    for name in required_vars:
        if re.search(rf'(?m)\bvariable\s+"{re.escape(name)}"\s*\{{', updated):
            continue
        block = f'\nvariable "{name}" {{\n  type = string\n}}\n'
        updated += block
        changed = True

    return updated, changed


def patch_hcl_provider_credentials(content: str) -> tuple[str, bool]:
    changed = False
    updated = content

    pattern = re.compile(r'(?is)provider\s+"huaweicloud"\s*\{')
    cursor = 0
    chunks: list[str] = []

    while True:
        match = pattern.search(updated, cursor)
        if not match:
            chunks.append(updated[cursor:])
            break

        block_start = match.start()
        open_brace_idx = updated.find("{", match.start(), match.end())
        if open_brace_idx < 0:
            chunks.append(updated[cursor:])
            break
        close_brace_idx = find_block_end(updated, open_brace_idx)
        if close_brace_idx < 0:
            chunks.append(updated[cursor:])
            break

        chunks.append(updated[cursor : open_brace_idx + 1])
        body = updated[open_brace_idx + 1 : close_brace_idx]
        patched_body, body_changed = patch_huaweicloud_provider_block(body)
        chunks.append(patched_body)
        chunks.append("}")
        changed = changed or body_changed
        cursor = close_brace_idx + 1

    merged = "".join(chunks)
    merged, var_changed = ensure_hcl_variables(merged)
    return merged, changed or var_changed


def normalize_tf_json(content: str) -> tuple[str, bool]:
    data = json.loads(content)
    changed = False

    def normalize_required_providers(obj: Any) -> None:
        nonlocal changed
        if not isinstance(obj, dict):
            return

        terraform_block = obj.get("terraform")
        if not isinstance(terraform_block, dict):
            return

        required = terraform_block.get("required_providers")
        provider_maps = []
        if isinstance(required, dict):
            provider_maps = [required]
        elif isinstance(required, list):
            provider_maps = [entry for entry in required if isinstance(entry, dict)]
        else:
            return

        for provider_map in provider_maps:
            for provider_name, provider_cfg in provider_map.items():
                if not isinstance(provider_cfg, dict):
                    continue

                source_value = str(provider_cfg.get("source", "")).lower()
                provider_name_l = str(provider_name).lower()

                if "huaweicloud" in provider_name_l or "huaweicloud" in source_value:
                    if provider_cfg.get("source") != "huaweicloud/huaweicloud":
                        provider_cfg["source"] = "huaweicloud/huaweicloud"
                        changed = True
                    continue

                if "kubernetes" in provider_name_l or "kubernetes" in source_value:
                    if provider_cfg.get("source") != "hashicorp/kubernetes":
                        provider_cfg["source"] = "hashicorp/kubernetes"
                        changed = True

    if isinstance(data, dict):
        normalize_required_providers(data)

    def patch_provider_cfg(provider_cfg: dict) -> None:
        nonlocal changed
        mapping = {
            "access_key": "${var.access_key}",
            "secret_key": "${var.secret_key}",
            "region": "${var.region}",
        }
        for key, expr in mapping.items():
            if provider_cfg.get(key) != expr:
                provider_cfg[key] = expr
                changed = True

    def normalize_provider_blocks(obj: Any) -> None:
        if not isinstance(obj, dict):
            return
        provider = obj.get("provider")
        if isinstance(provider, dict):
            for name, cfg in provider.items():
                if "huaweicloud" not in str(name).lower():
                    continue
                if isinstance(cfg, dict):
                    patch_provider_cfg(cfg)
                elif isinstance(cfg, list):
                    for item in cfg:
                        if isinstance(item, dict):
                            patch_provider_cfg(item)
        elif isinstance(provider, list):
            for entry in provider:
                if not isinstance(entry, dict):
                    continue
                for name, cfg in entry.items():
                    if "huaweicloud" not in str(name).lower():
                        continue
                    if isinstance(cfg, dict):
                        patch_provider_cfg(cfg)
                    elif isinstance(cfg, list):
                        for item in cfg:
                            if isinstance(item, dict):
                                patch_provider_cfg(item)

    def ensure_json_variables(obj: Any) -> None:
        nonlocal changed
        if not isinstance(obj, dict):
            return
        var_block = obj.get("variable")
        if not isinstance(var_block, dict):
            var_block = {}
            obj["variable"] = var_block
            changed = True
        for name in ("access_key", "secret_key", "region"):
            if name not in var_block:
                var_block[name] = {"type": "string"}
                changed = True

    if isinstance(data, dict):
        normalize_provider_blocks(data)
        ensure_json_variables(data)

    updated = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
    return updated, changed


def process_file(tf_file: Path, dry_run: bool) -> bool:
    original = tf_file.read_text(encoding="utf-8")

    if tf_file.name.endswith(".tf.json"):
        updated, changed = normalize_tf_json(original)
    else:
        updated_sources, changed_sources = normalize_source_text(original)
        updated, changed_provider = patch_hcl_provider_credentials(updated_sources)
        changed = changed_sources or changed_provider

    if changed and not dry_run:
        tf_file.write_text(updated, encoding="utf-8")

    return changed


def write_credentials_tfvars(
    root: Path,
    ak: str,
    sk: str,
    region: str,
    out_file: str,
) -> Path:
    out_path = (root / out_file).resolve()
    tfvars = {
        "access_key": ak,
        "secret_key": sk,
        "region": region,
    }
    out_path.write_text(json.dumps(tfvars, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return out_path


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Normalize Terraform provider source for huaweicloud and kubernetes providers. "
            "Optionally write AK/SK/region into terraform.auto.tfvars.json."
        )
    )
    parser.add_argument("directory", help="Directory containing Terraform files")
    parser.add_argument("--dry-run", action="store_true", help="Show changes without writing")
    parser.add_argument("--region", default="", help="HuaweiCloud region, e.g. cn-north-4")
    args = parser.parse_args()

    root = Path(args.directory).expanduser().resolve()
    if not root.exists() or not root.is_dir():
        print(f"Directory does not exist: {root}", file=sys.stderr)
        return 1

    tf_files = sorted(root.rglob("*.tf")) + sorted(root.rglob("*.tf.json"))
    if not tf_files:
        print(f"No Terraform files found in: {root}")
        return 2

    changed_files = []
    for tf_file in tf_files:
        if process_file(tf_file, args.dry_run):
            changed_files.append(tf_file)

    print(f"Scanned {len(tf_files)} .tf files in {root}")
    print(f"Changed {len(changed_files)} files")
    for path in changed_files:
        print(f" - {path}")

    # Resolve credentials: ak/sk from env vars, region from CLI arg
    ak = os.environ.get("HW_ACCESS_KEY", "")
    sk = os.environ.get("HW_SECRET_KEY", "")
    region = args.region
    tfvars_path = (root / "terraform.auto.tfvars.json").resolve()

    if ak and sk:
        print("Loaded AK/SK from environment variables.")
    else:
        print(
            f"AK/SK not found in environment variables. "
            f"Please manually edit {tfvars_path} to add access_key and secret_key.",
            file=sys.stderr,
        )

    if args.dry_run:
        print("Dry-run enabled: skip writing credentials tfvars file.")
    else:
        out_path = write_credentials_tfvars(root, ak, sk, region, "terraform.auto.tfvars.json")
        # Mask sensitive values in summary output.
        masked = mask_dict({"access_key": ak, "secret_key": sk, "region": region})
        print(f"Wrote credentials tfvars: {out_path}")
        print(f"  Contents: {json.dumps(masked, ensure_ascii=False)}")
        print("Keep this file local and do not commit it to git.")

    return 0


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

#!/usr/bin/env python
"""Shared playwright-cli helpers for SAC scripts."""

from __future__ import annotations

import json
import re
import shutil
import subprocess
import tempfile
from pathlib import Path


def resolve_command(command: str) -> str | None:
    return shutil.which(command)


def build_pw_command() -> list[str]:
    pw = resolve_command("playwright-cli")
    if pw:
        return [pw]
    npx = resolve_command("npx")
    if npx:
        return [npx, "playwright-cli"]
    raise RuntimeError(
        "playwright-cli is not installed. Install with: npm install -g @playwright/cli@latest"
    )


def run_pw(
    base_cmd: list[str],
    session: str,
    args: list[str],
    allow_failure: bool = False,
) -> subprocess.CompletedProcess:
    full_cmd = [*base_cmd, f"-s={session}", *args]
    proc = subprocess.run(
        full_cmd,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        shell=False,
    )

    if not allow_failure and proc.returncode != 0:
        detail = (proc.stderr or proc.stdout).strip()
        raise RuntimeError(f"playwright-cli {' '.join(args)} failed (exit {proc.returncode}): {detail}")

    return proc


def cleanup_output(text: str) -> str:
    text = re.sub(r"\x1b\[[0-9;]*m", "", text)
    text = re.sub(r"^#\s?", "", text, flags=re.MULTILINE)
    return text.strip()


def run_pw_code(base_cmd: list[str], session: str, script: str) -> subprocess.CompletedProcess:
    script_path = ""
    try:
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".js", delete=False) as handle:
            handle.write(script)
            script_path = handle.name

        proc = run_pw(base_cmd, session, ["run-code", f"--filename={script_path}"], allow_failure=True)
        if proc.returncode == 0:
            return proc

        detail = cleanup_output(f"{proc.stdout}\n{proc.stderr}")
        # Backward compatibility for older playwright-cli versions without --filename.
        if re.search(r"(unknown|unexpected).*(--filename|filename)", detail, flags=re.IGNORECASE):
            return run_pw(base_cmd, session, ["run-code", script])

        raise RuntimeError(
            f"playwright-cli run-code failed (exit {proc.returncode}): {detail or 'unknown error'}"
        )
    finally:
        if script_path:
            try:
                Path(script_path).unlink(missing_ok=True)
            except Exception:  # noqa: BLE001
                pass


def extract_last_json_structure(text: str):
    decoder = json.JSONDecoder()
    best = None
    best_len = -1
    for idx, ch in enumerate(text):
        if ch not in '[{"':
            continue
        try:
            parsed, consumed = decoder.raw_decode(text[idx:])
        except Exception:  # noqa: BLE001
            continue
        if isinstance(parsed, str):
            try:
                reparsed = json.loads(parsed)
            except Exception:  # noqa: BLE001
                reparsed = None
            if isinstance(reparsed, (list, dict)):
                if consumed > best_len:
                    best = reparsed
                    best_len = consumed
                continue
        if isinstance(parsed, (list, dict)):
            if consumed > best_len:
                best = parsed
                best_len = consumed
    return best


def extract_marked_json(output: str, marker: str):
    cleaned = cleanup_output(output)
    quoted_chunks = re.findall(r'"(?:[^"\\]|\\.)*"', cleaned, flags=re.DOTALL)
    for chunk in reversed(quoted_chunks):
        try:
            decoded = json.loads(chunk)
        except Exception:  # noqa: BLE001
            continue
        if marker in decoded:
            payload = decoded.split(marker, 1)[1]
            try:
                return json.loads(payload)
            except Exception:  # noqa: BLE001
                continue

    idx = cleaned.rfind(marker)
    if idx < 0:
        parsed = extract_last_json_structure(cleaned)
        if parsed is not None:
            return parsed

        lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
        for line in reversed(lines):
            candidate = line
            if (
                (candidate.startswith('"') and candidate.endswith('"'))
                or (candidate.startswith("'") and candidate.endswith("'"))
            ):
                candidate = candidate[1:-1]
            try:
                decoded = json.loads(candidate)
            except Exception:  # noqa: BLE001
                continue
            if isinstance(decoded, str):
                try:
                    decoded = json.loads(decoded)
                except Exception:  # noqa: BLE001
                    pass
            if isinstance(decoded, (list, dict)):
                return decoded

        tail = "\n".join(lines[-8:])
        raise RuntimeError(
            f"Unable to find marker {marker} in playwright-cli output. Last output lines:\n{tail}"
        )

    after = cleaned[idx + len(marker) :].strip()
    first_line = after.splitlines()[0].strip() if after else ""

    if (first_line.startswith('"') and first_line.endswith('"')) or (
        first_line.startswith("'") and first_line.endswith("'")
    ):
        first_line = first_line[1:-1]

    return json.loads(first_line)

#!/usr/bin/env python
"""Mask sensitive values (AK, SK, passwords) in script output.

Provides a masking utility for Huawei Cloud SAC scripts.
Any output containing access_key, secret_key, or password values
must be masked to prevent credential leakage.
"""

from __future__ import annotations

import re
from typing import Any

# Keys whose values should always be masked in output.
SENSITIVE_KEY_PATTERNS = [
    re.compile(r"(?:access_key|secret_key|password|passwd|token|secret)", re.IGNORECASE),
]

# Minimum length before partial reveal (show last 4 chars).
REVEAL_THRESHOLD = 8


def mask_value(value: str, visible_tail: int = 4) -> str:
    """Mask a sensitive value, optionally revealing the last N characters.

    Args:
        value: The sensitive string to mask.
        visible_tail: Number of trailing characters to reveal (default 4).

    Returns:
        Masked string like "****abcd" or "***" for short values.
    """
    if not value:
        return "***"
    if len(value) >= REVEAL_THRESHOLD and visible_tail > 0:
        return f"****{value[-visible_tail:]}"
    return "***"


def is_sensitive_key(key: str) -> bool:
    """Check if a key name refers to a sensitive field."""
    return any(p.search(key) for p in SENSITIVE_KEY_PATTERNS)


def mask_dict(data: dict[str, Any], visible_tail: int = 4) -> dict[str, Any]:
    """Return a copy of the dict with sensitive values masked.

    Args:
        data: Dictionary that may contain sensitive key-value pairs.
        visible_tail: Number of trailing characters to reveal.

    Returns:
        New dict with sensitive values replaced by masked strings.
    """
    result = {}
    for key, value in data.items():
        if is_sensitive_key(key) and isinstance(value, str):
            result[key] = mask_value(value, visible_tail)
        else:
            result[key] = value
    return result


def mask_line(line: str, visible_tail: int = 4) -> str:
    """Mask sensitive key=value patterns in a single output line.

    Handles formats like:
        access_key=ABCDEF123456
        "access_key": "ABCDEF123456"
        access_key : ABCDEF123456

    Args:
        line: A single line of output text.
        visible_tail: Number of trailing characters to reveal.

    Returns:
        The line with sensitive values masked.
    """
    # Pattern: key = value  or  key: value  or  "key": "value"
    pattern = re.compile(
        r"""(
            (?:access_key|secret_key|password|passwd|token|secret)
            \s*(?:=|:)\s*
        )
        (?:
            "([^"]*)"
            |
            '([^']*)'
            |
            (\S+)
        )""",
        re.IGNORECASE | re.VERBOSE,
    )

    def _replace(m: re.Match) -> str:
        prefix = m.group(1)
        value = m.group(2) or m.group(3) or m.group(4) or ""
        return f"{prefix}{mask_value(value, visible_tail)}"

    return pattern.sub(_replace, line)

async (page) => {
  const timeoutMs = __TIMEOUT__;

  const clickByText = async (text) => {
    const locator = page.getByRole('button', { name: text }).first();
    if (await locator.count()) {
      try { await locator.click({ timeout: 800 }); } catch {}
    }
  };

  for (const label of ['知道了', '关闭', '我知道了', '同意', '稍后再说']) {
    await clickByText(label);
  }

  await page.waitForTimeout(Math.min(timeoutMs, 4000));
  // Best-effort scroll to trigger lazy-rendered sections (price cards may be below fold).
  try {
    await page.evaluate(() => window.scrollTo(0, document.body.scrollHeight));
    await page.waitForTimeout(700);
    await page.evaluate(() => window.scrollTo(0, 0));
  } catch {}

  const extractFromPage = async (targetPage, collectDeployLinks) =>
    await targetPage.evaluate((collectDeployLinks) => {
      const normalize = (s) => (s || '').replace(/\s+/g, ' ').trim();
      const title = normalize(
        document.querySelector('h1')?.textContent ||
          document.querySelector('h2')?.textContent ||
          document.title ||
          ''
      );

      const rawBodyText = document.body?.innerText || '';
      const bodyText = normalize(rawBodyText);
      const priceMatches = [];
      const amountPattern =
        /([¥￥]?\s*[0-9][0-9,.]*(?:\s*(?:~|-|至|到)\s*[0-9][0-9,.]*)?\s*(?:元|人民币|USD|美元)?(?:\s*\/\s*(?:小时|天|月|年|GB|TB|次))?(?:\s*\+\s*[^。；;\n]{1,40})?)/i;
      const patterns = [
        /(?:预估价格|参考价格|价格)[^。:：]{0,40}[:：]?\s*([¥￥]?\s*[0-9][0-9,.]*(?:\s*(?:~|-|至|到)\s*[0-9][0-9,.]*)?\s*(?:元|人民币|USD|美元)?(?:\s*\/\s*(?:小时|天|月|年|GB|TB|次))?(?:\s*\+\s*[^。；;\n]{1,40})?)/g,
        /(?:预估成本|参考成本|成本|花费|费用)[^。:：]{0,40}[:：]?\s*([¥￥]?\s*[0-9][0-9,.]*(?:\s*(?:~|-|至|到)\s*[0-9][0-9,.]*)?\s*(?:元|人民币|USD|美元)?(?:\s*\/\s*(?:小时|天|月|年|GB|TB|次))?(?:\s*\+\s*[^。；;\n]{1,40})?)/g,
      ];

      for (const p of patterns) {
        let m;
        while ((m = p.exec(bodyText)) !== null) {
          if (m[0]) priceMatches.push(m[0]);
        }
      }

      // Line-based fallback: catch formats like "每月预估花费：2~4元（按需计费...）"
      const lines = rawBodyText.split('\n').map((line) => normalize(line)).filter(Boolean);
      for (const line of lines) {
        if (!/(?:预估|参考|价格|成本|花费|费用|计费)/.test(line)) continue;
        const amount = line.match(amountPattern);
        if (amount) {
          priceMatches.push(line);
        }
      }

      const allLinks = [];
      const deployLinks = [];
      for (const el of Array.from(document.querySelectorAll('a[href], button'))) {
        const text = normalize(el.textContent || '');
        const href =
          el.tagName.toLowerCase() === 'a'
            ? el.getAttribute('href') || ''
            : el.closest('a')?.getAttribute('href') || '';
        if (!href) continue;

        const abs = href.startsWith('http') ? href : new URL(href, window.location.href).toString();
        const signal = `${text} ${abs}`;
        allLinks.push({ text, url: abs });
        const isTfTemplateFile = /\.tf(?:\.json)?(?:$|[?#])/i.test(abs);
        if (collectDeployLinks && (/部署|deploy|template|模板|下载|download|terraform/i.test(signal) || isTfTemplateFile)) {
          deployLinks.push({ text, url: abs });
        }
      }

      const costDocCandidates = allLinks.filter((item) =>
        /预估成本|成本规划|资源和成本|费用说明|计费说明/i.test(`${item.text} ${item.url}`) ||
        /support\.huaweicloud\.com\/.*(_02\.html|cost|price|billing)/i.test(item.url)
      );

      const dedupe = (arr) => {
        const map = new Map();
        for (const item of arr) {
          const key = `${item.url}|${item.text}`;
          if (!map.has(key)) map.set(key, item);
        }
        return Array.from(map.values());
      };

      return {
        title,
        page_url: window.location.href,
        price_text_candidates: Array.from(new Set(priceMatches)).slice(0, 20),
        deploy_links: dedupe(deployLinks),
        cost_doc_links: dedupe(costDocCandidates),
      };
    }, collectDeployLinks);

  const data = await extractFromPage(page, true);

  if (!(data.price_text_candidates || []).length) {
    const docUrl = (data.cost_doc_links || []).find((x) => /^https?:\/\//i.test(x.url))?.url || '';
    if (docUrl) {
      const docPage = await page.context().newPage();
      try {
        await docPage.goto(docUrl, { waitUntil: 'domcontentloaded', timeout: timeoutMs });
        await docPage.waitForTimeout(800);
        const docData = await extractFromPage(docPage, false);
        data.price_text_candidates = Array.from(
          new Set([...(data.price_text_candidates || []), ...(docData.price_text_candidates || [])])
        ).slice(0, 20);
        data.price_source_url = docData.page_url || docUrl;
      } catch {
        // Ignore doc-page errors; keep primary-page result.
      } finally {
        try { await docPage.close(); } catch {}
      }
    }
  }

  return '__MARKER__' + JSON.stringify(data);
}

{
  "region": "cn-north-4",
  "access_key": "",
  "secret_key": "",
  "ecs_password": ""
}