Unlocking the power of Amazon Q Developer: Metrics-driven strategies for better AI coding

#!/usr/bin/env python3
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0
"""
Amazon Q Metrics Processor

This script processes user activity data and subscription information from S3,
combines them, and generates monthly CSV reports.
"""

import os
import boto3
import botocore
from botocore.config import Config
import pandas as pd
import numpy as np
from datetime import datetime
import logging
from io import StringIO
import re
import argparse
from pathlib import Path  # from pathlib import Path
import sys
from urllib.parse import urlparse

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('q-metrics')

def escape_log_data(data):
    """
    Sanitize data for secure logging to prevent log injection (CWE-117, CWE-93).
    
    Args:
        data: The data to be sanitized for logging
        
    Returns:
        str: Sanitized string safe for logging
    """
    if data is None:
        return 'None'
    
    # Convert to string if not already
    data_str = str(data)
    
    # Replace potentially dangerous characters
    # This prevents log forging by removing newlines and other control characters
    data_str = data_str.replace('\n', '\\n').replace('\r', '\\r')
    
    # Escape other control characters
    result = ''
    for char in data_str:
        if ord(char) < 32 or ord(char) == 127:  # Control characters
            result += f'\\x{ord(char):02x}'
        else:
            result += char
            
    return result

def validate_s3_bucket_name(bucket_name):
    """
    Validate S3 bucket name according to AWS naming rules.

    Args:
        bucket_name (str): The bucket name to validate

    Returns:
        bool: True if valid, False otherwise
    """
    if not bucket_name or not isinstance(bucket_name, str):
        return False

    # Check length (3-63 characters)
    if len(bucket_name)  63:
        return False

    # Check if it contains only allowed characters
    if not re.match(r'^[a-z0-9][a-z0-9\.-]*[a-z0-9]$', bucket_name):
        return False

    # Check if it doesn't contain consecutive periods
    if '..' in bucket_name:
        return False

    # Check if it's not an IP address
    if re.match(r'^\d+\.\d+\.\d+\.\d+$', bucket_name):
        return False

    # Check if it doesn't start with 'xn--' or end with '-s3alias'
    try:
        if bucket_name.startswith('xn--') or bucket_name.endswith('-s3alias'):
            return False
    except ValueError:
        return False

    return True

def validate_s3_path(path):
    """
    Validate an S3 path to ensure it doesn't contain dangerous characters.

    Args:
        path (str): The S3 path to validate

    Returns:
        bool: True if valid, False otherwise
    """
    if not path or not isinstance(path, str):
        return False

    # Check for path traversal attempts
    if '..' in path:
        return False

    # Check for invalid characters
    try:
        if re.search(r'[:"|?*\x00-\x1F]', path):
            return False
    except ValueError:
        return False

    return True

def validate_month_format(month_str):
    """
    Validate that a month string is in the format YYYY-MM.

    Args:
        month_str (str): The month string to validate

    Returns:
        bool: True if valid, False otherwise
    """
    if not month_str:
        return True  # Month is optional

    if not isinstance(month_str, str):
        return False

    # Check format YYYY-MM
    if not re.match(r'^[0-9]{4}-[0-9]{2}$', month_str):
        return False

    # Validate month range
    try:
        year, month = month_str.split('-')
        month_num = int(month)
        if month_num  12:
            return False
    except (ValueError, TypeError):
        return False

    return True

def sanitize_output_path(path):
    """
    Sanitize and normalize an output directory path.

    Args:
        path (str): The path to sanitize

    Returns:
        str: The sanitized absolute path
    """
    if not path or not isinstance(path, str):
        return os.path.abspath('./output')

    # Convert to Path object for safe handling
    safe_path = Path(path).resolve()

    # Ensure the path doesn't escape the intended directory structure
    try:
        # Make sure it's a valid path
        if not safe_path.is_absolute():
            safe_path = Path.cwd() / safe_path
    except (ValueError, TypeError):
        logger.warning("Invalid path: %s, using default", escape_log_data(repr(path)))
        return os.path.abspath('./output')

    return str(safe_path)

def validate_csv_content(df, expected_columns, file_type):
    """
    Validate that a DataFrame has the expected columns and structure.

    Args:
        df (DataFrame): The pandas DataFrame to validate
        expected_columns (list): List of required column names
        file_type (str): Type of file for logging purposes

    Returns:
        bool: True if valid, False otherwise
    """
    if df is None or df.empty:
        logger.error("Empty %s data", file_type)
        return False

    # Check for required columns
    missing_columns = [col for col in expected_columns if col not in df.columns]
    if missing_columns:
        logger.error("Missing required columns in %s data: %s", file_type, ', '.join(missing_columns))
        return False

    # Check for empty required columns
    for col in expected_columns:
        if df[col].isna().all():
            logger.error("Column '%s' in %s data is completely empty", col, file_type)
            return False

    # Check for reasonable row count
    if len(df) == 0:
        logger.error("No data rows in %s data", file_type)
        return False

    # Check for duplicate rows
    if df.duplicated().any():
        dup_count = df.duplicated().sum()
        logger.warning("Found %d duplicate rows in %s data", dup_count, file_type)

    return True

def validate_subscription_data(df):
    """
    Validate subscription data format and content.

    Args:
        df (DataFrame): The subscription data DataFrame

    Returns:
        bool: True if valid, False otherwise
    """
    required_columns = [
        'Name',
        'Subscription type',
        'Subscription status',
        'Identity provider user ID'
    ]

    return validate_csv_content(df, required_columns, "subscription")

def validate_activity_data(df):
    """
    Validate activity data format and content.

    Args:
        df (DataFrame): The activity data DataFrame

    Returns:
        bool: True if valid, False otherwise
    """
    required_columns = ['UserId', 'Date']

    # Basic validation
    if not validate_csv_content(df, required_columns, "activity"):
        return False

    # Check for at least one metric column
    metric_cols = df.select_dtypes(include=[np.number]).columns
    if len(metric_cols) == 0:
        logger.error("No metric columns found in activity data")
        return False

    # Check for valid date format
    try:
        valid_dates = 0
        total_dates = len(df['Date'])

        for date_str in df['Date']:
            if parse_date(date_str) is not None:
                valid_dates += 1

        if valid_dates == 0:
            logger.error("No valid dates found in activity data")
            return False
        elif valid_dates < total_dates:
            logger.warning("Only %d out of %d dates are valid in activity data", valid_dates, total_dates)
    except Exception as e:
        logger.error("Error validating dates in activity data: %s", str(e))
        return False

    return True



def parse_args():
    """Parse command line arguments with enhanced security validation."""
    parser = argparse.ArgumentParser(description='Process Amazon Q metrics data')
    parser.add_argument('--bucket', required=True, help='S3 bucket name')
    parser.add_argument('--prefix', required=True, help='Initial S3 prefix (e.g., "logs")')
    parser.add_argument('--subscription-path', required=True,
                        help='S3 path to subscription file (e.g., user-activities/subscriptions.csv)')
    parser.add_argument('--output-dir', default='./output', help='Local directory for output files')
    parser.add_argument('--month', help='Process only a specific month (format: YYYY-MM, e.g., 2025-01)')
    parser.add_argument('--per-user', action='store_true', help='Generate individual reports for each user')
    parser.add_argument('--user', help='Generate report for a specific user (by name)')

    args = parser.parse_args()

    # Validate bucket name with enhanced checks
    if not validate_s3_bucket_name(args.bucket):
        logger.error("Invalid S3 bucket name: %r", escape_log_data(args.bucket))
        sys.exit(1)
    
    # Additional bucket name security check - prevent command injection via bucket names
    if re.search(r'[;&|`$]', args.bucket):
        logger.error("Potentially malicious characters in bucket name: %r", escape_log_data(args.bucket))
        sys.exit(1)

    # Derive activity-prefix from the provided prefix
    try:
        # Get AWS account ID
        sts_client = boto3.client('sts')
        account_id = sts_client.get_caller_identity()['Account']
        
        # Get current region
        session = boto3.session.Session()
        region = session.region_name or os.environ.get('AWS_REGION', 'us-east-1')
        
        # Check if region is supported
        supported_regions = ['us-east-1', 'eu-central-1']
        if region not in supported_regions:
            logger.error("Region %s is not supported. This solution currently only works in %s", 
                        escape_log_data(region), ' or '.join(supported_regions))
            sys.exit(1)
        
        # Get current year
        current_year = str(datetime.now().year)
        
        # If month is provided, extract year from it
        if args.month:
            try:
                year, _ = args.month.split('-')
                current_year = year
            except (ValueError, TypeError):
                pass
                
        # Construct the activity prefix using the provided prefix
        activity_prefix = f"{args.prefix}/AWSLogs/{account_id}/QDeveloperLogs/by_user_analytic/{region}/{current_year}"
        logger.info("Derived activity prefix: %s", escape_log_data(activity_prefix))
    except Exception as e:
        logger.error("Failed to derive activity prefix: %s", escape_log_data(str(e)))
        sys.exit(1)
    
    # Validate S3 paths with enhanced checks
    for path_arg, path_name in [(activity_prefix, "activity prefix"), 
                               (args.subscription_path, "subscription path")]:
        if not validate_s3_path(path_arg):
            logger.error("Invalid S3 %s: %r", path_name, escape_log_data(path_arg))
            sys.exit(1)
        
        # Additional path security checks
        if re.search(r'[;&|`$]', path_arg):
            logger.error("Potentially malicious characters in %s: %r", path_name, escape_log_data(path_arg))
            sys.exit(1)
        
        # Check for path traversal attempts with more patterns
        if any(pattern in path_arg for pattern in ['../', '..\\', '../', '..\\']):
            logger.error("Path traversal attempt detected in %s: %r", path_name, escape_log_data(path_arg))
            sys.exit(1)

    # Validate month format if provided with enhanced checks
    if args.month:
        if not validate_month_format(args.month):
            logger.error("Invalid month format: %r. Expected format: YYYY-MM (e.g., 2025-01)", escape_log_data(args.month))
            sys.exit(1)
        
        # Additional validation for month - check for reasonable date range
        try:
            year, month = args.month.split('-')
            year_num = int(year)
            month_num = int(month)
            
            current_year = datetime.now().year
            
            # Check for reasonable year range (past 5 years to next year)
            if year_num  current_year + 1:
                logger.warning("Month year %d is outside the reasonable range (%d-%d)", 
                              year_num, current_year - 5, current_year + 1)
        except (ValueError, TypeError):
            # Already validated by validate_month_format, this is just an extra check
            pass

    # Validate user argument if provided
    if args.user:
        # Check for reasonable length
        if len(args.user) > 100:
            logger.error("User name too long: %d characters (max 100)", len(args.user))
            sys.exit(1)
        
        # Check for potentially dangerous characters
        if re.search(r'[;&|`$]', args.user):
            logger.error("Potentially malicious characters in user name: %r", escape_log_data(args.user))
            sys.exit(1)

    # Sanitize output directory with enhanced security
    args.output_dir = sanitize_output_path(args.output_dir)
    
    # Additional output directory security check
    try:
        output_path = Path(args.output_dir)
        
        # Check if the path exists and is a directory
        if output_path.exists() and not output_path.is_dir():
            logger.error("Output path exists but is not a directory: %r", escape_log_data(args.output_dir))
            sys.exit(1)
            
        # Check if we have write permissions to the directory or its parent
        parent_dir = output_path if output_path.exists() else output_path.parent
        if not os.access(parent_dir, os.W_OK):
            logger.error("No write permission for output directory: %r", escape_log_data(args.output_dir))
            sys.exit(1)
    except Exception as e:
        logger.error("Error validating output directory: %s", escape_log_data(str(e)))
        sys.exit(1)
        
    logger.info("Using output directory: %r", escape_log_data(args.output_dir))
    
    # Add the derived activity_prefix to args for use in the rest of the program
    args.activity_prefix = activity_prefix

    return args

def get_s3_client():
    """Create and return an S3 client with proper configuration."""
    try:
        # Configure S3 client with timeouts and retries
        config = Config(
            connect_timeout=5,  # 5 seconds connection timeout
            read_timeout=30,    # 30 seconds read timeout
            retries={'max_attempts': 3},  # Retry configuration
            signature_version='s3v4'  # Use more secure signature version
        )
        
        # Create the client with the custom configuration
        s3 = boto3.client('s3', config=config)
        
        # Test the client with a simple operation to validate credentials
        try:
            # Use head_bucket which is a lightweight operation
            s3.head_bucket(Bucket='aws-sdk-resources')
        except botocore.exceptions.ClientError as e:
            error_code = e.response.get('Error', {}).get('Code', '')
            if error_code == '403':
                logger.warning("AWS credentials appear valid but have insufficient permissions")
            elif error_code == '401':
                logger.error("Invalid AWS credentials")
                sys.exit(1)
        except Exception as e:
            logger.warning("Error testing S3 client: %s", str(e))
        
        return s3
    except Exception as e:
        logger.error("Failed to create S3 client: %s", str(e))
        sys.exit(1)

def list_activity_files(s3_client, bucket, prefix, month=None):
    """List all activity files in the specified S3 bucket and prefix.

    If month is specified, only return files for that month (format: YYYY-MM).

    Args:
        s3_client: The boto3 S3 client
        bucket (str): The S3 bucket name
        prefix (str): The S3 prefix for activity files
        month (str, optional): Month in YYYY-MM format

    Returns:
        list: List of S3 keys for activity files
    """
    # Validate inputs again for safety
    if not validate_s3_bucket_name(bucket):
        logger.error("Invalid bucket name: %r", escape_log_data(bucket))
        return []

    if not validate_s3_path(prefix):
        logger.error("Invalid S3 prefix: %r", escape_log_data(prefix))
        return []

    if month and not validate_month_format(month):
        logger.error("Invalid month format: %r", escape_log_data(month))
        return []

    # Sanitize prefix to ensure it doesn't have path traversal
    prefix = prefix.replace('..', '').replace('//', '/')

    logger.info("Listing activity files in s3://%s/%s", escape_log_data(bucket), escape_log_data(prefix))

    activity_files = []
    paginator = s3_client.get_paginator('list_objects_v2')

    try:
        # If month is specified, filter by the month prefix
        search_prefix = prefix
        if month:
            try:
                year, month_num = month.split('-')
                # Make sure we're not duplicating the year in the path
                if not prefix.endswith(year + '/'):
                    search_prefix = f"{prefix}{month_num}/"
                else:
                    search_prefix = f"{prefix}{month_num}/"
                logger.info("Filtering for month %s using prefix: %s", escape_log_data(month), escape_log_data(search_prefix))
            except ValueError:
                logger.error("Failed to parse month %r for filtering", escape_log_data(month))
                return []

        for page in paginator.paginate(Bucket=bucket, Prefix=search_prefix):
            if 'Contents' in page:
                for obj in page['Contents']:
                    key = obj['Key']
                    # Skip directories or non-CSV files
                    if not key.endswith('/') and key.endswith('.csv'):
                        activity_files.append(key)

        logger.info("Found %d activity files", len(activity_files))
        return activity_files
    except Exception as e:
        logger.error("Error listing activity files: %s", escape_log_data(str(e)))
        if isinstance(e, client('s3').exceptions.NoSuchBucket):
            logger.error("Bucket %r does not exist", escape_log_data(bucket))
        elif isinstance(e, client('s3').exceptions.AccessDenied):
            logger.error("Access denied to bucket %r", escape_log_data(bucket))
        return []

def read_csv_from_s3(s3_client, bucket, key):
    """Read a CSV file from S3 into a pandas DataFrame.

    Args:
        s3_client: The boto3 S3 client
        bucket (str): The S3 bucket name
        key (str): The S3 object key

    Returns:
        DataFrame or None: Pandas DataFrame with CSV content or None if error
    """
    # Validate inputs
    if not validate_s3_bucket_name(bucket):
        logger.error("Invalid bucket name: %r", escape_log_data(bucket))
        return None

    if not validate_s3_path(key):
        logger.error("Invalid S3 key: %r", escape_log_data(key))
        return None

    logger.info("Reading s3://%s/%s", escape_log_data(bucket), escape_log_data(key))

    try:
        # Get object metadata first to check file size
        try:
            head_response = s3_client.head_object(Bucket=bucket, Key=key)
            file_size_bytes = head_response.get('ContentLength', 0)
            
            # Set a reasonable size limit (e.g., 100MB)
            MAX_FILE_SIZE_BYTES = 100 * 1024 * 1024  # 100MB
            
            if file_size_bytes > MAX_FILE_SIZE_BYTES:
                logger.error("File %r size (%d bytes) exceeds the maximum allowed size (%d bytes)", 
                             escape_log_data(key), file_size_bytes, MAX_FILE_SIZE_BYTES)
                return None
        except Exception as e:
            logger.warning("Could not check file size for %r: %s", escape_log_data(key), escape_log_data(str(e)))
            # Continue anyway, we'll have other checks later

        response = s3_client.get_object(Bucket=bucket, Key=key)
        content = response['Body'].read().decode('utf-8')

        # Check for CSV injection attempts
        if any(suspicious_pattern in content for suspicious_pattern in ['=cmd|', '=cmd:', '@cmd', '+cmd', '-cmd', '=DDE', '=SUM(', '=HYPERLINK(']):
            logger.error("Potential CSV injection detected in %r", escape_log_data(key))
            return None

        # Use a StringIO buffer to safely parse the CSV
        csv_buffer = StringIO(content)

        # Read with error handling for malformed CSV
        try:
            # First check number of rows by reading just the header
            row_count = sum(1 for _ in csv_buffer) - 1  # Subtract 1 for header
            csv_buffer.seek(0)  # Reset buffer position
            
            # Set a hard limit on number of rows
            MAX_ROWS = 500000  # Hard limit of 500K rows
            
            if row_count > MAX_ROWS:
                logger.error("CSV file %r has %d rows, which exceeds the maximum limit of %d rows", 
                             escape_log_data(key), row_count, MAX_ROWS)
                return None
                
            # Use converters to sanitize string inputs
            df = pd.read_csv(csv_buffer, converters={col: str for col in ['Name', 'UserId', 'Subscription', 'Subscription type', 'Subscription status']})

            # Check memory usage of the DataFrame
            memory_usage_bytes = df.memory_usage(deep=True).sum()
            MAX_MEMORY_USAGE = 500 * 1024 * 1024  # 500MB limit
            
            if memory_usage_bytes > MAX_MEMORY_USAGE:
                logger.error("DataFrame for %r uses %d bytes of memory, exceeding the limit of %d bytes", 
                             escape_log_data(key), memory_usage_bytes, MAX_MEMORY_USAGE)
                del df  # Explicitly delete to free memory
                return None

            # Check for reasonable file size (soft warning)
            if len(df) > 100000:  # Lower the warning threshold to 100K rows
                logger.warning("CSV file %r has %d rows, which exceeds the recommended limit", escape_log_data(key), len(df))

            return df
        except pd.errors.ParserError as e:
            logger.error("Error parsing CSV %r: %s", escape_log_data(key), escape_log_data(str(e)))
            return None

    except Exception as e:
        logger.error("Error reading %r: %s", escape_log_data(key), escape_log_data(str(e)))
        return None

def parse_date(date_str):
    """Parse date strings into datetime objects."""
    if pd.isna(date_str) or date_str == 'N/A':
        return None

    # Handle different date formats
    date_formats = [
        '%m-%d-%Y',  # 02-03-2025
        '%B %d %Y',  # March 13 2025
    ]

    for fmt in date_formats:
        try:
            return datetime.strptime(date_str, fmt)
        except ValueError:
            continue

    logger.warning("Could not parse date: %r", escape_log_data(date_str))
    return None

def extract_month_year(date_obj):
    """Extract month and year from a datetime object."""
    if date_obj is None:
        return None
    try:
        return f"{date_obj.year}-{int(date_obj.month):02d}"
    except (AttributeError, ValueError, TypeError) as e:
        logger.debug("Failed to extract month/year from date object: %s", escape_log_data(str(e)))
        return None

def process_subscription_data(df):
    """Process subscription data."""
    logger.info("Processing subscription data")

    # Validate subscription data
    if not validate_subscription_data(df):
        logger.error("Invalid subscription data format")
        return None, {}

    # Clean up subscription data
    df_clean = df.copy()
    
    # Add 'Subscription' column if it doesn't exist (using Subscription type as default)
    if 'Subscription' not in df_clean.columns:
        df_clean['Subscription'] = df_clean['Subscription type']
        logger.info("Added 'Subscription' column based on 'Subscription type'")

    # Map user IDs to names
    id_to_name_map = dict(zip(df_clean['Identity provider user ID'], df_clean['Name']))

    # Parse last activity date
    df_clean['Last activity date'] = df_clean['Last activity date'].apply(parse_date)
    df_clean['Last activity month'] = df_clean['Last activity date'].apply(extract_month_year)

    return df_clean, id_to_name_map

def process_activity_data(activity_dfs, id_to_name_map):
    """Process and combine activity data."""
    logger.info("Processing activity data")

    if not activity_dfs:
        logger.warning("No activity data to process")
        return pd.DataFrame()

    # Validate each activity dataframe
    valid_dfs = []
    for i, df in enumerate(activity_dfs):
        if validate_activity_data(df):
            valid_dfs.append(df)
        else:
            logger.warning("Skipping invalid activity data file #%d", i+1)

    if not valid_dfs:
        logger.error("No valid activity data files found")
        return pd.DataFrame()

    try:
        # Combine all activity dataframes
        combined_df = pd.concat(valid_dfs, ignore_index=True)

        # Parse dates and extract month-year
        combined_df['Date'] = combined_df['Date'].apply(parse_date)
        combined_df['Month'] = combined_df['Date'].apply(extract_month_year)

        # Add user names based on ID mapping
        combined_df['Name'] = combined_df['UserId'].map(id_to_name_map)
        
        # Check if any user IDs couldn't be mapped to names
        unmapped_ids = combined_df[combined_df['Name'].isna()]['UserId'].unique()
        if len(unmapped_ids) > 0:
            logger.warning("Found %d user IDs without matching names", len(unmapped_ids))
            if len(unmapped_ids)  MAX_REPORTS:
        logger.warning("Too many months (%d). Limiting to %d most recent months.", 
                      len(months), MAX_REPORTS)
        # Sort months and take the most recent ones
        months = sorted(months, reverse=True)[:MAX_REPORTS]

    reports_generated = 0
    for month in months:
        if pd.isna(month):
            logger.debug("Skipping null month value")
            continue

        # Skip if specific_month is provided and doesn't match current month
        if specific_month and month != specific_month:
            continue

        # Validate month format
        if not validate_month_format(month):
            logger.warning("Skipping invalid month format: %r", escape_log_data(month))
            continue

        try:
            month_df = agg_df[agg_df['Month'] == month]

            # Create a safe filename
            safe_month = re.sub(r'[^\w\-]', '_', month)
            
            # Additional security check for filename
            if not safe_month or safe_month.startswith('.'):
                logger.warning("Invalid month format after sanitization: %r", escape_log_data(safe_month))
                continue
                
            output_file = os.path.join(output_dir, f"q-metrics-{safe_month}.csv")
            
            # Verify the output path is still within the intended directory
            output_path = Path(output_file).resolve()
            if not str(output_path).startswith(str(Path(output_dir).resolve())):
                logger.error("Path traversal attempt detected in output file: %r", escape_log_data(output_file))
                continue

            # Check if the dataframe is empty
            if month_df.empty:
                logger.warning("No data for month %r, skipping report generation", escape_log_data(month))
                continue
                
            # Limit the size of the output file
            if len(month_df) > 100000:
                logger.warning("Month %r has too many rows (%d). Truncating to 100,000 rows.", 
                              escape_log_data(month), len(month_df))
                month_df = month_df.head(100000)

            # Write with secure file handling
            temp_file = output_file + '.tmp'
            month_df.to_csv(temp_file, index=False)
            
            # Use atomic rename for safer file writing
            os.replace(temp_file, output_file)
            
            # Set secure permissions
            os.chmod(output_file, 0o640)  # rw-r-----
            
            logger.info("Generated report for %s: %s", escape_log_data(month), escape_log_data(output_file))
            reports_generated += 1
        except (PermissionError, OSError) as e:
            logger.error("Error writing to %r: %s", escape_log_data(output_file), escape_log_data(str(e)))
        except Exception as e:
            logger.error("Error generating report for month %r: %s", escape_log_data(month), escape_log_data(str(e)))

    return reports_generated

def generate_user_reports(agg_df, output_dir, specific_user=None):
    """Generate per-user CSV reports.

    If specific_user is provided, only generate a report for that user.

    Args:
        agg_df (DataFrame): Aggregated data frame
        output_dir (str): Output directory path
        specific_user (str, optional): Specific user name

    Returns:
        int: Number of reports generated
    """
    logger.info("Generating per-user reports")

    # Validate output directory
    output_dir = sanitize_output_path(output_dir)

    # Create output directory for user reports
    user_dir = os.path.join(output_dir, "users")
    try:
        os.makedirs(user_dir, exist_ok=True)
    except (PermissionError, OSError) as e:
        logger.error("Cannot create user directory %s: %s", escape_log_data(user_dir), escape_log_data(str(e)))
        return 0

    # Check if dataframe is empty
    if agg_df.empty:
        logger.warning("No data available to generate user reports")
        return 0

    # Check if Name column exists
    if 'Name' not in agg_df.columns:
        logger.error("Required column 'Name' not found in data")
        return 0

    # Get list of users
    users = agg_df['Name'].unique()
    if len(users) == 0:
        logger.warning("No user data available in the aggregated dataframe")
        return 0

    reports_generated = 0
    for user in users:
        if pd.isna(user):
            logger.debug("Skipping null user value")
            continue

        # Skip if specific_user is provided and doesn't match current user
        if specific_user and user != specific_user:
            continue

        try:
            user_df = agg_df[agg_df['Name'] == user]
            
            # Check if the dataframe is empty
            if user_df.empty:
                logger.warning("No data for user %r, skipping report generation", escape_log_data(user))
                continue

            # Create a safe filename from the user name using a more secure approach
            # Only allow alphanumeric characters, underscores, and hyphens
            safe_name = re.sub(r'[^\w\-]', '_', str(user))

            # Prevent directory traversal by removing any path components
            safe_name = os.path.basename(safe_name)

            # Ensure the filename is not empty and doesn't start with a dot
            if not safe_name or safe_name.startswith('.'):
                safe_name = f"user_{hash(user) % 10000}"

            output_file = os.path.join(user_dir, f"q-metrics-{safe_name}.csv")

            user_df.to_csv(output_file, index=False)
            logger.info("Generated report for user %r: %s", escape_log_data(user), escape_log_data(output_file))
            reports_generated += 1
        except (PermissionError, OSError) as e:
            logger.error("Error writing to %s: %s", escape_log_data(output_file), escape_log_data(str(e)))
        except Exception as e:
            logger.error("Error generating report for user %r: %s", escape_log_data(user), escape_log_data(str(e)))

    return reports_generated

def main():
    """Main function to process metrics data with enhanced security."""
    try:
        # Memory limit has been removed as requested
        
        # Parse and validate arguments
        args = parse_args()

        # Initialize S3 client with secure configuration
        s3_client = get_s3_client()

        # Set timeout for operations
        import signal
        
        def timeout_handler(signum, frame):
            logger.error("Operation timed out")
            sys.exit(1)
        
        # Set a global timeout of 10 minutes for the entire process
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(600)  # 600 seconds = 10 minutes

        # Read subscription data with integrity verification
        subscription_df = read_csv_from_s3(s3_client, args.bucket, args.subscription_path)
        if subscription_df is None:
            logger.error("Failed to read subscription data. Exiting.")
            return 1

        # Process subscription data
        subscription_df, id_to_name_map = process_subscription_data(subscription_df)
        if subscription_df is None:
            logger.error("Failed to process subscription data. Exiting.")
            return 1

        # List and read activity files
        activity_files = list_activity_files(s3_client, args.bucket, args.activity_prefix, args.month)
        if not activity_files:
            logger.warning("No activity files found. Proceeding with empty activity data.")
        
        # Limit the number of files processed to prevent resource exhaustion
        MAX_FILES = 100
        if len(activity_files) > MAX_FILES:
            logger.warning("Too many activity files found (%d). Limiting to %d files.", 
                          len(activity_files), MAX_FILES)
            activity_files = activity_files[:MAX_FILES]
        
        activity_dfs = []
        for file_key in activity_files:
            # Check for timeout or interruption between file processing
            if hasattr(signal, 'SIGINFO'):  # macOS specific
                signal.signal(signal.SIGINFO, lambda signum, frame: logger.info("Processing file: %s", file_key))
                
            df = read_csv_from_s3(s3_client, args.bucket, file_key)
            if df is not None:
                activity_dfs.append(df)
            else:
                logger.warning("Skipping invalid activity file: %r", file_key)

        # Process activity data
        activity_df = process_activity_data(activity_dfs, id_to_name_map)
        if activity_df.empty and activity_files:
            logger.warning("No valid activity data could be processed from %d files", len(activity_files))

        # Merge data
        merged_df = merge_data(activity_df, subscription_df)

        # Aggregate by month
        agg_df = aggregate_by_month(merged_df)
        if agg_df.empty:
            logger.warning("No data available after aggregation. Reports will be empty.")

        # Generate monthly reports
        num_reports = generate_monthly_reports(agg_df, args.output_dir, args.month)
        logger.info("Successfully generated %d monthly reports", num_reports)

        # Generate per-user reports if requested
        if args.per_user or args.user:
            num_user_reports = generate_user_reports(agg_df, args.output_dir, args.user)
            logger.info("Successfully generated %d user reports", num_user_reports)
            
        # Cancel the alarm since we're done
        signal.alarm(0)
        
        return 0

    except Exception as e:
        # Limit error disclosure in logs
        logger.error("Error processing metrics: %s", escape_log_data(str(e)))
        # Only log full traceback in debug mode
        if logger.level <= logging.DEBUG:
            logger.debug("Detailed error information:", exc_info=True)
        return 1
    finally:
        # Ensure we cancel any pending alarms
        if 'signal' in locals():
            try:
                signal.alarm(0)
            except:
                pass

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

Create a Python script that processes Amazon Q Developer metrics data from S3. 

1. Input Sources:
    - User Activity Data:
        - Location: s3://YOURBUCKET/USER_DATA_KEY
        - Structure: CSV files organized in month/day folders
        - Key field: UserId
        - Date field: format MM-DD-YYYY
    - User Subscription Data:
        - Location: s3://YOURBUCKET/Amazon_Q_Users_Subscription_List.csv
        - Key field: Identity provider user ID (maps to UserId in activity data)
2. Processing Requirements:
    - Read all CSV files from the activity data folder structure
    - Join with subscription data using UserId/'Identity provider user ID'
    - Aggregate data by user and month using the Date field
    - Calculate monthly totals for all metrics
    - Include security features like input validation and sanitization
    - Replace NaN with 0
    - Some columns does not exists in all csv files
    - Use boto3 to interact with AWS S3
    - Use pandas for data processing
    - Include proper error handling and logging
    - Validate all inputs to prevent security issues
3. Script Requirements:
    - The script should be well-structured with separate functions
    - Argument parsing and validation
    - S3 client initialization
    - File listing and reading from S3
    - Data processing and aggregation
    - Report generation
4. Support command-line arguments for:
   - S3 bucket name
   - S3 prefix for logs
   - Path to subscription file
   - Output directory for reports
   - Optional filtering by month (YYYY-MM format)
   - Optional per-user report generation
   - Optional filtering for a specific user