generate_sales_data.py

#!/usr/bin/env python3
"""
Messy Sales Transaction Dataset Generator

Generates intentionally messy, semi-structured sales transaction data for learning purposes.
Demonstrates common data quality issues found in poorly maintained production systems.

Usage:
    python generate_sales_data.py --rows 1000000 --format jsonl --output my_sales_data
    python generate_sales_data.py --rows 10000 --format csv --output sales
    python generate_sales_data.py --format parquet --output data
"""

import random
import json
import csv
import argparse
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple
import os
import math


class DataGenerator:
    """Generates messy sales transaction records with various data quality issues."""
    
    # Product catalog with IDs, categories, and base prices
    PRODUCTS = {
        1: {"name": "Wireless Mouse", "category": "accessories", "base_price": 29.99, "popularity": 0.15},
        2: {"name": "Mechanical Keyboard", "category": "accessories", "base_price": 89.99, "popularity": 0.12},
        3: {"name": "LED Monitor", "category": "electronics", "base_price": 299.99, "popularity": 0.08},
        4: {"name": "USB-C Hub", "category": "accessories", "base_price": 39.99, "popularity": 0.10},
        5: {"name": "Laptop Stand", "category": "accessories", "base_price": 49.99, "popularity": 0.07},
        6: {"name": "Desk Chair", "category": "furniture", "base_price": 249.99, "popularity": 0.05},
        7: {"name": "Notebook Set", "category": "office_supplies", "base_price": 12.99, "popularity": 0.09},
        8: {"name": "Pen Collection", "category": "office_supplies", "base_price": 8.99, "popularity": 0.11},
        9: {"name": "Stapler", "category": "office_supplies", "base_price": 6.99, "popularity": 0.04},
        10: {"name": "Coffee Mug", "category": "office_supplies", "base_price": 9.99, "popularity": 0.06},
        11: {"name": "Water Bottle", "category": "office_supplies", "base_price": 14.99, "popularity": 0.08},
        12: {"name": "Desk Lamp", "category": "accessories", "base_price": 34.99, "popularity": 0.07},
        13: {"name": "Headphones", "category": "electronics", "base_price": 79.99, "popularity": 0.13},
        14: {"name": "Webcam", "category": "electronics", "base_price": 69.99, "popularity": 0.09},
        15: {"name": "Mouse Pad", "category": "accessories", "base_price": 15.99, "popularity": 0.10},
        16: {"name": "Cable Organizer", "category": "accessories", "base_price": 11.99, "popularity": 0.05},
        17: {"name": "Phone Stand", "category": "accessories", "base_price": 19.99, "popularity": 0.08},
        18: {"name": "Desk Mat", "category": "accessories", "base_price": 24.99, "popularity": 0.06},
        19: {"name": "Monitor Arm", "category": "furniture", "base_price": 129.99, "popularity": 0.04},
        20: {"name": "Ergonomic Keyboard", "category": "accessories", "base_price": 119.99, "popularity": 0.07},
    }
    
    # Typo variations for product names
    TYPO_VARIATIONS = {
        "Wireless Mouse": ["Wireles Mouse", "Wireless Mose", "Wirelss Mouse"],
        "Mechanical Keyboard": ["Mechanical Keybord", "Mecanical Keyboard", "Mech Keyboard"],
        "LED Monitor": ["LED Moniter", "LED Monitr", "L.E.D. Monitor"],
        "Headphones": ["Headphons", "Head Phones", "Headfones"],
        "Coffee Mug": ["Coffe Mug", "Coffee Mag", "Cofee Mug"],
        "Desk Chair": ["Desk Chiar", "Deskchair", "Desk Char"],
        "Notebook Set": ["Notbook Set", "Note Book Set", "Notebook St"],
        "Webcam": ["Web Cam", "Webkam", "Web-Cam"],
    }
    
    # Seasonal multipliers by month (1-12)
    SEASONAL_MULTIPLIERS = {
        1: 0.65,   # January - Post-holiday slump
        2: 0.70,   # February - Still slow
        3: 0.85,   # March - Recovery
        4: 0.95,   # April - Spring
        5: 1.00,   # May - Normal
        6: 0.80,   # June - Summer slowdown starts
        7: 0.75,   # July - Summer low
        8: 1.20,   # August - Back to school prep
        9: 1.30,   # September - Back to school peak
        10: 1.10,  # October - Fall
        11: 1.50,  # November - Black Friday/Holiday prep
        12: 1.80,  # December - Holiday peak
    }
    
    # Day of week multipliers (0=Monday, 6=Sunday)
    DAY_OF_WEEK_MULTIPLIERS = {
        0: 0.85,   # Monday - Slow start
        1: 1.00,   # Tuesday - Normal
        2: 1.05,   # Wednesday - Mid-week peak
        3: 1.10,   # Thursday - Building up
        4: 1.35,   # Friday - Peak day
        5: 1.25,   # Saturday - Weekend shopping
        6: 0.90,   # Sunday - Lower activity
    }
    
    # Payment methods
    PAYMENT_METHODS = ["Credit Card", "Debit Card", "PayPal", "Cash", "Bank Transfer"]
    
    # Payment method variations
    PAYMENT_VARIATIONS = {
        "Credit Card": ["CC", "credit card", "CREDIT CARD", "Cradit Card", "Visa", "Mastercard", "Credit  Card"],
        "Debit Card": ["DC", "debit card", "DEBIT CARD", "Debit Crd", "Debit"],
        "PayPal": ["PP", "paypal", "PAYPAL", "Paypal", "Pay Pal"],
        "Cash": ["cash", "CASH", "Cach", "CSH"],
        "Bank Transfer": ["BT", "bank transfer", "BANK TRANSFER", "Wire Transfer", "Bank Trnsfr"],
    }
    
    def __init__(self, seed: Optional[int] = None, messiness_level: int = 35, column_config: Optional[Dict[str, Dict[str, Any]]] = None):
        """
        Initialize the data generator.
        
        Args:
            seed: Random seed for reproducibility
            messiness_level: Default percentage of records that should have data quality issues (0-100)
            column_config: Optional dictionary mapping column names to their configuration
                          e.g., {"product_id": {"messiness": 0}, "product_name": {"messiness": 70}}
        """
        if seed is not None:
            random.seed(seed)
        
        self.messiness_level = messiness_level / 100.0
        self.column_config = column_config or {}
        
        self.stats = {
            "total_records": 0,
            "messy_product_id": 0,
            "messy_product_name": 0,
            "messy_timestamp": 0,
            "messy_amount": 0,
            "messy_payment_method": 0,
            "duplicates": 0,
        }
    
    def _should_be_messy(self, column_name: str) -> bool:
        """
        Determine if this field should have a data quality issue.
        
        Args:
            column_name: Name of the column to check
            
        Returns:
            True if the field should be messy, False otherwise
        """
        # Check if column has specific configuration
        if column_name in self.column_config:
            column_messiness = self.column_config[column_name].get("messiness", self.messiness_level * 100) / 100.0
            return random.random() < column_messiness
        
        # Fall back to global messiness level
        return random.random() < self.messiness_level
    
    def _get_seasonal_multiplier(self, dt: datetime) -> float:
        """
        Get the seasonal sales multiplier for a given date.
        
        Args:
            dt: The datetime to get the multiplier for
            
        Returns:
            Seasonal multiplier (0.65 to 1.80)
        """
        return self.SEASONAL_MULTIPLIERS.get(dt.month, 1.0)
    
    def _get_day_of_week_multiplier(self, dt: datetime) -> float:
        """
        Get the day-of-week sales multiplier.
        
        Args:
            dt: The datetime to get the multiplier for
            
        Returns:
            Day of week multiplier (0.85 to 1.35)
        """
        return self.DAY_OF_WEEK_MULTIPLIERS.get(dt.weekday(), 1.0)
    
    def _get_time_of_day_weight(self, hour: int) -> float:
        """
        Get probability weight for a given hour of day.
        Business hours (9-17) are more likely.
        
        Args:
            hour: Hour of day (0-23)
            
        Returns:
            Weight for this hour
        """
        if 9 <= hour <= 17:
            return 3.0  # Business hours - 3x more likely
        elif 18 <= hour <= 21:
            return 1.5  # Evening - 1.5x more likely
        else:
            return 0.5  # Night/early morning - 0.5x less likely
    
    def _select_weighted_product(self, dt: datetime = None) -> int:
        """
        Select a product ID using weighted random selection based on popularity.
        Adds daily variation so not all products are sold every day.
        
        Args:
            dt: Optional datetime to create date-based product availability
        
        Returns:
            Product ID (1-20)
        """
        products = list(self.PRODUCTS.keys())
        weights = [self.PRODUCTS[pid]["popularity"] for pid in products]
        
        # Add daily product availability variation
        # Use date as seed component to make certain products less available on certain days
        if dt:
            date_seed = dt.year * 10000 + dt.month * 100 + dt.day
            # For each product, determine if it's "in stock" today
            # Less popular products have higher chance of being out of stock
            available_products = []
            available_weights = []
            
            for pid, weight in zip(products, weights):
                # Use deterministic randomness based on date and product
                # This ensures same products are unavailable on same dates across runs
                availability_seed = (date_seed * pid) % 100
                # Products with lower popularity have higher chance of being unavailable
                # Popularity ranges from 0.04 to 0.15, so we scale this
                availability_threshold = 70 + (weight * 100)  # Range: 74-85
                
                if availability_seed < availability_threshold:
                    available_products.append(pid)
                    available_weights.append(weight)
            
            # If we filtered out too many products, use all products
            if len(available_products) < 10:
                return random.choices(products, weights=weights, k=1)[0]
            
            # Normalize weights for available products
            total_weight = sum(available_weights)
            normalized_weights = [w / total_weight for w in available_weights]
            
            return random.choices(available_products, weights=normalized_weights, k=1)[0]
        
        return random.choices(products, weights=weights, k=1)[0]
    
    def generate_product_id(self, clean_id: int) -> Any:
        """
        Generate a product ID with potential inconsistencies.
        
        Args:
            clean_id: The clean product ID number
            
        Returns:
            Product ID in various formats (clean or messy)
        """
        if not self._should_be_messy("product_id"):
            return f"PROD-{clean_id:04d}"
        
        self.stats["messy_product_id"] += 1
        
        patterns = [
            lambda: f"{clean_id:04d}",                          # No prefix
            lambda: f"P{clean_id:04d}",                         # Short prefix
            lambda: f"product_{clean_id:04d}",                  # Underscore format
            lambda: f" PROD-{clean_id:04d} ",                   # Extra whitespace
            lambda: f"{clean_id}",                              # No leading zeros
            lambda: f"prod-{clean_id:04d}",                     # Lowercase
            lambda: random.choice(["NULL", "N/A", "", None]),   # Missing values
            lambda: random.choice(["unknown", "TBD", "???"]),   # Invalid values
        ]
        
        return random.choice(patterns)()
    
    def generate_product_name(self, product_id: int) -> Any:
        """
        Generate a product name with potential quality issues.
        
        Args:
            product_id: The product ID to get the name for
            
        Returns:
            Product name with various quality issues
        """
        # Get base product name from new structure
        product_info = self.PRODUCTS.get(product_id, {"name": "Unknown Product"})
        base_name = product_info["name"]
        
        if not self._should_be_messy("product_name"):
            return base_name
        
        self.stats["messy_product_name"] += 1
        
        patterns = [
            lambda: base_name.lower(),                                          # Lowercase
            lambda: base_name.upper(),                                          # Uppercase
            lambda: f"  {base_name}  ",                                        # Extra whitespace
            lambda: f"{base_name} ",                                           # Trailing space
            lambda: f" {base_name}",                                           # Leading space
            lambda: base_name.replace(" ", "  "),                              # Double spaces
            lambda: random.choice(self.TYPO_VARIATIONS.get(base_name, [base_name])),  # Typos
            lambda: base_name[:3] if len(base_name) > 3 else base_name,       # Abbreviation
            lambda: random.choice(["N/A", "", None, "Unknown Product"]),       # Missing
        ]
        
        return random.choice(patterns)()
    
    def generate_timestamp(self) -> Tuple[Any, datetime]:
        """
        Generate a sales timestamp with various format inconsistencies.
        Uses weighted selection for time of day to create realistic patterns.
        
        Returns:
            Tuple of (formatted timestamp, datetime object for use in other calculations)
        """
        # Generate random datetime in last 2 years
        days_ago = random.randint(0, 730)
        
        # Weighted hour selection (business hours more likely)
        hour_weights = [self._get_time_of_day_weight(h) for h in range(24)]
        hours = random.choices(range(24), weights=hour_weights, k=1)[0]
        
        minutes = random.randint(0, 59)
        seconds = random.randint(0, 59)
        
        dt = datetime.now() - timedelta(days=days_ago, hours=hours, minutes=minutes, seconds=seconds)
        
        if not self._should_be_messy("sales_timestamp"):
            return (dt.isoformat() + "Z", dt)
        
        self.stats["messy_timestamp"] += 1
        
        # Reduced to 4 common formats + invalid (easier to handle in cleaning)
        formats = [
            lambda: dt.strftime("%m/%d/%Y %I:%M %p"),                          # US format with AM/PM
            lambda: dt.strftime("%d/%m/%Y %H:%M"),                             # EU format (24h)
            lambda: dt.strftime("%Y-%m-%d"),                                   # Date only (ISO)
            lambda: str(int(dt.timestamp())),                                  # Unix timestamp (seconds)
            lambda: random.choice(["invalid date", "TBD", None, "", "N/A"]),  # Invalid (~12.5% of messy)
        ]
        
        return (random.choice(formats)(), dt)
    
    def generate_amount(self, product_id: int, dt: datetime) -> Any:
        """
        Generate a sales amount with various format inconsistencies.
        Now uses product base price with realistic variation and seasonal adjustments.
        
        Args:
            product_id: The product ID to get base price for
            dt: The datetime for seasonal adjustments
            
        Returns:
            Amount in various formats (float, string with currency, etc.)
        """
        # Get base price for product
        product_info = self.PRODUCTS.get(product_id, {"base_price": 50.0})
        base_price = product_info["base_price"]
        
        # Add random variation (±20% of base price)
        price_variation = random.uniform(0.80, 1.20)
        
        # Apply seasonal multiplier
        seasonal_mult = self._get_seasonal_multiplier(dt)
        
        # Apply day of week multiplier (affects quantity/demand, reflected in slight price adjustments)
        dow_mult = self._get_day_of_week_multiplier(dt)
        
        # Combine all factors (seasonal has more impact than day-of-week on pricing)
        amount = base_price * price_variation * (0.9 + 0.1 * seasonal_mult) * (0.95 + 0.05 * dow_mult)
        amount = round(amount, 2)
        
        if not self._should_be_messy("sales_amount"):
            return amount
        
        self.stats["messy_amount"] += 1
        
        formats = [
            lambda: f"${amount:.2f}",                                          # With $ symbol
            lambda: f"USD {amount:.2f}",                                       # USD prefix
            lambda: f"{amount:.2f} USD",                                       # USD suffix
            lambda: f"€{amount * 0.85:.2f}",                                   # Euro
            lambda: f"£{amount * 0.75:.2f}",                                   # Pound
            lambda: f"{amount:,.2f}",                                          # Thousands separator
            lambda: f" {amount:.2f} ",                                         # Extra whitespace
            lambda: f"{amount:.1f}",                                           # One decimal
            lambda: f"{int(amount)}",                                          # No decimals
            lambda: f"-{amount:.2f}",                                          # Negative (refund)
            lambda: random.choice(["free", "N/A", None, "", "0", "pending"]), # Invalid
        ]
        
        return random.choice(formats)()
    
    def generate_payment_method(self, amount: float) -> Any:
        """
        Generate a payment method with various inconsistencies.
        Payment method selection is influenced by transaction amount.
        
        Args:
            amount: The transaction amount (influences payment method choice)
            
        Returns:
            Payment method with variations (simplified to common cases)
        """
        # Payment method preferences based on amount
        if amount < 50:
            # Small purchases: more cash and debit
            weights = [0.15, 0.35, 0.20, 0.25, 0.05]  # CC, DC, PayPal, Cash, Bank Transfer
        elif amount < 200:
            # Medium purchases: balanced
            weights = [0.30, 0.30, 0.20, 0.15, 0.05]
        else:
            # Large purchases: more credit cards
            weights = [0.50, 0.25, 0.15, 0.05, 0.05]
        
        base_method = random.choices(self.PAYMENT_METHODS, weights=weights, k=1)[0]
        
        if not self._should_be_messy("payment_method"):
            return base_method
        
        self.stats["messy_payment_method"] += 1
        
        variations = self.PAYMENT_VARIATIONS.get(base_method, [base_method])
        
        # Simplified: 70% use variation, 20% whitespace, 10% invalid
        rand = random.random()
        if rand < 0.70:
            return random.choice(variations)                                   # Use variation (most common)
        elif rand < 0.90:
            return f" {base_method} "                                          # Extra whitespace
        else:
            return random.choice(["NULL", "N/A", None, "", "unknown"])        # Invalid/missing
    
    def generate_record(self, record_id: int) -> Dict[str, Any]:
        """
        Generate a single sales transaction record with realistic patterns.
        
        Args:
            record_id: Unique identifier for this record
            
        Returns:
            Dictionary containing the sales transaction data
        """
        self.stats["total_records"] += 1
        
        # Generate timestamp first (returns tuple of formatted timestamp and datetime object)
        timestamp_formatted, dt = self.generate_timestamp()
        
        # Use weighted product selection based on popularity and date-based availability
        product_id_num = self._select_weighted_product(dt)
        
        # Generate amount based on product and datetime (for seasonal adjustments)
        amount = self.generate_amount(product_id_num, dt)
        
        # Extract numeric amount for payment method selection
        if isinstance(amount, (int, float)):
            numeric_amount = amount
        else:
            # If amount is messy (string/None), use a default for payment method selection
            numeric_amount = 50.0
        
        record = {
            "product_id": self.generate_product_id(product_id_num),
            "product_name": self.generate_product_name(product_id_num),
            "sales_timestamp": timestamp_formatted,
            "sales_amount": amount,
            "payment_method": self.generate_payment_method(numeric_amount),
        }
        
        # Occasionally add schema variations (extra fields or missing fields)
        if random.random() < 0.05:
            # Add extra unexpected fields
            extra_fields = {
                "notes": random.choice(["Customer requested gift wrap", "", None]),
                "customer_id": random.randint(1000, 9999),
                "location": random.choice(["Store A", "Store B", "Online"]),
            }
            record.update(random.choice([extra_fields, {}]))
        
        if random.random() < 0.03:
            # Remove a random field to create schema inconsistency
            # Only remove fields that have messiness > 0 (respect column configuration)
            removable_fields = []
            for field_name in record.keys():
                # Check if field can be removed based on messiness configuration
                if field_name in self.column_config:
                    field_messiness = self.column_config[field_name].get("messiness", self.messiness_level * 100)
                    if field_messiness > 0:
                        removable_fields.append(field_name)
                elif self.messiness_level > 0:
                    # If no specific config, use global messiness level
                    removable_fields.append(field_name)
            
            # Only remove a field if there are removable fields
            if removable_fields:
                field_to_remove = random.choice(removable_fields)
                del record[field_to_remove]
        
        return record


class OutputWriter:
    """Handles writing records to various output formats."""
    
    @staticmethod
    def write_jsonl(records: List[Dict[str, Any]], filename: str) -> None:
        """
        Write records to JSON Lines format (one JSON object per line).
        
        Args:
            records: List of record dictionaries
            filename: Output filename
        """
        with open(filename, 'w', encoding='utf-8') as f:
            for record in records:
                json.dump(record, f, default=str)
                f.write('\n')
    
    @staticmethod
    def write_csv(records: List[Dict[str, Any]], filename: str) -> None:
        """
        Write records to CSV format.
        
        Args:
            records: List of record dictionaries
            filename: Output filename
        """
        if not records:
            return
        
        # Get all possible field names from all records
        fieldnames = set()
        for record in records:
            fieldnames.update(record.keys())
        fieldnames = sorted(fieldnames)
        
        with open(filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(records)
    
    @staticmethod
    def write_parquet(records: List[Dict[str, Any]], filename: str) -> None:
        """
        Write records to Parquet format.
        
        Args:
            records: List of record dictionaries
            filename: Output filename
        """
        try:
            import pandas as pd
            df = pd.DataFrame(records)
            df.to_parquet(filename, engine='pyarrow', index=False)
        except ImportError:
            print("Warning: pandas or pyarrow not available. Cannot write Parquet format.")
            print("Install with: pip install pandas pyarrow")


def add_duplicates(records: List[Dict[str, Any]], duplicate_rate: float = 0.05) -> List[Dict[str, Any]]:
    """
    Add duplicate records to simulate poor data maintenance.
    
    Args:
        records: Original list of records
        duplicate_rate: Percentage of records to duplicate (0.0-1.0)
        
    Returns:
        List with duplicates added
    """
    num_duplicates = int(len(records) * duplicate_rate)
    
    for _ in range(num_duplicates):
        if records:
            # Pick a random record to duplicate
            original = random.choice(records)
            
            # 70% exact duplicate, 30% near duplicate with slight variation
            if random.random() < 0.7:
                duplicate = original.copy()
            else:
                # Near duplicate - change one field slightly
                duplicate = original.copy()
                field_to_modify = random.choice(list(duplicate.keys()))
                if field_to_modify == "sales_amount" and isinstance(duplicate[field_to_modify], (int, float)):
                    duplicate[field_to_modify] = duplicate[field_to_modify] + random.uniform(-0.01, 0.01)
            
            records.append(duplicate)
    
    return records


def print_statistics(stats: Dict[str, int], generation_time: float, filename: str) -> None:
    """
    Print generation statistics.
    
    Args:
        stats: Statistics dictionary from DataGenerator
        generation_time: Time taken to generate data
        filename: Output filename
    """
    print("\n" + "="*60)
    print("MESSY SALES TRANSACTION DATASET - GENERATION REPORT")
    print("="*60)
    
    print(f"\n📊 Generation Summary:")
    print(f"  Total records generated: {stats['total_records']:,}")
    print(f"  Generation time: {generation_time:.2f} seconds")
    print(f"  Records per second: {stats['total_records']/generation_time:,.0f}")
    
    if os.path.exists(filename):
        file_size = os.path.getsize(filename)
        file_size_mb = file_size / (1024 * 1024)
        print(f"  Output file: {filename}")
        print(f"  File size: {file_size_mb:.2f} MB")
    
    print(f"\n🔍 Data Quality Issues:")
    total = stats['total_records']
    print(f"  Messy product_id: {stats['messy_product_id']:,} ({stats['messy_product_id']/total*100:.1f}%)")
    print(f"  Messy product_name: {stats['messy_product_name']:,} ({stats['messy_product_name']/total*100:.1f}%)")
    print(f"  Messy sales_timestamp: {stats['messy_timestamp']:,} ({stats['messy_timestamp']/total*100:.1f}%)")
    print(f"  Messy sales_amount: {stats['messy_amount']:,} ({stats['messy_amount']/total*100:.1f}%)")
    print(f"  Messy payment_method: {stats['messy_payment_method']:,} ({stats['messy_payment_method']/total*100:.1f}%)")
    print(f"  Duplicate records: {stats['duplicates']:,} ({stats['duplicates']/total*100:.1f}%)")
    
    print("\n💡 Next Steps:")
    print("  1. Load this dataset into your data processing tool")
    print("  2. Analyze the data quality issues")
    print("  3. Build transformation pipelines to clean the data")
    print("  4. Practice handling missing values, type conversions, and standardization")
    
    print("\n" + "="*60 + "\n")


def load_column_config(config_file: str) -> Dict[str, Dict[str, Any]]:
    """
    Load column configuration from JSON file.
    
    Args:
        config_file: Path to JSON configuration file
        
    Returns:
        Dictionary mapping column names to their configuration
    """
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            config = json.load(f)
            return config.get("columns", {})
    except FileNotFoundError:
        print(f"Error: Configuration file '{config_file}' not found")
        return {}
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in configuration file: {e}")
        return {}


def parse_arguments():
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Generate messy sales transaction dataset for learning purposes",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Generate 1M records in JSON Lines format (default)
  python generate_sales_data.py
  
  # Generate 10K records in CSV format with custom filename
  python generate_sales_data.py --rows 10000 --format csv --output sales_data
  
  # Generate with custom messiness level and seed
  python generate_sales_data.py --rows 5000 --messiness 50 --seed 42 --output test_data
  
  # Generate with per-column messiness control
  python generate_sales_data.py --rows 10000 --config configs/ids_clean_rest_messy.json
  
  # Generate Parquet file with custom output name
  python generate_sales_data.py --format parquet --output my_sales_data
  
  # Generate small sample for testing
  python generate_sales_data.py --rows 100 --seed 42 --output sample
        """
    )
    
    parser.add_argument(
        "--rows",
        type=int,
        default=1_000_000,
        help="Number of records to generate (default: 1,000,000)"
    )
    
    parser.add_argument(
        "--format",
        choices=["jsonl", "csv", "parquet"],
        default="jsonl",
        help="Output format (default: jsonl)"
    )
    
    parser.add_argument(
        "--output",
        type=str,
        default=None,
        help="Output filename without extension (default: sales_transactions_YYYYMMDD_HHMMSS)"
    )
    
    parser.add_argument(
        "--seed",
        type=int,
        default=None,
        help="Random seed for reproducibility (optional)"
    )
    
    parser.add_argument(
        "--messiness",
        type=int,
        default=35,
        help="Percentage of records with data quality issues, 0-100 (default: 35)"
    )
    
    parser.add_argument(
        "--batch-size",
        type=int,
        default=10000,
        help="Number of records to generate per batch (default: 10,000)"
    )
    
    parser.add_argument(
        "--config",
        type=str,
        default=None,
        help="Path to JSON configuration file for per-column messiness control (optional)"
    )
    
    return parser.parse_args()


def main():
    """Main execution function."""
    args = parse_arguments()
    
    # Validate messiness level
    if not 0 <= args.messiness <= 100:
        print("Error: --messiness must be between 0 and 100")
        return 1
    
    # Determine output filename and add extension based on format
    extensions = {"jsonl": ".jsonl", "csv": ".csv", "parquet": ".parquet"}
    
    if args.output is None:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        base_filename = f"sales_transactions_{timestamp}"
    else:
        base_filename = args.output
    
    # Add extension based on format
    args.output = f"{base_filename}{extensions[args.format]}"
    
    # Load column configuration if provided
    column_config = None
    if args.config:
        column_config = load_column_config(args.config)
        if not column_config:
            print("Warning: No valid column configuration loaded. Using default messiness level.")
    
    print("\n" + "="*60)
    print("MESSY SALES TRANSACTION DATASET GENERATOR")
    print("="*60)
    print(f"\n⚙️  Configuration:")
    print(f"  Records to generate: {args.rows:,}")
    print(f"  Output format: {args.format}")
    print(f"  Output file: {args.output}")
    
    if column_config:
        print(f"  Configuration file: {args.config}")
        print(f"  Per-column messiness:")
        for col, config in column_config.items():
            messiness = config.get("messiness", args.messiness)
            print(f"    - {col}: {messiness}%")
    else:
        print(f"  Messiness level: {args.messiness}% (global)")
    
    print(f"  Batch size: {args.batch_size:,}")
    if args.seed is not None:
        print(f"  Random seed: {args.seed}")
    print(f"\n🔄 Generating data...")
    
    # Initialize generator
    generator = DataGenerator(seed=args.seed, messiness_level=args.messiness, column_config=column_config)
    
    # Start timing
    start_time = time.time()
    
    # Generate records in batches
    all_records = []
    for batch_start in range(0, args.rows, args.batch_size):
        batch_end = min(batch_start + args.batch_size, args.rows)
        
        # Generate batch
        batch = [generator.generate_record(i) for i in range(batch_start, batch_end)]
        all_records.extend(batch)
        
        # Progress update
        progress = (batch_end / args.rows) * 100
        print(f"  Progress: {batch_end:,}/{args.rows:,} ({progress:.1f}%)", end='\r')
    
    print(f"  Progress: {args.rows:,}/{args.rows:,} (100.0%)")
    
    # Add duplicates
    print(f"\n🔄 Adding duplicate records...")
    num_before_duplicates = len(all_records)
    all_records = add_duplicates(all_records, duplicate_rate=0.05)
    generator.stats["duplicates"] = len(all_records) - num_before_duplicates
    
    # Shuffle to mix duplicates throughout
    random.shuffle(all_records)
    
    # Write output
    print(f"💾 Writing to {args.format.upper()} file...")
    
    if args.format == "jsonl":
        OutputWriter.write_jsonl(all_records, args.output)
    elif args.format == "csv":
        OutputWriter.write_csv(all_records, args.output)
    elif args.format == "parquet":
        OutputWriter.write_parquet(all_records, args.output)
    
    # Calculate generation time
    generation_time = time.time() - start_time
    
    # Print statistics
    print_statistics(generator.stats, generation_time, args.output)
    
    return 0


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