#!/usr/bin/env python
"""
Script to gather all CloudFormation resource types available across AWS regions.

The collected data is written to `localstack/services/cloudformation/resources.py` as

* `AWS_AVAILABLE_CFN_RESOURCES`: a dict, mapping resource names to the regions they are available in.
* `AWS_CFN_REGIONS_SCANNED`: a set of regions for which listing CloudFormation types succeeded.

The script expects valid AWS credentials either via the environment or configured profiles (for local execution).

Note:
We evaluated scraping the public documentation/zip schema endpoints, but they do not have resources like
``AWS::OpsWorksCM::Server`` that are still returned by the CloudFormation API.
To make sure we cover all resources, we rely on the list of available types from the API.
"""

import argparse
import logging
import sys
from collections.abc import Iterable
from pathlib import Path

import boto3
from botocore.exceptions import ClientError

LOG = logging.getLogger(__name__)

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_RESOURCE_FILE = (
    REPO_ROOT / "localstack-core" / "localstack" / "services" / "cloudformation" / "resources.py"
)


def get_regions(
    session: boto3.session.Session, explicit_regions: Iterable[str] | None = None
) -> list[str]:
    if explicit_regions:
        return sorted(set(explicit_regions))

    return sorted(session.get_available_regions("cloudformation"))


def collect_region_resource_types(
    session: boto3.session.Session, region: str
) -> tuple[set[str], bool]:
    client = session.client("cloudformation", region_name=region)
    resources: set[str] = set()
    token: str | None = None
    succeeded = True

    while True:
        try:
            params = {
                "Visibility": "PUBLIC",
                "Type": "RESOURCE",
                "Filters": {"Category": "AWS_TYPES"},
            }
            if token:
                params["NextToken"] = token
            response = client.list_types(**params)
        except ClientError as exc:
            LOG.warning("Skipping region %s due to error while listing types: %s", region, exc)
            succeeded = False
            break

        for summary in response.get("TypeSummaries", []):
            type_name = summary.get("TypeName")
            if type_name:
                resources.add(type_name)

        token = response.get("NextToken")
        if not token:
            break

    return resources, succeeded


def collect_all_resource_types(
    session: boto3.session.Session, regions: Iterable[str]
) -> tuple[dict[str, set[str]], set[str]]:
    aggregated: dict[str, set[str]] = {}
    successful_regions: set[str] = set()
    for region in regions:
        print(f"Collecting CloudFormation resource types in region {region}")
        region_resources, succeeded = collect_region_resource_types(session, region)
        if not succeeded:
            continue

        successful_regions.add(region)
        for resource in region_resources:
            aggregated.setdefault(resource, set()).add(region)
    return aggregated, successful_regions


def render_resource_file(resources: dict[str, set[str]], successful_regions: set[str]) -> str:
    lines: list[str] = [
        '"""Generated by scripts/update_cfn_resources.py – do not edit manually."""',
        "",
    ]

    if resources:
        lines.append("AWS_AVAILABLE_CFN_RESOURCES = {")
        for resource in sorted(resources):
            regions = sorted(resources[resource])
            if regions:
                lines.append(f'    "{resource}": [')
                for region_name in regions:
                    lines.append(f'        "{region_name}",')
                lines.append("    ],")
            else:
                lines.append(f'    "{resource}": [],')
        lines.append("}")
    else:
        lines.append("AWS_AVAILABLE_CFN_RESOURCES = {}")

    lines.append("")

    if successful_regions:
        lines.append("AWS_CFN_REGIONS_SCANNED = {")
        for region in sorted(successful_regions):
            lines.append(f'    "{region}",')
        lines.append("}")
    else:
        lines.append("AWS_CFN_REGIONS_SCANNED = set()")

    lines.append("")

    return "\n".join(lines)


def write_resource_file(path: Path, content: str) -> None:
    path.write_text(content, encoding="utf-8")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Update the AWS_AVAILABLE_CFN_RESOURCES constant.")
    parser.add_argument("--profile", help="AWS profile name to use")
    parser.add_argument(
        "--regions",
        nargs="*",
        help="Optional list of AWS regions to scan. Defaults to all regions supporting CloudFormation.",
    )
    parser.add_argument(
        "--resource-file",
        default=str(DEFAULT_RESOURCE_FILE),
        help="Path to the resources.py file that should be updated.",
    )
    parser.add_argument(
        "--dry-run", action="store_true", help="Do not write the file, only print the resources."
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    session_kwargs = {}
    if args.profile:
        session_kwargs["profile_name"] = args.profile

    try:
        session = boto3.session.Session(**session_kwargs)
    except Exception as exc:
        LOG.error("Failed to create boto3 session: %s", exc)
        return 1

    regions = get_regions(session, args.regions)
    if not regions:
        LOG.error("Could not determine any regions to scan.")
        return 1

    resources, successful_regions = collect_all_resource_types(session, regions)
    if not resources:
        LOG.error("No CloudFormation resources were discovered.")
        return 1

    content = render_resource_file(resources, successful_regions)

    if args.dry_run:
        sys.stdout.write(content)
        return 0

    resource_file = Path(args.resource_file)
    write_resource_file(resource_file, content)
    print(
        f"Updated {resource_file} with {len(resources)} CloudFormation resource types "
        f"across {len(successful_regions)} regions."
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())
