# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Shared utilities for audit tools."""

import json
import os
import re
import tempfile
from datetime import datetime, timezone
from loguru import logger
from typing import Any, Dict, List, Optional, Tuple, Union


# Constants
DEFAULT_BATCH_SIZE = 5
FUZZY_MATCH_THRESHOLD = 30  # Minimum similarity score for fuzzy matching
HIGH_CONFIDENCE_MATCH_THRESHOLD = 85  # High confidence threshold for exact fuzzy matches


async def execute_audit_api(input_obj: Dict[str, Any], region: str, banner: str) -> str:
    """Execute the Application Signals audit API call with the given input object."""
    from .aws_clients import applicationsignals_client

    # File log path
    desired_log_path = os.environ.get('AUDITOR_LOG_PATH', tempfile.gettempdir())
    try:
        if desired_log_path.endswith(os.sep) or os.path.isdir(desired_log_path):
            os.makedirs(desired_log_path, exist_ok=True)
            log_path = os.path.join(desired_log_path, 'aws_api.log')
        else:
            os.makedirs(os.path.dirname(desired_log_path) or '.', exist_ok=True)
            log_path = desired_log_path
    except Exception:
        temp_dir = tempfile.gettempdir()
        os.makedirs(temp_dir, exist_ok=True)
        log_path = os.path.join(temp_dir, 'aws_api.log')

    # Process targets in batches if needed
    targets = input_obj.get('AuditTargets', [])
    batch_size = DEFAULT_BATCH_SIZE
    target_batches = []

    if len(targets) > batch_size:
        logger.info(f'Processing {len(targets)} targets in batches of {batch_size}')
        for i in range(0, len(targets), batch_size):
            batch = targets[i : i + batch_size]
            target_batches.append(batch)
    else:
        target_batches.append(targets)

    all_batch_results = []

    for batch_idx, batch_targets in enumerate(target_batches, 1):
        logger.info(
            f'Processing batch {batch_idx}/{len(target_batches)} with {len(batch_targets)} targets'
        )

        # Build API input for this batch
        batch_input_obj = {
            'StartTime': datetime.fromtimestamp(input_obj['StartTime'], tz=timezone.utc),
            'EndTime': datetime.fromtimestamp(input_obj['EndTime'], tz=timezone.utc),
            'AuditTargets': batch_targets,
        }
        if 'Auditors' in input_obj:
            batch_input_obj['Auditors'] = input_obj['Auditors']

        # Log API invocation details
        api_pretty_input = json.dumps(
            {
                'StartTime': input_obj['StartTime'],
                'EndTime': input_obj['EndTime'],
                'AuditTargets': batch_targets,
                'Auditors': input_obj.get('Auditors', []),
            },
            indent=2,
        )

        # Also log the actual batch_input_obj that will be sent to AWS API
        batch_input_for_logging = {
            'StartTime': batch_input_obj['StartTime'].isoformat(),
            'EndTime': batch_input_obj['EndTime'].isoformat(),
            'AuditTargets': batch_input_obj['AuditTargets'],
        }
        if 'Auditors' in batch_input_obj:
            batch_input_for_logging['Auditors'] = batch_input_obj['Auditors']

        batch_payload_json = json.dumps(batch_input_for_logging, indent=2)

        logger.info('═' * 80)
        logger.info(
            f'BATCH {batch_idx}/{len(target_batches)} - {datetime.now(timezone.utc).isoformat()}'
        )
        logger.info(banner.strip())
        logger.info('---- API INVOCATION ----')
        logger.info('applicationsignals_client.list_audit_findings()')
        logger.info('---- API PARAMETERS (JSON) ----')
        logger.info(api_pretty_input)
        logger.info('---- ACTUAL AWS API PAYLOAD ----')
        logger.info(batch_payload_json)
        logger.info('---- END PARAMETERS ----')

        # Write detailed payload to log file
        try:
            with open(log_path, 'a') as f:
                f.write('═' * 80 + '\n')
                f.write(
                    f'BATCH {batch_idx}/{len(target_batches)} - {datetime.now(timezone.utc).isoformat()}\n'
                )
                f.write(banner.strip() + '\n')
                f.write('---- API INVOCATION ----\n')
                f.write('applicationsignals_client.list_audit_findings()\n')
                f.write('---- API PARAMETERS (JSON) ----\n')
                f.write(api_pretty_input + '\n')
                f.write('---- ACTUAL AWS API PAYLOAD ----\n')
                f.write(batch_payload_json + '\n')
                f.write('---- END PARAMETERS ----\n\n')
        except Exception as log_error:
            logger.warning(f'Failed to write audit log to {log_path}: {log_error}')

        # Call the Application Signals API for this batch
        try:
            response = applicationsignals_client.list_audit_findings(**batch_input_obj)  # type: ignore[attr-defined]

            # Format and log output for this batch
            observation_text = json.dumps(response, indent=2, default=str)
            all_batch_results.append(response)

            if not response.get('AuditFindings'):
                try:
                    with open(log_path, 'a') as f:
                        f.write(f'📭 Batch {batch_idx}: No findings returned.\n')
                        f.write('---- END RESPONSE ----\n\n')
                except Exception as log_error:
                    logger.warning(f'Failed to write audit log to {log_path}: {log_error}')
                logger.info(f'📭 Batch {batch_idx}: No findings returned.\n---- END RESPONSE ----')
            else:
                try:
                    with open(log_path, 'a') as f:
                        f.write(f'---- BATCH {batch_idx} API RESPONSE (JSON) ----\n')
                        f.write(observation_text + '\n')
                        f.write('---- END RESPONSE ----\n\n')
                except Exception as log_error:
                    logger.warning(f'Failed to write audit log to {log_path}: {log_error}')
                logger.info(
                    f'---- BATCH {batch_idx} API RESPONSE (JSON) ----\n'
                    + observation_text
                    + '\n---- END RESPONSE ----'
                )

        except Exception as e:
            error_msg = str(e)
            try:
                with open(log_path, 'a') as f:
                    f.write(f'---- BATCH {batch_idx} API ERROR ----\n')
                    f.write(error_msg + '\n')
                    f.write('---- END ERROR ----\n\n')
            except Exception as log_error:
                logger.warning(f'Failed to write audit log to {log_path}: {log_error}')
            logger.error(
                f'---- BATCH {batch_idx} API ERROR ----\n' + error_msg + '\n---- END ERROR ----'
            )

            batch_error_result = {
                'error': f'API call failed: {error_msg}',
                'targets': batch_targets,
            }
            all_batch_results.append(batch_error_result)
            continue

    # Aggregate results from all batches
    if not all_batch_results:
        return banner + 'Result: No findings from any batch.'

    # Aggregate the findings from all successful batches
    aggregated_findings = []
    failed_batches = 0

    for batch_idx, batch_result in enumerate(all_batch_results):
        if isinstance(batch_result, dict):
            if 'error' in batch_result:
                failed_batches += 1
                continue

            batch_findings = batch_result.get('AuditFindings', [])
            aggregated_findings.extend(batch_findings)

    # Create final aggregated response
    final_result = {
        'AuditFindings': aggregated_findings,
    }

    # Add any error information if there were failed batches
    if failed_batches > 0:
        error_details = []
        for batch_result in all_batch_results:
            if isinstance(batch_result, dict) and 'error' in batch_result:
                error_details.append(
                    {
                        'error': batch_result['error'],
                        'targets': batch_result['targets'],
                    }
                )
        final_result['ListAuditFindingsErrors'] = error_details

    final_observation_text = json.dumps(final_result, indent=2, default=str)
    return banner + final_observation_text


def _create_service_target(
    service_name: str, environment: str, aws_account_id: Optional[str] = None
) -> Dict[str, Any]:
    """Create a standardized service target configuration."""
    service_config = {
        'Type': 'Service',
        'Name': service_name,
        'Environment': environment,
    }
    if aws_account_id:
        service_config['AwsAccountId'] = aws_account_id

    return {
        'Type': 'service',
        'Data': {'Service': service_config},
    }


def _filter_instrumented_services(all_services: List[Any]) -> List[Dict[str, Any]]:
    """Filter out uninstrumented and aws native services.

    Args:
        all_services: List of service summaries from list_services API
    Returns:
        List of services that are instrumented
    """
    instrumented_services = []

    for service in all_services:
        service_attrs = service.get('KeyAttributes', {})
        service_name = service_attrs.get('Name', '')
        service_type = service_attrs.get('Type', '')
        environment = service_attrs.get('Environment', '')

        # Filter out services without proper names or that are not actual services
        if not service_name or service_name == 'Unknown' or service_type != 'Service':
            logger.debug(
                f"Skipping service: Name='{service_name}', Type='{service_type}', Environment='{environment}'"
            )
            continue

        # Check InstrumentationType in AttributeMaps to filter out UNINSTRUMENTED and AWS_NATIVE services
        attribute_maps = service.get('AttributeMaps', [])
        is_instrumented = True

        for attr_map in attribute_maps:
            if isinstance(attr_map, dict) and 'InstrumentationType' in attr_map:
                instrumentation_type = attr_map['InstrumentationType']
                if (
                    instrumentation_type == 'UNINSTRUMENTED'
                    or instrumentation_type == 'AWS_NATIVE'
                ):
                    is_instrumented = False
                    logger.debug(
                        f"Filtering out uninstrumented service: Name='{service_name}', InstrumentationType='{instrumentation_type}'"
                    )
                    break

        if is_instrumented:
            instrumented_services.append(service)
            logger.debug(
                f"Including instrumented service: Name='{service_name}', Environment='{environment}'"
            )

    logger.info(
        f'Filtered services: {len(instrumented_services)} instrumented out of {len(all_services)} total services'
    )
    return instrumented_services


def _fetch_instrumented_services_with_pagination(
    unix_start: int,
    unix_end: int,
    next_token: Optional[str] = None,
    max_results: int = 5,
    applicationsignals_client=None,
) -> tuple[List[Dict[str, Any]], Optional[str], List[str], Dict[str, int]]:
    """Common pagination logic for fetching instrumented services.

    Args:
        unix_start: Start time as unix timestamp
        unix_end: End time as unix timestamp
        next_token: Token for pagination from previous list_services call
        max_results: Maximum number of services to return per batch
        applicationsignals_client: AWS Application Signals client
    Returns:
        Tuple of (instrumented_services, next_token, all_service_names, filtering_stats)
        filtering_stats contains: {'total_services': int, 'instrumented_services': int, 'filtered_out': int}
    """
    if applicationsignals_client is None:
        from .aws_clients import applicationsignals_client

    all_service_names = []
    filtering_stats = {'total_services': 0, 'instrumented_services': 0, 'filtered_out': 0}

    # Initialize variables for the loop
    current_next_token = next_token
    total_services_viewed = 0
    total_filtered_out = 0
    instrumented_services = []
    returned_next_token = None

    # Loop until we find instrumented services or run out of pages
    while True:
        # Build list_services parameters
        list_services_params = {
            'StartTime': datetime.fromtimestamp(unix_start, tz=timezone.utc),
            'EndTime': datetime.fromtimestamp(unix_end, tz=timezone.utc),
            'MaxResults': max_results,
        }

        # Add NextToken if provided for pagination
        if current_next_token:
            list_services_params['NextToken'] = current_next_token

        logger.info(f'Fetching batch (viewed so far: {total_services_viewed} services)')

        services_response = applicationsignals_client.list_services(**list_services_params)
        services_batch = services_response.get('ServiceSummaries', [])
        returned_next_token = services_response.get('NextToken')

        # Collect all service names from this batch (no filtering)
        for service in services_batch:
            service_attrs = service.get('KeyAttributes', {})
            service_name = service_attrs.get('Name', '')
            all_service_names.append(service_name)

        # Update total services viewed
        total_services_viewed += len(services_batch)

        logger.debug(
            f'Retrieved {len(services_batch)} services in this batch, NextToken: {returned_next_token is not None}'
        )

        # Filter out uninstrumented services using the helper function
        instrumented_services = _filter_instrumented_services(services_batch)

        # Update totals
        batch_filtered_out = len(services_batch) - len(instrumented_services)
        total_filtered_out += batch_filtered_out

        logger.info(
            f'Fetch instrumented services batch results: {len(services_batch)} total, {len(instrumented_services)} instrumented, {batch_filtered_out} filtered out'
        )
        logger.info(
            f'Fetch instrumented services cumulative: {total_services_viewed} total viewed, {total_filtered_out} filtered out'
        )

        # Check if we found instrumented services - if so, exit the loop immediately
        if len(instrumented_services) > 0:
            logger.info(
                f'Found {len(instrumented_services)} instrumented services, proceeding with expansion'
            )
            break
        elif not returned_next_token:
            logger.warning(
                f'No instrumented services found after viewing {total_services_viewed} total services across all pages'
            )
            break
        else:
            logger.info(
                'No instrumented services in this batch, continuing to next page (next_token available)'
            )
            current_next_token = returned_next_token

    # Update filtering stats with final totals
    filtering_stats['total_services'] = total_services_viewed
    filtering_stats['instrumented_services'] = len(instrumented_services)
    filtering_stats['filtered_out'] = total_filtered_out

    return (instrumented_services, returned_next_token, all_service_names, filtering_stats)


def parse_auditors(
    auditors_value: Union[str, None, Any], default_auditors: List[str]
) -> List[str]:
    """Parse and validate auditors parameter."""
    # Handle Pydantic Field objects that may be passed instead of actual values
    if hasattr(auditors_value, 'default') and hasattr(auditors_value, 'description'):
        # This is a Pydantic Field object, use its default value
        auditors_value = getattr(auditors_value, 'default', None)

    if auditors_value is None:
        user_prompt_text = os.environ.get('MCP_USER_PROMPT', '') or ''
        wants_root_cause = 'root cause' in user_prompt_text.lower()
        raw_a = default_auditors if not wants_root_cause else []
    elif str(auditors_value).lower() == 'all':
        raw_a = []  # Empty list means use all auditors
    else:
        raw_a = [a.strip() for a in str(auditors_value).split(',') if a.strip()]

    # Validate auditors
    if len(raw_a) == 0:
        return []  # Empty list means use all auditors
    else:
        allowed = {
            'slo',
            'operation_metric',
            'trace',
            'log',
            'dependency_metric',
            'top_contributor',
            'service_quota',
        }
        invalid = [a for a in raw_a if a not in allowed]
        if invalid:
            raise ValueError(
                f'Invalid auditor(s): {", ".join(invalid)}. Allowed: {", ".join(sorted(allowed))}'
            )
        return raw_a


def expand_service_wildcard_patterns(
    targets: List[dict],
    unix_start: int,
    unix_end: int,
    next_token: Optional[str] = None,
    max_results: int = 5,
    applicationsignals_client=None,
) -> Tuple[List[dict], Optional[str], List[str], Dict[str, int]]:
    """Expand wildcard patterns for service targets with pagination support.

    Args:
        targets: List of target dictionaries
        unix_start: Start time as unix timestamp
        unix_end: End time as unix timestamp
        next_token: Token for pagination from previous list_services call
        max_results: Maximum number of services to return
        applicationsignals_client: AWS Application Signals client

    Returns:
        Tuple of (expanded_targets, next_token, all_service_names, filtering_stats)
        filtering_stats contains: {'total_services': int, 'instrumented_services': int, 'filtered_out': int}
    """
    from .utils import calculate_name_similarity

    if applicationsignals_client is None:
        from .aws_clients import applicationsignals_client

    expanded_targets = []
    service_patterns = []
    service_fuzzy_matches = []
    all_service_names = []
    filtering_stats = {'total_services': 0, 'instrumented_services': 0, 'filtered_out': 0}

    logger.debug(
        f'expand_service_wildcard_patterns_paginated: Processing {len(targets)} targets with max_results={max_results}'
    )
    logger.debug(f'Received next_token: {next_token is not None}')

    # First pass: identify patterns and collect non-wildcard targets
    for i, target in enumerate(targets):
        logger.debug(f'Target {i}: {target}')

        if not isinstance(target, dict):
            expanded_targets.append(target)
            continue

        target_type = target.get('Type', '').lower()
        logger.debug(f'Target {i} type: {target_type}')

        if target_type == 'service':
            # Check multiple possible locations for service name
            service_name = None

            # Check Data.Service.Name (full format)
            service_data = target.get('Data', {})
            if isinstance(service_data, dict):
                service_info = service_data.get('Service', {})
                if isinstance(service_info, dict):
                    service_name = service_info.get('Name', '')

            # Check shorthand Service field
            if not service_name:
                service_name = target.get('Service', '')

            logger.debug(f"Target {i} service name: '{service_name}'")

            if isinstance(service_name, str) and service_name:
                if '*' in service_name:
                    logger.debug(f"Target {i} identified as wildcard pattern: '{service_name}'")
                    service_patterns.append((target, service_name))
                else:
                    # Check if this might be a fuzzy match candidate
                    service_fuzzy_matches.append((target, service_name))
            else:
                logger.debug(f'Target {i} has no valid service name, passing through')
                expanded_targets.append(target)
        else:
            # Non-service targets pass through unchanged
            logger.debug(f'Target {i} is not a service target, passing through')
            expanded_targets.append(target)

    # Expand service patterns and fuzzy matches with pagination
    if service_patterns or service_fuzzy_matches:
        logger.debug(
            f'Expanding {len(service_patterns)} service wildcard patterns and {len(service_fuzzy_matches)} fuzzy matches with pagination'
        )
        try:
            # Use the common pagination function
            instrumented_services, returned_next_token, all_service_names, filtering_stats = (
                _fetch_instrumented_services_with_pagination(
                    unix_start, unix_end, next_token, max_results, applicationsignals_client
                )
            )

            # Handle wildcard patterns
            for original_target, pattern in service_patterns:
                matches_found = 0
                compiled_pattern = _compile_wildcard_pattern(pattern)
                for service in instrumented_services:
                    service_attrs = service.get('KeyAttributes', {})
                    service_name = service_attrs.get('Name', '')
                    environment = service_attrs.get('Environment', '')

                    # Apply wildcard pattern matching
                    if _matches_wildcard_pattern(service_name, compiled_pattern):
                        expanded_targets.append(_create_service_target(service_name, environment))
                        matches_found += 1
                        logger.debug(
                            f"Added instrumented service: Name='{service_name}', Environment='{environment}'"
                        )

                logger.debug(
                    f"Service pattern '{pattern}' expanded to {matches_found} instrumented targets in this batch"
                )

            # Handle fuzzy matches for inexact service names
            for original_target, inexact_name in service_fuzzy_matches:
                best_matches = []

                # Calculate similarity scores for all instrumented services
                for service in instrumented_services:
                    service_attrs = service.get('KeyAttributes', {})
                    service_name = service_attrs.get('Name', '')
                    if not service_name:
                        continue

                    score = calculate_name_similarity(inexact_name, service_name, 'service')

                    if score >= FUZZY_MATCH_THRESHOLD:  # Minimum threshold for consideration
                        best_matches.append(
                            (service_name, service_attrs.get('Environment'), score)
                        )

                # Sort by score and take the best matches
                best_matches.sort(key=lambda x: x[2], reverse=True)

                if best_matches:
                    # If we have a very high score match, use only that
                    if best_matches[0][2] >= HIGH_CONFIDENCE_MATCH_THRESHOLD:
                        matched_services = [best_matches[0]]
                    else:
                        # Otherwise, take top 3 matches above threshold
                        matched_services = best_matches[:3]

                    logger.info(
                        f"Fuzzy matching service '{inexact_name}' found {len(matched_services)} instrumented candidates in this batch:"
                    )
                    for service_name, environment, score in matched_services:
                        logger.info(f"  - '{service_name}' in '{environment}' (score: {score})")
                        expanded_targets.append(_create_service_target(service_name, environment))
                else:
                    logger.warning(
                        f"No fuzzy matches found for service name '{inexact_name}' (no candidates above threshold) in this batch"
                    )
                    # Keep the original target - let the API handle the error
                    expanded_targets.append(original_target)

            return (expanded_targets, returned_next_token, all_service_names, filtering_stats)

        except Exception as e:
            logger.warning(f'Failed to expand service patterns and fuzzy matches: {e}')
            # When expansion fails, we need to return an error rather than passing wildcards to validation
            # This prevents the validation phase from seeing wildcard patterns
            if service_patterns or service_fuzzy_matches:
                pattern_names = [pattern for _, pattern in service_patterns] + [
                    name for _, name in service_fuzzy_matches
                ]
                raise ValueError(
                    f'Failed to expand service wildcard patterns {pattern_names}. '
                    f'This may be due to AWS API access issues or missing services. '
                    f'Error: {str(e)}'
                )

    return expanded_targets, None, all_service_names, filtering_stats


def expand_slo_wildcard_patterns(
    targets: List[dict],
    next_token: Optional[str] = None,
    max_results: int = 5,
    applicationsignals_client=None,
) -> Tuple[List[dict], Optional[str], List[str]]:
    """Expand wildcard patterns for SLO targets with pagination support.

    Args:
        targets: List of target dictionaries
        next_token: Token for pagination from previous list_service_level_objectives call
        max_results: Maximum number of SLOs to return
        applicationsignals_client: AWS Application Signals client

    Returns:
        Tuple of (expanded_targets, next_token, slo_names_in_batch)
    """
    if applicationsignals_client is None:
        from .aws_clients import applicationsignals_client

    expanded_targets = []
    wildcard_patterns = []
    slo_names_in_batch = []

    for target in targets:
        if isinstance(target, dict):
            ttype = target.get('Type', '').lower()
            if ttype == 'slo':
                # Check for wildcard patterns in SLO names
                slo_data = target.get('Data', {}).get('Slo', {})

                # BUG FIX: Handle case where Slo is a string instead of dict
                if isinstance(slo_data, str):
                    # Malformed input - Slo should be a dict with SloName key
                    raise ValueError(
                        f"Invalid SLO target format. Expected {{'Type':'slo','Data':{{'Slo':{{'SloName':'name'}}}}}} "
                        f"but got {{'Slo':'{slo_data}'}}. The 'Slo' field must be a dictionary with 'SloName' key."
                    )
                elif isinstance(slo_data, dict):
                    slo_name = slo_data.get('SloName', '')
                else:
                    # Handle other unexpected types
                    raise ValueError(
                        f"Invalid SLO target format. The 'Slo' field must be a dictionary with 'SloName' key, "
                        f'but got {type(slo_data).__name__}: {slo_data}'
                    )

                if '*' in slo_name:
                    wildcard_patterns.append((target, slo_name))
                else:
                    expanded_targets.append(target)
            else:
                expanded_targets.append(target)
        else:
            expanded_targets.append(target)

    # Expand wildcard patterns for SLOs
    if wildcard_patterns:
        logger.debug(f'Expanding {len(wildcard_patterns)} SLO wildcard patterns')
        try:
            list_slos_params = {
                'MaxResults': max_results,
                'IncludeLinkedAccounts': True,
            }

            if next_token:
                list_slos_params['NextToken'] = next_token

            slos_response = applicationsignals_client.list_service_level_objectives(
                **list_slos_params
            )
            slos_batch = slos_response.get('SloSummaries', [])
            returned_next_token = slos_response.get('NextToken')

            # Collect all SLO names from this batch
            for slo in slos_batch:
                slo_name = slo.get('Name', '')
                slo_names_in_batch.append(slo_name)

            # Handle wildcard patterns
            for original_target, pattern in wildcard_patterns:
                matches_found = 0
                compiled_pattern = _compile_wildcard_pattern(pattern)
                for slo in slos_batch:
                    slo_name = slo.get('Name', '')
                    if _matches_wildcard_pattern(slo_name, compiled_pattern):
                        expanded_targets.append(
                            {
                                'Type': 'slo',
                                'Data': {
                                    'Slo': {'SloName': slo_name, 'SloArn': slo.get('Arn', '')}
                                },
                            }
                        )
                        matches_found += 1

                logger.debug(f"SLO pattern '{pattern}' expanded to {matches_found} targets")
            return expanded_targets, returned_next_token, slo_names_in_batch
        except Exception as e:
            logger.warning(f'Failed to expand SLO patterns: {e}')
            raise ValueError(f'Failed to expand SLO wildcard patterns. {str(e)}')

    return expanded_targets, None, slo_names_in_batch


def expand_service_operation_wildcard_patterns(
    targets: List[dict],
    unix_start: int,
    unix_end: int,
    next_token: Optional[str] = None,
    max_results: int = 5,
    applicationsignals_client=None,
) -> Tuple[List[dict], Optional[str], List[str], Dict[str, int]]:
    """Expand wildcard patterns for service operation targets with pagination support.

    Args:
        targets: List of target dictionaries
        unix_start: Start time as unix timestamp
        unix_end: End time as unix timestamp
        next_token: Token for pagination from previous list_services call
        max_results: Maximum number of services to return
        applicationsignals_client: AWS Application Signals client

    Returns:
        Tuple of (expanded_targets, next_token, all_service_names, filtering_stats)
        filtering_stats contains: {'total_services': int, 'instrumented_services': int, 'filtered_out': int}
    """
    if applicationsignals_client is None:
        from .aws_clients import applicationsignals_client

    expanded_targets = []
    wildcard_patterns = []
    all_service_names = []
    filtering_stats = {'total_services': 0, 'instrumented_services': 0, 'filtered_out': 0}

    for target in targets:
        if isinstance(target, dict):
            ttype = target.get('Type', '').lower()
            if ttype == 'service_operation':
                # Check for wildcard patterns in service names OR operation names
                service_op_data = target.get('Data', {}).get('ServiceOperation', {})
                service_data = service_op_data.get('Service', {})
                service_name = service_data.get('Name', '')
                operation = service_op_data.get('Operation', '')

                # Check if either service name or operation has wildcards
                if '*' in service_name or '*' in operation:
                    wildcard_patterns.append((target, service_name, operation))
                else:
                    expanded_targets.append(target)
            else:
                expanded_targets.append(target)
        else:
            expanded_targets.append(target)

    # Expand wildcard patterns for service operations
    if wildcard_patterns:
        logger.debug(
            f'Expanding {len(wildcard_patterns)} service operation wildcard patterns with pagination'
        )
        try:
            # Use the common pagination function
            instrumented_services, returned_next_token, all_service_names, filtering_stats = (
                _fetch_instrumented_services_with_pagination(
                    unix_start, unix_end, next_token, max_results, applicationsignals_client
                )
            )

            for original_target, service_pattern, operation_pattern in wildcard_patterns:
                compiled_service_pattern = _compile_wildcard_pattern(service_pattern)
                compiled_operation_pattern = _compile_wildcard_pattern(operation_pattern)
                matches_found = 0

                # Get the original metric type from the pattern
                service_op_data = original_target.get('Data', {}).get('ServiceOperation', {})
                metric_type = service_op_data.get('MetricType', 'Latency')

                # Find matching services from instrumented services only
                matching_services = []
                for service in instrumented_services:
                    service_attrs = service.get('KeyAttributes', {})
                    service_name = service_attrs.get('Name', '')

                    # Check if service matches the pattern using wildcard matching
                    if _matches_wildcard_pattern(service_name, compiled_service_pattern):
                        matching_services.append(service)

                logger.debug(
                    f"Found {len(matching_services)} instrumented services matching pattern '{service_pattern}'"
                )

                # For each matching service, get operations and expand operation patterns
                for service in matching_services:
                    service_attrs = service.get('KeyAttributes', {})
                    service_name = service_attrs.get('Name', '')
                    environment = service_attrs.get('Environment', '')

                    try:
                        # Get operations for this service
                        operations_response = applicationsignals_client.list_service_operations(
                            StartTime=datetime.fromtimestamp(unix_start, tz=timezone.utc),
                            EndTime=datetime.fromtimestamp(unix_end, tz=timezone.utc),
                            KeyAttributes=service_attrs,
                            MaxResults=100,
                        )

                        operations = operations_response.get('ServiceOperations', [])
                        logger.debug(
                            f"Found {len(operations)} operations for service '{service_name}'"
                        )

                        # Filter operations based on operation pattern
                        for operation in operations:
                            operation_name = operation.get('Name', '')

                            # Check if operation matches the pattern using wildcard matching
                            if _matches_wildcard_pattern(
                                operation_name, compiled_operation_pattern
                            ):
                                # Check if this operation has the required metric type
                                metric_refs = operation.get('MetricReferences', [])
                                has_metric_type = any(
                                    ref.get('MetricType', '').casefold() == metric_type.casefold()
                                    or (
                                        metric_type.casefold() == 'Availability'.casefold()
                                        and ref.get('MetricType', '') == 'FAULT'
                                    )
                                    for ref in metric_refs
                                )

                                if has_metric_type:
                                    service_target = _create_service_target(
                                        service_name, environment
                                    )
                                    expanded_targets.append(
                                        {
                                            'Type': 'service_operation',
                                            'Data': {
                                                'ServiceOperation': {
                                                    'Service': service_target['Data']['Service'],
                                                    'Operation': operation_name,
                                                    'MetricType': metric_type,
                                                }
                                            },
                                        }
                                    )
                                    matches_found += 1
                                    logger.debug(
                                        f'Added operation: {service_name} -> {operation_name} ({metric_type})'
                                    )
                                else:
                                    logger.debug(
                                        f'Skipping operation {operation_name} - no {metric_type} metric available'
                                    )

                    except Exception as e:
                        logger.warning(
                            f"Failed to get operations for service '{service_name}': {e}"
                        )
                        continue

                logger.debug(
                    f"Service operation pattern '{service_pattern}' + '{operation_pattern}' expanded to {matches_found} targets"
                )

            return (
                expanded_targets,
                returned_next_token,
                all_service_names,
                filtering_stats,
            )

        except Exception as e:
            logger.warning(f'Failed to expand service operation patterns: {e}')
            raise ValueError(f'Failed to expand service operation wildcard patterns. {str(e)}')

    return expanded_targets, None, all_service_names, filtering_stats


def _compile_wildcard_pattern(pattern: Optional[str]) -> Optional[re.Pattern]:
    """Compile wildcard pattern once for reuse.

    Args:
        pattern: Wildcard pattern with * for any characters

    Returns:
        Compiled regex pattern or None if pattern is invalid

    Examples:
        _compile_wildcard_pattern('hello*world') -> compiled regex for 'hello.*world'
        _compile_wildcard_pattern('*payment*') -> compiled regex for '.*payment.*'
        _compile_wildcard_pattern('*') -> compiled regex for '.*'
    """
    # Handle patterns that are empty or only contain wildcards (match everything)
    if pattern is None or pattern.strip('*') == '':
        # Empty or all-wildcard patterns match everything, including empty strings
        return re.compile('^.*$', re.IGNORECASE)

    # Escape special regex characters except *
    escaped = re.escape(pattern)

    # Replace escaped \* with regex .*
    regex_pattern = escaped.replace(r'\*', '.*')

    # Anchor the pattern to match the entire string
    regex_pattern = f'^{regex_pattern}$'

    return re.compile(regex_pattern, re.IGNORECASE)


def _matches_wildcard_pattern(text: Optional[str], compiled_pattern: Optional[re.Pattern]) -> bool:
    """Check if text matches pre-compiled wildcard pattern.

    Args:
        text: Text to test
        compiled_pattern: Pre-compiled regex pattern from _compile_wildcard_pattern

    Returns:
        True if text matches pattern

    Examples:
        pattern = _compile_wildcard_pattern('hello*world')
        _matches_wildcard_pattern('hello123world', pattern) -> True
        _matches_wildcard_pattern('helloworld', pattern) -> True
        _matches_wildcard_pattern('hello', pattern) -> False
    """
    if not compiled_pattern:
        return False

    # Handle case where text is None by treating it as empty string
    if text is None:
        text = ''

    return compiled_pattern.match(text) is not None


async def run_service_trace_audit(service_name: Optional[str], hours: int) -> str:
    """Run a trace-only Application Signals audit for one service (or all via wildcard).

    This is the importable seam used by the service_events ``get_incidents`` fallback: when
    no service_events service_events incidents are found but Application Signals is enabled,
    it asks Application Signals for trace-level findings. It performs the same
    single/wildcard service audit slice as the ``audit_services`` tool, restricted to
    the ``trace`` auditor.

    Args:
        service_name: Service name to audit. ``None`` audits all services (wildcard ``*``).
        hours: Look-back window in hours.

    Returns:
        The formatted audit result string from the Application Signals audit API.
    """
    from .aws_clients import AWS_REGION, applicationsignals_client
    from .service_audit_utils import (
        normalize_service_targets,
        validate_and_enrich_service_targets,
    )

    region = AWS_REGION.strip()
    end_dt = datetime.now(timezone.utc)
    unix_end = int(end_dt.timestamp())
    unix_start = unix_end - hours * 3600

    target_name = service_name or '*'
    provided = [{'Type': 'service', 'Data': {'Service': {'Type': 'Service', 'Name': target_name}}}]

    service_names_in_batch: List[str] = []
    if '*' in target_name:
        (provided, _next_token, service_names_in_batch, _stats) = expand_service_wildcard_patterns(
            provided,
            unix_start,
            unix_end,
            None,
            5,
            applicationsignals_client,
        )
        if not provided:
            return 'No services found matching the wildcard pattern.'

    normalized_targets = normalize_service_targets(provided)
    normalized_targets = validate_and_enrich_service_targets(
        normalized_targets, applicationsignals_client, unix_start, unix_end
    )

    auditors_list = parse_auditors('trace', ['trace'])

    banner = (
        '[MCP-SERVICE] Application Signals Trace Audit\n'
        f'🎯 Scope: {len(normalized_targets)} service target(s) | Region: {region}\n'
        f'⏰ Time: {unix_start}–{unix_end}\n\n'
    )

    input_obj = {
        'StartTime': unix_start,
        'EndTime': unix_end,
        'AuditTargets': normalized_targets,
    }
    if auditors_list:
        input_obj['Auditors'] = auditors_list

    return await execute_audit_api(input_obj, region, banner)
