Huawei Cloud Flexus L Server Scripts Excute

#!/usr/bin/env python3
# coding: utf-8
"""
Huawei Cloud COC Script Management CLI Tool

Manages COC scripts using Huawei Cloud SDK, supporting script creation and execution.

Parameters:
    --ak: Huawei Cloud Access Key (required)
    --sk: Huawei Cloud Secret Key (required)
    --security_token: Temporary security token (optional, for temporary credentials)
    --region: Region (default: cn-north-4)

Commands:
    create  - Create custom script (standalone flow)
    execute - Execute script (standalone flow)
    show    - View script details
    list    - List scripts

Each command has its own interactive flow
"""

import argparse
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from scripts.lib import create_script, execute_script, get_script_detail, list_scripts, get_script_job_batch

# Huawei Cloud global region mapping (for target instance region selection)
REGIONS = {
    # China mainland regions
    "cn-north-4": "North China - Beijing 4",
    "cn-north-9": "North China - Ulanqab 1",
    "cn-east-3": "East China - Shanghai 1",
    "cn-south-1": "South China - Guangzhou",
    "cn-south-4": "South China - Guangzhou (Friendly User Env)",
    "cn-southwest-2": "Southwest China - Guiyang 1",
    "cn-east-5": "East China - Qingdao",
    "cn-east-4": "East China 2",
    "cn-north-11": "North China - Ulanqab 201",
    "cn-north-12": "North China - Ulanqab 202",
    # International regions
    "ap-southeast-1": "Hong Kong, China",
    "ap-southeast-2": "Asia Pacific - Bangkok",
    "ap-southeast-3": "Asia Pacific - Singapore",
    "ap-southeast-4": "Asia Pacific - Jakarta",
    "ap-southeast-5": "Asia Pacific - Manila",
    "af-south-1": "Africa - Johannesburg",
    "af-north-1": "Africa - Cairo",
    "la-north-2": "Latin America - Mexico City 2",
    "sa-brazil-1": "Latin America - Sao Paulo 1",
    "la-south-2": "Latin America - Santiago",
    "tr-west-1": "Turkey - Istanbul",
    "me-east-1": "Middle East - Riyadh",
}

# Get all region IDs (for target instance regions)
REGION_IDS = list(REGIONS.keys())

# Dynamically get COC service supported regions from SDK
def get_coc_service_regions():
    """
    Dynamically get COC service supported regions and their name mapping from SDK.
    
    Returns:
        dict, key is region ID, value is region name
    """
    from huaweicloudsdkcoc.v1.region.coc_region import CocRegion
    
    # Region ID to name mapping
    region_name_map = {
        "cn-north-4": "North China - Beijing 4 (default)",
        "ap-southeast-3": "Asia Pacific - Singapore",
        "eu-west-101": "Europe - Frankfurt",
    }
    
    # Dynamically get regions defined in SDK
    sdk_regions = CocRegion.static_fields.keys()
    
    # Return dict containing only SDK supported regions
    return {region: region_name_map.get(region, region) for region in sdk_regions}


# Get COC service region IDs list
def get_coc_service_region_ids():
    """Get list of region IDs supported by COC service"""
    return list(get_coc_service_regions().keys())


def print_region_list():
    """
    Print Huawei Cloud global region list (for target instance region selection)
    """
    print("\nHuawei Cloud Global Regions (Target Instance Regions):")
    print("-" * 55)
    print(f"{'Region ID':<20} {'Region Name':<25}")
    print("-" * 55)
    
    # Print China mainland regions
    print("\n[China Mainland Regions]")
    for region_id, name in REGIONS.items():
        if region_id.startswith("cn-"):
            print(f"{region_id:<20} {name:<25}")
    
    # Print international regions
    print("\n[International Regions]")
    for region_id, name in REGIONS.items():
        if not region_id.startswith("cn-"):
            print(f"{region_id:<20} {name:<25}")
    print("-" * 55)


def print_coc_service_region_list():
    """
    Print COC service supported regions (dynamically fetched from SDK)
    """
    coc_regions = get_coc_service_regions()
    print("\nCOC Service Supported Regions:")
    print("-" * 40)
    print(f"{'Region ID':<20} {'Region Name':<20}")
    print("-" * 40)
    for region_id, name in coc_regions.items():
        print(f"{region_id:<20} {name:<20}")
    print("-" * 40)


def prompt_for_region(prompt, required=False, default="cn-north-4", is_coc_service=False):
    """
    Interactive prompt for user to select region, displays region list
    
    Args:
        prompt: Prompt message
        required: Whether it's required
        default: Default value
        is_coc_service: Whether it's COC service region selection (dynamically fetched from SDK)
    """
    if is_coc_service:
        regions = get_coc_service_regions()
        region_ids = get_coc_service_region_ids()
        print_func = print_coc_service_region_list
    else:
        regions = REGIONS
        region_ids = REGION_IDS
        print_func = print_region_list
    
    while True:
        # Display region list
        print_func()
        
        if default:
            display_prompt = f"\n{prompt} (default: {default} [{regions.get(default, '')}]) (enter 'list' to view list): "
        else:
            display_prompt = f"\n{prompt} (enter 'list' to view list): "
        
        user_input = input(display_prompt).strip()
        
        if user_input.lower() == 'list':
            continue
        
        if user_input:
            # Check if valid region ID
            if user_input in region_ids:
                return user_input
            # Try to find by name
            for region_id, name in regions.items():
                if user_input == name:
                    return region_id
            print(f"\nError: '{user_input}' is not a valid region ID or name. Please enter a valid region.")
            continue
        elif default:
            return default
        elif not required:
            return None
        else:
            print("Error: This parameter is required")


def prompt_for_input(prompt, required=True, default=None, choices=None, hide_input=False):
    """
    Interactive prompt for user input.
    """
    while True:
        if default:
            display_prompt = f"{prompt} (default: {default}): "
        elif choices:
            display_prompt = f"{prompt} ({'/'.join(choices)}): "
        else:
            display_prompt = f"{prompt}: "

        if hide_input:
            try:
                from getpass import getpass
                user_input = getpass(display_prompt).strip()
            except ImportError:
                user_input = input(display_prompt).strip()
        else:
            user_input = input(display_prompt).strip()

        if user_input:
            if choices and user_input not in choices:
                print(f"Error: Please select one of the following options: {', '.join(choices)}")
                continue
            return user_input
        elif default:
            return default
        elif not required:
            return None
        else:
            print("Error: This parameter is required")


def setup_credentials_interactive():
    """
    Interactively set up AK/SK/Security Token credentials (called only when needed)
    
    Returns:
        tuple: (ak, sk, security_token, region)
    """
    print("\nPlease configure Huawei Cloud credentials:")
    print("-" * 40)
    
    ak = prompt_for_input("Huawei Cloud AK", required=True)
    sk = prompt_for_input("Huawei Cloud SK", required=True, hide_input=True)
    security_token = prompt_for_input("Huawei Cloud Security Token", required=False, hide_input=True)
    region = prompt_for_region("Enter COC service region ID", required=False, default="cn-north-4", is_coc_service=True)
    
    print(f"\n✓ Credentials configured successfully")
    print(f"  AK: {ak[:4]}...{ak[-4:]}")
    print(f"  SK: {'*' * len(sk)}")
    print(f"  Security Token: {'*' * len(security_token)}")
    print(f"  Region: {region} ({get_coc_service_regions().get(region, REGIONS.get(region, ''))})")
    
    return ak, sk, security_token, region


def do_create_script(args):
    """
    Standalone flow for creating a script
    """
    print("=" * 60)
    print("Create Custom Script")
    print("=" * 60)
    
    # Get credential parameters
    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region or "cn-north-4"

    # If credentials not provided, prompt interactively
    if not ak or not sk:
        print("\nPlease provide Huawei Cloud credentials:")
        ak, sk, security_token, region = setup_credentials_interactive()

    print("\nPlease provide script information:")

    # Get parameters for script creation
    name = args.name or prompt_for_input("Script name", required=True)
    script_type = args.type or prompt_for_input(
        "Script type", required=True, choices=['SHELL', 'PYTHON', 'BAT']
    )
    content = args.content or prompt_for_input("Script content", required=True)
    description = args.description or prompt_for_input("Script description", required=True)
    risk_level = args.risk_level or prompt_for_input(
        "Risk level", required=False, default='LOW', choices=['LOW', 'MEDIUM', 'HIGH']
    )
    version = args.version or prompt_for_input("Script version", required=False, default='1.0.0')
    
    print("\n" + "=" * 60)
    print("Creating script...")
    
    try:
        result = create_script(
            name=name,
            script_type=script_type,
            content=content,
            description=description,
            ak=ak,
            sk=sk,
            security_token=security_token,
            region=region,
            risk_level=risk_level,
            version=version
        )
        
        if result.get("ok"):
            print("\n✓ Script created successfully!")
            print(f"  {result['text']}")
            print("\nNote: This skill only supports script execution on L instances. When executing, only provide the L instance resource ID (resource_id).")
        else:
            print(f"\n✗ Creation failed: {result.get('error', {}).get('message')}")
            print("\nNote: Please check if the script content format is correct, or contact administrator for assistance.")
                
    except Exception as e:
        print(f"\n✗ Error occurred: {str(e)}")
        print("\nNote: Please check if the script content format is correct, or contact administrator for assistance.")


def do_execute_script(args):
    """
    Standalone flow for executing a script
    
    Required parameters:
    - resource_id: Resource ID of target L instance
    - region_id: Region where the target L instance is located
    
    Optional parameters:
    - script_uuid: Script UUID (uses most recently created script by default)
    - execute_user: Execution user (default: root)
    - timeout: Timeout in seconds (default: 300)
    - success_rate: Success rate threshold (default: 1)
    - rotation_strategy: Rotation strategy (default: CONTINUE)
    """
    print("=" * 60)
    print("Execute Script")
    print("=" * 60)
    
    # Get credential parameters
    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region or "cn-north-4"

    # If credentials not provided, prompt interactively
    if not ak or not sk:
        print("\nPlease provide Huawei Cloud credentials:")
        ak, sk, security_token, region = setup_credentials_interactive()

    # Get target instance information (L instance only, required)
    print("\nPlease provide target L instance information (required):")
    
    resource_id = prompt_for_input("L instance resource ID (resource_id)", required=True)
    region_id = prompt_for_region("L instance region (region_id)", required=True)
    
    # Get execution parameters (optional, press Enter for defaults)
    print("\nPlease provide execution information (optional, press Enter for defaults):")
    
    script_uuid = args.script_uuid or prompt_for_input("Script UUID", required=False, default='')
    execute_user = args.execute_user or prompt_for_input("Execution user", required=False, default='root')
    
    # If script UUID is empty, try to get the most recently created script
    if not script_uuid:
        print("\nGetting most recently created script...")
        try:
            list_result = list_scripts(ak, sk, security_token, region, limit=1)
            if list_result.get("ok") and list_result.get("result"):
                scripts = list_result.get("result", {}).get("scripts", [])
                if scripts:
                    script_uuid = scripts[0].get("script_uuid", "")
                    script_name = scripts[0].get("name", "")
                    print(f"  Using most recently created script: {script_name} ({script_uuid})")
        except Exception as e:
            print(f"  Failed to get script list: {str(e)}")
    
    # If still no script UUID, prompt user
    if not script_uuid:
        script_uuid = prompt_for_input("Script UUID", required=True)
    timeout = args.timeout if args.timeout else int(prompt_for_input(
        "Timeout (seconds, 5-1800)", required=False, default='300'
    ))
    success_rate = args.success_rate if args.success_rate else float(prompt_for_input(
        "Success rate (supports one decimal place, e.g., 1 or 100)", required=False, default='1'
    ))
    
    target_instances = [{
        "resource_id": resource_id,
        "region_id": region_id,
        "provider": "HCSS",
        "type": "L-INSTANCE"
    }]
    
    rotation_strategy = args.rotation_strategy or prompt_for_input(
        "Rotation strategy", required=False, default='CONTINUE', choices=['CONTINUE', 'PAUSE']
    )
    
    print("\n" + "=" * 60)
    print("Executing script...")
    
    try:
        result = execute_script(
            script_uuid=script_uuid,
            execute_user=execute_user,
            timeout=timeout,
            success_rate=success_rate,
            target_instances=target_instances,
            ak=ak,
            sk=sk,
            security_token=security_token,
            region=region,
            rotation_strategy=rotation_strategy
        )
        
        if result.get("ok"):
            print("\n✓ Script executed successfully!")
            print(f"  {result['text']}")
        else:
            print(f"\n✗ Execution failed: {result.get('error', {}).get('message')}")
            print("\nNote: This skill only supports script execution on L instances. Please verify the L instance server ID and region before running.")
                
    except Exception as e:
        print(f"\n✗ Error occurred: {str(e)}")
        print("\nNote: This skill only supports script execution on L instances. Please verify the L instance server ID and region before running.")


def do_show_script(args):
    """
    Standalone flow for viewing script details
    """
    print("=" * 60)
    print("View Script Details")
    print("=" * 60)
    
    # Get credential parameters
    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region or "cn-north-4"
    
    # If credentials not provided, prompt interactively
    if not ak or not sk:
        print("\nPlease provide Huawei Cloud credentials:")
        ak, sk, security_token, region = setup_credentials_interactive()

    print("\nPlease provide script information:")

    script_uuid = args.script_uuid or prompt_for_input("Script UUID", required=True)
    
    print("\n" + "=" * 60)
    print("Getting script details...")
    
    try:
        result = get_script_detail(script_uuid, ak, sk, security_token, region)
        
        if result.get("ok"):
            print("\n✓ Script details retrieved successfully!")
            script_data = result.get("result", {})
            print("\nScript Details:")
            print("-" * 40)
            print(f"Script UUID: {script_data.get('script_uuid', '')}")
            print(f"Script Name: {script_data.get('name', '')}")
            print(f"Script Type: {script_data.get('type', '')}")
            print(f"Risk Level: {script_data.get('risk_level', '')}")
            print(f"Version: {script_data.get('version', '')}")
            print(f"Description: {script_data.get('description', '')}")
            print(f"Created Time: {script_data.get('create_time', '')}")
            print("\nScript Content:")
            print("-" * 40)
            print(script_data.get('content', ''))
            print("-" * 40)
        else:
            print(f"\n✗ Failed to retrieve: {result.get('error', {}).get('message')}")
                
    except Exception as e:
        print(f"\n✗ Error occurred: {str(e)}")


def do_list_scripts(args):
    """
    Standalone flow for listing scripts
    """
    print("=" * 60)
    print("List Scripts")
    print("=" * 60)
    
    # Get credential parameters
    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region or "cn-north-4"
    
    # If credentials not provided, prompt interactively
    if not ak or not sk:
        print("\nPlease provide Huawei Cloud credentials:")
        ak, sk, security_token, region = setup_credentials_interactive()

    page = args.page if args.page else int(prompt_for_input("Page number", required=False, default='1'))
    size = args.size if args.size else int(prompt_for_input("Page size", required=False, default='10'))
    
    print("\n" + "=" * 60)
    print("Getting script list...")
    
    try:
        result = list_scripts(ak, sk, security_token, region, page, size)
        
        if result.get("ok"):
            print(f"\n✓ Script list retrieved successfully!")
            data = result.get("result", {})
            scripts = data.get("scripts", [])
            total = data.get("total", 0)
            
            print(f"\nFound {total} scripts (Page {page}, {size} per page):")
            print("-" * 80)
            print(f"{'No.':<6} {'Script UUID':<30} {'Name':<20} {'Type':<10} {'Risk Level':<10}")
            print("-" * 80)
            
            for i, script in enumerate(scripts, start=(page-1)*size+1):
                print(f"{i:<6} {script.get('script_uuid', '')[:20]:<30} {script.get('name', ''):<20} {script.get('type', ''):<10} {script.get('risk_level', ''):<10}")
            
            print("-" * 80)
        else:
            print(f"\n✗ Failed to retrieve: {result.get('error', {}).get('message')}")
                
    except Exception as e:
        print(f"\n✗ Error occurred: {str(e)}")


def do_query_execution(args):
    """
    Standalone flow for querying script execution result
    """
    print("=" * 60)
    print("Query Script Execution Result")
    print("=" * 60)
    
    # Get credential parameters
    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region or "cn-north-4"
    
    # If credentials not provided, prompt interactively
    if not ak or not sk:
        print("\nPlease provide Huawei Cloud credentials:")
        ak, sk, security_token, region = setup_credentials_interactive()

    print("\nPlease provide execution information:")

    execute_uuid = args.execute_uuid or prompt_for_input("Execute UUID (format: SCTxxxxxxxxxxxxxxx)", required=True)
    
    print("\n" + "=" * 60)
    print("Querying execution result...")
    
    try:
        result = get_script_job_batch(execute_uuid, ak, sk, security_token)
        
        data = result.get("data", {})
        if data:
            print("\n[OK] Execution result retrieved successfully!")
            
            print("\nExecution Details:")
            print("-" * 40)
            print(f"Execute UUID: {execute_uuid}")
            print(f"Batch Index: {data.get('batch_index', 'N/A')}")
            print(f"Total Instances: {data.get('total_instances', 'N/A')}")
            
            instances = data.get("execute_instances", [])
            if instances:
                print("\nInstance Execution Results:")
                print("-" * 40)
                for inst in instances:
                    status = inst.get("status", "UNKNOWN")
                    print(f"\nStatus: {status}")
                    print(f"Instance ID: {inst.get('id', 'N/A')}")
                    print(f"Command UUID: {inst.get('cmd_uuid', 'N/A')}")
                    print(f"Execution Time: {inst.get('execute_costs', 'N/A')}ms")
                    
                    target = inst.get("target_instance", {})
                    print(f"\nTarget Instance:")
                    print(f"  Resource ID: {target.get('resource_id', 'N/A')}")
                    print(f"  Region: {target.get('region_id', 'N/A')}")
                    print(f"  Provider: {target.get('provider', 'N/A')}")
                    
                    message = inst.get("message", "")
                    if message:
                        print(f"\nScript Output:")
                        print("-" * 40)
                        print(message)
                        print("-" * 40)
            
                # Status summary
                status_summary = ", ".join([inst.get("status", "UNKNOWN") for inst in instances])
                print(f"\nStatus Summary: {status_summary}")
                
                # Check if all instances finished successfully
                all_finished = all(inst.get("status") == "FINISHED" for inst in instances)
                if all_finished:
                    print("\n[OK] All instances executed successfully!")
                else:
                    print("\n[WARNING] Some instances may have failed or are still running.")
            else:
                print("\nNo execution instances found.")
        else:
            print(f"\n[FAIL] Failed to retrieve: {result.get('error', {}).get('message', 'Unknown error')}")
                
    except Exception as e:
        print(f"\n[FAIL] Error occurred: {str(e)}")


def main():
    parser = argparse.ArgumentParser(
        description="Huawei Cloud COC Script Management Tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Create script
  python {baseDir}/scripts/caller.py create --ak "your_ak" --sk "your_sk" --security-token "your_security_token" --name "test_script" --type SHELL --content "echo 'Hello'" --description "test"

  # Execute script
  python {baseDir}/scripts/caller.py execute --ak "your_ak" --sk "your_sk" --security-token "your_security_token" --script-uuid "SC20231025..." --execute-user root

  # View script details
  python {baseDir}/scripts/caller.py show --ak "your_ak" --sk "your_sk" --security-token "your_security_token" --script-uuid "SC20231025..."

  # List scripts
  python {baseDir}/scripts/caller.py list --ak "your_ak" --sk "your_sk" --security-token "your_security_token" --page 1 --size 10

  # Query execution result
  python {baseDir}/scripts/caller.py query --ak "your_ak" --sk "your_sk" --security-token "your_security_token" --execute-uuid "SCT20231025..."

Configuration:
  Use --ak and --sk --security-token parameters to pass Huawei Cloud credentials
  If not provided, will prompt interactively when needed
        """,
    )
    
    # Global parameters (shared by all commands)
    parser.add_argument('--ak', help='Huawei Cloud Access Key')
    parser.add_argument('--sk', help='Huawei Cloud Secret Key')
    parser.add_argument('--security-token', help='Huawei Cloud Security Token')
    parser.add_argument('--region', default='cn-north-4', help='COC service region (default: cn-north-4)')
    
    subparsers = parser.add_subparsers(dest='command', help='Available commands')

    # create command
    create_parser = subparsers.add_parser('create', help='Create custom script')
    create_parser.add_argument('--name', help='Script name')
    create_parser.add_argument('--type', choices=['SHELL', 'PYTHON', 'BAT'], help='Script type')
    create_parser.add_argument('--content', help='Script content')
    create_parser.add_argument('--description', help='Script description')
    create_parser.add_argument('--risk-level', choices=['LOW', 'MEDIUM', 'HIGH'], help='Risk level')
    create_parser.add_argument('--version', help='Script version')

    # execute command
    execute_parser = subparsers.add_parser('execute', help='Execute script')
    execute_parser.add_argument('--script-uuid', help='Script UUID')
    execute_parser.add_argument('--execute-user', help='Execution user')
    execute_parser.add_argument('--timeout', type=int, help='Timeout in seconds')
    execute_parser.add_argument('--success-rate', type=float, help='Success rate threshold (1-100)')
    execute_parser.add_argument('--rotation-strategy', choices=['CONTINUE', 'PAUSE'], help='Rotation strategy')

    # show command
    show_parser = subparsers.add_parser('show', help='View script details')
    show_parser.add_argument('--script-uuid', help='Script UUID')

    # list command
    list_parser = subparsers.add_parser('list', help='List scripts')
    list_parser.add_argument('--page', type=int, help='Page number')
    list_parser.add_argument('--size', type=int, help='Page size')

    # query command
    query_parser = subparsers.add_parser('query', help='Query script execution result')
    query_parser.add_argument('--execute-uuid', help='Execute UUID')

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    try:
        if args.command == 'create':
            do_create_script(args)
        elif args.command == 'execute':
            do_execute_script(args)
        elif args.command == 'show':
            do_show_script(args)
        elif args.command == 'list':
            do_list_scripts(args)
        elif args.command == 'query':
            do_query_execution(args)

    except KeyboardInterrupt:
        print("\n\nOperation interrupted by user")
        sys.exit(0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
# coding: utf-8
"""
Huawei Cloud COC Script Management Library

Manages COC scripts using Huawei Cloud SDK, supporting script creation and execution.
AK/SK/Security Token are passed as parameters: ak, sk, security_token, region
"""

import logging
from typing import Any, Optional, List, Dict

from huaweicloudsdkcore.auth.credentials import GlobalCredentials
from huaweicloudsdkcoc.v1.region.coc_region import CocRegion
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcoc.v1 import *
from huaweicloudsdkcoc.v1.model.script_execute_model import ScriptExecuteModel
from huaweicloudsdkcoc.v1.model.script_execute_param import ScriptExecuteParam
from huaweicloudsdkcoc.v1.model.execute_instances_batch_info import ExecuteInstancesBatchInfo
from huaweicloudsdkcoc.v1.model.execute_resource_instance import ExecuteResourceInstance
from huaweicloudsdkcoc.v1.model.script_properties_model import ScriptPropertiesModel
from huaweicloudsdkcoc.v1.model.add_script_model import AddScriptModel

logger = logging.getLogger(__name__)

# Constants
VALID_SCRIPT_TYPES = ["SHELL", "PYTHON", "BAT"]
VALID_RISK_LEVELS = ["LOW", "MEDIUM", "HIGH"]
VALID_ROTATION_STRATEGIES = ["CONTINUE", "PAUSE"]


def get_valid_coc_regions() -> list[str]:
    """
    Dynamically get the list of regions supported by COC service from SDK.
    
    Returns:
        List of valid COC service regions
    """
    from huaweicloudsdkcoc.v1.region.coc_region import CocRegion
    return list(CocRegion.static_fields.keys())


def get_client(ak: str, sk: str, security_token: str, region: str = "cn-north-4") -> CocClient:
    """
    Create and return a COC client.

    Args:
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication (optional)
        region: COC service region, default cn-north-4

    Returns:
        CocClient instance

    Raises:
        ValueError: If AK/SK not provided or region is invalid
    """
    if not ak:
        raise ValueError("ak is required")
    if not sk:
        raise ValueError("sk is required")
    
    valid_regions = get_valid_coc_regions()
    if region not in valid_regions:
        raise ValueError(f"region must be one of {valid_regions}")

    # Determine authentication method based on security_token presence
    if security_token:
        credentials = GlobalCredentials(ak, sk).with_security_token(security_token)
    else:
        credentials = GlobalCredentials(ak, sk)

    client = CocClient.new_builder() \
        .with_credentials(credentials) \
        .with_region(CocRegion.value_of(region)) \
        .build()

    return client


def create_script(
    name: str,
    script_type: str,
    content: str,
    description: str,
    ak: str,
    sk: str,
    security_token: str,
    region: str = "cn-north-4",
    risk_level: str = "LOW",
    version: str = "1.0.0",
    script_params: Optional[List[Dict[str, Any]]] = None,
) -> dict[str, Any]:
    """
    Create a custom script in COC.

    Args:
        name: Script name
        script_type: Script type (SHELL/PYTHON/BAT)
        content: Script content
        description: Script description
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication
        region: COC service region, default cn-north-4
        risk_level: Risk level (LOW/MEDIUM/HIGH), default LOW
        version: Script version, default 1.0.0
        script_params: Script parameters list

    Returns:
        {
            "ok": True,
            "text": "Script created successfully: SC2023102521413701c4a8a62",
            "result": { Raw API response },
            "error": None
        }
    """
    # Validate input
    if not name:
        return _error("INPUT_ERROR", "name is required")
    if script_type not in VALID_SCRIPT_TYPES:
        return _error("INPUT_ERROR", f"script_type must be one of {VALID_SCRIPT_TYPES}")
    if not content:
        return _error("INPUT_ERROR", "content is required")
    if not description:
        return _error("INPUT_ERROR", "description is required")
    if risk_level not in VALID_RISK_LEVELS:
        return _error("INPUT_ERROR", f"risk_level must be one of {VALID_RISK_LEVELS}")

    try:
        client = get_client(ak, sk, security_token, region)
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    try:
        request = CreateScriptRequest()
        properties = ScriptPropertiesModel(
            risk_level=risk_level,
            version=version
        )
        request.body = AddScriptModel(
            name=name,
            type=script_type,
            content=content,
            description=description,
            properties=properties
        )

        response = client.create_script(request)
        script_uuid = response.data if hasattr(response, 'data') else str(response)

        return {
            "ok": True,
            "text": f"Script created successfully: {script_uuid}",
            "result": response,
            "error": None,
        }

    except exceptions.ClientRequestException as e:
        return _error("API_ERROR", f"{e.error_code}: {e.error_msg}")
    except Exception as e:
        return _error("UNKNOWN_ERROR", str(e))


def execute_script(
    script_uuid: str,
    execute_user: str,
    timeout: int,
    success_rate: float,
    target_instances: List[Dict[str, str]],
    ak: str,
    sk: str,
    security_token: str,
    region: str = "cn-north-4",
    rotation_strategy: str = "CONTINUE",
) -> dict[str, Any]:
    """
    Execute a custom script on target instances.

    Args:
        script_uuid: UUID of the script to execute
        execute_user: User to execute the script (e.g., root)
        timeout: Execution timeout in seconds (5 < timeout < 1800)
        success_rate: Success rate (supports one decimal place, e.g., 1 or 100)
        target_instances: List of target instances, each containing:
            - resource_id: Instance ID (required)
            - region_id: Region where server is located (required)
            - provider: Resource provider (not required for ECS instances, default "HCSS" for L instances)
            - type: Resource type (not required for ECS instances, default "L-INSTANCE" for L instances)
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication
        region: COC service region, default cn-north-4
        rotation_strategy: Rotation strategy (CONTINUE/PAUSE), default CONTINUE

    Returns:
        {
            "ok": True,
            "text": "Script execution started: SCT2023083109562601af694bf",
            "result": { Raw API response },
            "error": None
        }
    """
    # Validate input
    if not script_uuid:
        return _error("INPUT_ERROR", "script_uuid is required")
    if not execute_user:
        return _error("INPUT_ERROR", "execute_user is required")
    if timeout <= 5 or timeout > 1800:
        return _error("INPUT_ERROR", "timeout must be between 5 and 1800 seconds")
    if success_rate < 0 or success_rate > 100:
        return _error("INPUT_ERROR", "success_rate must be between 0 and 100")
    if not target_instances or not isinstance(target_instances, list):
        return _error("INPUT_ERROR", "target_instances is required")
    if rotation_strategy not in VALID_ROTATION_STRATEGIES:
        return _error("INPUT_ERROR", f"rotation_strategy must be one of {VALID_ROTATION_STRATEGIES}")

    try:
        client = get_client(ak, sk, security_token, region)
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    try:
        request = ExecuteScriptRequest()
        request.script_uuid = script_uuid

        # Build execution parameters
        execute_param = ScriptExecuteParam(
            timeout=timeout,
            success_rate=success_rate,
            execute_user=execute_user
        )

        # Build target instance list
        listTargetInstancesExecuteBatches = []
        for instance_info in target_instances:
            # Build instance object, add provider/type fields if available
            instance_kwargs = {
                "resource_id": instance_info.get("resource_id", ""),
                "region_id": instance_info.get("region_id", "cn-north-4")
            }
            
            # Add provider and type fields for L instances
            provider = instance_info.get("provider")
            instance_type = instance_info.get("type")
            if provider:
                instance_kwargs["provider"] = provider
            if instance_type:
                instance_kwargs["type"] = instance_type
            
            instance = ExecuteResourceInstance(**instance_kwargs)
            listTargetInstancesExecuteBatches.append(instance)

        # Build batch
        listExecuteBatchesbody = [
            ExecuteInstancesBatchInfo(
                batch_index=1,
                target_instances=listTargetInstancesExecuteBatches,
                rotation_strategy=rotation_strategy
            )
        ]

        # Build request body
        request.body = ScriptExecuteModel(
            execute_batches=listExecuteBatchesbody,
            execute_param=execute_param
        )

        response = client.execute_script(request)
        execute_uuid = response.data if hasattr(response, 'data') else str(response)

        return {
            "ok": True,
            "text": f"Script execution started: {execute_uuid}",
            "result": response,
            "error": None,
        }

    except exceptions.ClientRequestException as e:
        return _error("API_ERROR", f"{e.error_code}: {e.error_msg}")
    except Exception as e:
        return _error("UNKNOWN_ERROR", str(e))


def get_script_detail(script_uuid: str, ak: str, sk: str, security_token: str, region: str = "cn-north-4") -> dict[str, Any]:
    """
    Get script details.

    Args:
        script_uuid: Script UUID
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication
        region: COC service region, default cn-north-4

    Returns:
        {
            "ok": True,
            "text": "Script details retrieved",
            "result": { 
                "script_uuid": "xxx",
                "name": "Script name",
                "type": "SHELL",
                "content": "Script content",
                "description": "Description",
                "risk_level": "LOW",
                "version": "1.0.0",
                "create_time": "Creation time"
            },
            "error": None
        }
    """
    if not script_uuid:
        return _error("INPUT_ERROR", "script_uuid is required")

    try:
        client = get_client(ak, sk, security_token, region)
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    try:
        request = GetScriptRequest()
        request.script_uuid = script_uuid
        
        response = client.get_script(request)
        
        # Parse response
        script_data = {}
        if hasattr(response, 'data'):
            data = response.data
            script_data = {
                "script_uuid": data.script_uuid if hasattr(data, 'script_uuid') else "",
                "name": data.name if hasattr(data, 'name') else "",
                "type": data.type if hasattr(data, 'type') else "",
                "content": data.content if hasattr(data, 'content') else "",
                "description": data.description if hasattr(data, 'description') else "",
                "risk_level": data.risk_level if hasattr(data, 'risk_level') else "",
                "version": data.version if hasattr(data, 'version') else "",
                "create_time": data.create_time if hasattr(data, 'create_time') else ""
            }

        return {
            "ok": True,
            "text": "Script details retrieved",
            "result": script_data,
            "error": None,
        }

    except exceptions.ClientRequestException as e:
        return _error("API_ERROR", f"{e.error_code}: {e.error_msg}")
    except Exception as e:
        return _error("UNKNOWN_ERROR", str(e))


def list_scripts(ak: str, sk: str, security_token: str, region: str = "cn-north-4", page: int = 1, limit: int = 10) -> dict[str, Any]:
    """
    List scripts.

    Args:
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication
        region: COC service region, default cn-north-4
        page: Page number (starting from 1)
        limit: Page size

    Returns:
        {
            "ok": True,
            "text": "Scripts listed",
            "result": {
                "scripts": [...],
                "total": 100
            },
            "error": None
        }
    """
    try:
        client = get_client(ak, sk, security_token, region)
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    try:
        request = ListScriptsRequest()
        request.page = page
        request.limit = limit
        
        response = client.list_scripts(request)
        
        scripts = []
        total = 0
        if hasattr(response, 'data'):
            data = response.data
            if hasattr(data, 'data'):
                for script in data.data:
                    # Extract risk level and version from properties
                    risk_level = ""
                    version = ""
                    if hasattr(script, 'properties') and script.properties:
                        props = script.properties
                        risk_level = props.risk_level if hasattr(props, 'risk_level') else ""
                        version = props.version if hasattr(props, 'version') else ""
                    
                    scripts.append({
                        "script_uuid": script.script_uuid if hasattr(script, 'script_uuid') else "",
                        "name": script.name if hasattr(script, 'name') else "",
                        "type": script.type if hasattr(script, 'type') else "",
                        "description": script.description if hasattr(script, 'description') else "",
                        "risk_level": risk_level,
                        "version": version,
                        "create_time": script.gmt_created if hasattr(script, 'gmt_created') else ""
                    })
            if hasattr(data, 'total'):
                total = data.total

        return {
            "ok": True,
            "text": f"Found {total} scripts",
            "result": {
                "scripts": scripts,
                "total": total
            },
            "error": None,
        }

    except exceptions.ClientRequestException as e:
        return _error("API_ERROR", f"{e.error_code}: {e.error_msg}")
    except Exception as e:
        return _error("UNKNOWN_ERROR", str(e))


def get_script_job_batch(
    execute_uuid: str,
    ak: str,
    sk: str,
    security_token: str
) -> dict[str, Any]:
    """
    Query script execution result by execute UUID.
    
    API: GET /v1/job/script/orders/{execute_uuid}/batches/1
    
    Args:
        execute_uuid: Execution UUID (format: SCTxxxxxxxxxxxxxxxbf)
        ak: Huawei Cloud Access Key
        sk: Huawei Cloud Secret Key
        security_token: Temporary security token for STS authentication

    Returns:
        {
            "data": {
                "batch_index": 1,
                "total_instances": 1,
                "execute_instances": [{
                    "id": 40304358,
                    "cmd_uuid": "xxx",
                    "job_sign": null,
                    "status": "FINISHED",  # or "ABNORMAL", "RUNNING", etc.
                    "message": "Execution log...",
                    "gmt_created": 1779934038727,
                    "gmt_finished": 1779934107670,
                    "execute_costs": 68943,
                    "target_instance": {
                        "resource_id": "xxx",
                        "agent_sn": "xxx",
                        "agent_status": null,
                        "agent_version": "1.1.8",
                        "region_id": "cn-north-4",
                        "project_id": null,
                        "properties": {
                            "host_name": "dify-test-001",
                            "fixed_ip": null,
                            "floating_ip": null,
                            "region_id": "cn-north-4",
                            "zone_id": null,
                            "application": null,
                            "group": null,
                            "project_id": null
                        },
                        "custom_attributes": null,
                        "provider": "hcss",
                        "type": "l-instance"
                    }
                }]
            }
        }
    """
    if not execute_uuid:
        return _error("INPUT_ERROR", "execute_uuid is required")

    # Fixed parameters
    region = "cn-north-4"
    batch_index = 1
    limit = 50

    try:
        client = get_client(ak, sk, security_token, region)
    except ValueError as e:
        return _error("CONFIG_ERROR", str(e))

    try:
        request = GetScriptJobBatchRequest()
        request.batch_index = batch_index
        request.execute_uuid = execute_uuid
        request.limit = limit
        
        response = client.get_script_job_batch(request)
        
        result_dict = {}
        if hasattr(response, 'data') and response.data is not None:
            data = response.data
            if hasattr(data, 'batch_index'):
                result_dict['batch_index'] = data.batch_index
            if hasattr(data, 'total_instances'):
                result_dict['total_instances'] = data.total_instances
            if hasattr(data, 'execute_instances') and data.execute_instances:
                instances = []
                for instance in data.execute_instances:
                    instance_dict = {}
                    if hasattr(instance, 'id'):
                        instance_dict['id'] = instance.id
                    if hasattr(instance, 'cmd_uuid'):
                        instance_dict['cmd_uuid'] = instance.cmd_uuid
                    if hasattr(instance, 'status'):
                        instance_dict['status'] = instance.status
                    if hasattr(instance, 'message'):
                        instance_dict['message'] = instance.message
                    if hasattr(instance, 'execute_costs'):
                        instance_dict['execute_costs'] = instance.execute_costs
                    if hasattr(instance, 'gmt_created'):
                        instance_dict['gmt_created'] = instance.gmt_created
                    if hasattr(instance, 'gmt_finished'):
                        instance_dict['gmt_finished'] = instance.gmt_finished
                    if hasattr(instance, 'target_instance') and instance.target_instance:
                        target = instance.target_instance
                        target_dict = {}
                        if hasattr(target, 'resource_id'):
                            target_dict['resource_id'] = target.resource_id
                        if hasattr(target, 'agent_sn'):
                            target_dict['agent_sn'] = target.agent_sn
                        if hasattr(target, 'agent_status'):
                            target_dict['agent_status'] = target.agent_status
                        if hasattr(target, 'agent_version'):
                            target_dict['agent_version'] = target.agent_version
                        if hasattr(target, 'region_id'):
                            target_dict['region_id'] = target.region_id
                        if hasattr(target, 'project_id'):
                            target_dict['project_id'] = target.project_id
                        if hasattr(target, 'properties') and target.properties:
                            props = target.properties
                            props_dict = {}
                            if hasattr(props, 'host_name'):
                                props_dict['host_name'] = props.host_name
                            if hasattr(props, 'fixed_ip'):
                                props_dict['fixed_ip'] = props.fixed_ip
                            if hasattr(props, 'floating_ip'):
                                props_dict['floating_ip'] = props.floating_ip
                            if hasattr(props, 'region_id'):
                                props_dict['region_id'] = props.region_id
                            if hasattr(props, 'zone_id'):
                                props_dict['zone_id'] = props.zone_id
                            if hasattr(props, 'application'):
                                props_dict['application'] = props.application
                            if hasattr(props, 'group'):
                                props_dict['group'] = props.group
                            if hasattr(props, 'project_id'):
                                props_dict['project_id'] = props.project_id
                            target_dict['properties'] = props_dict
                        if hasattr(target, 'custom_attributes'):
                            target_dict['custom_attributes'] = target.custom_attributes
                        if hasattr(target, 'provider'):
                            target_dict['provider'] = target.provider
                        if hasattr(target, 'type'):
                            target_dict['type'] = target.type
                        instance_dict['target_instance'] = target_dict
                    if hasattr(instance, 'job_sign'):
                        instance_dict['job_sign'] = instance.job_sign
                    instances.append(instance_dict)
                result_dict['execute_instances'] = instances
        else:
            result_dict['status'] = 'UNKNOWN'
            result_dict['message'] = 'Execution record not found or empty'

        return {
            "data": result_dict
        }

    except exceptions.ClientRequestException as e:
        return _error("API_ERROR", f"{e.error_code}: {e.error_msg}")
    except Exception as e:
        return _error("UNKNOWN_ERROR", str(e))


def _error(code: str, message: str) -> dict:
    """Create error response."""
    return {
        "ok": False,
        "text": "",
        "result": None,
        "error": {"code": code, "message": message},
    }

[project]
name = "huawei-cloud-flexus-l-server-scripts-excute"
version = "1.0.0"
dependencies = [
    "huaweicloudsdkcore>=3.1.0",
    "huaweicloudsdkcoc>=3.1.0",
]

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

#!/usr/bin/env python3
# coding: utf-8
"""
Huawei Cloud COC Skill Smoke Test

Usage:
    python smoke_test.py --ak "your_ak" --sk "your_sk" --security_token "your_security_token" [--region "cn-north-4"]
"""

import os
import sys
import argparse

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from scripts.lib import create_script, execute_script, list_scripts, get_script_detail


def test_credentials(ak, sk, security_token, region):
    """Test credential configuration."""
    print(f"AK: {'✓' if ak else '✗'}")
    print(f"SK: {'✓' if sk else '✗'}")
    print(f"Security Token: {'✓' if security_token else '✗'}")
    print(f"Region: {region}")

    if not ak or not sk or not security_token:
        print("\nError: Please provide --ak, --sk, and --security_token parameters")
        return False

    return True


def test_create_script(ak, sk, security_token, region):
    """Test script creation functionality."""
    print("\nTest: Create Script")
    print("-" * 40)

    import time
    timestamp = str(int(time.time()))[-6:]
    result = create_script(
        name=f"test_script_{timestamp}",
        script_type="SHELL",
        content="echo 'Hello from COC SDK'",
        description="Test script",
        ak=ak,
        sk=sk,
        security_token=security_token,
        region=region,
        risk_level="LOW",
        version="1.0.0"
    )

    print(f"Result: {result['text']}")
    if result.get("ok"):
        print("✓ Script created successfully")
        return result.get("text", "").split(":")[-1].strip()
    else:
        print(f"✗ Failed to create script: {result.get('error', {}).get('message')}")
        return None


def test_list_scripts(ak, sk, security_token, region):
    """Test script listing functionality."""
    print("\nTest: List Scripts")
    print("-" * 40)

    result = list_scripts(ak, sk, security_token, region, page=1, limit=5)

    if result.get("ok"):
        data = result.get("result", {})
        scripts = data.get("scripts", [])
        total = data.get("total", 0)
        print(f"✓ Successfully retrieved {total} scripts")
        if scripts:
            print(f"Latest script: {scripts[0].get('name', '')}")
        return True
    else:
        print(f"✗ Failed to list scripts: {result.get('error', {}).get('message')}")
        return False


def test_execute_script(ak, sk, security_token, region, script_uuid):
    """Test script execution functionality."""
    print("\nTest: Execute Script")
    print("-" * 40)

    target_instances = [
        {
            "resource_id": "your_l_instance_resource_id",
            "region_id": "cn-north-4",
            "provider": "HCSS",
            "type": "L-INSTANCE"
        }
    ]

    result = execute_script(
        script_uuid=script_uuid,
        execute_user="root",
        timeout=60,
        success_rate=100,
        target_instances=target_instances,
        ak=ak,
        sk=sk,
        security_token=security_token,
        region=region
    )

    if result.get("ok"):
        print("✓ Script executed successfully")
        return True
    else:
        print(f"Note: {result.get('error', {}).get('message')}")
        print("(Requires real target instance information to complete execution test)")
        return False


def main():
    parser = argparse.ArgumentParser(description="Huawei Cloud COC Skill Smoke Test")
    parser.add_argument('--ak', required=True, help='Huawei Cloud Access Key')
    parser.add_argument('--sk', required=True, help='Huawei Cloud Secret Key')
    parser.add_argument('--security_token', required=True, help='Temporary security token for STS authentication')
    parser.add_argument('--region', default='cn-north-4', help='COC service region')
    args = parser.parse_args()

    print("=" * 60)
    print("Huawei Cloud COC Skill Smoke Test")
    print("=" * 60)

    ak = args.ak
    sk = args.sk
    security_token = args.security_token
    region = args.region

    # Test credentials
    if not test_credentials(ak, sk, security_token, region):
        return

    # Test list scripts
    test_list_scripts(ak, sk, security_token, region)

    # Test create script
    script_uuid = test_create_script(ak, sk, security_token, region)
    if not script_uuid:
        return

    # Test execute script (requires real instance, skipped by default)
    # test_execute_script(ak, sk, security_token, region, script_uuid)

    print("\n" + "=" * 60)
    print("Test completed!")
    print("=" * 60)
    print(f"\nCreated script UUID: {script_uuid}")
    print("\nNote: Script execution requires real target instance information")


if __name__ == "__main__":
    main()