reducer.py

#!/usr/bin/env python3
"""
Word Count Reducer Module
Improved version with better error handling and performance
"""

import sys
import logging
from typing import List, Tuple
from utils import safe_parse_line, handle_broken_pipe, setup_logging

# Setup logging
setup_logging()
logger = logging.getLogger(__name__)

def print_word_count(word: str, count: int) -> None:
    """Print word count in tab-separated format"""
    print(f'{word}\t{count}')

def reduce_word_counts() -> List[Tuple[str, int]]:
    """
    Reduce word counts from mapper output
    
    Returns:
        List of (word, count) tuples sorted by count descending
    """
    current_word = None
    current_count = 0
    word_counts = []

    try:
        # Read input from stdin
        for line in sys.stdin:
            parsed = safe_parse_line(line, expected_parts=2)
            if not parsed:
                continue
                
            word, count_str = parsed
            
            try:
                count = int(count_str)
            except ValueError:
                logger.warning(f"Invalid count value: {count_str}")
                continue
            
            # Sum up counts for the same word
            if current_word == word:
                current_count += count
            else:
                # Store previous word count if exists
                if current_word:
                    word_counts.append((current_word, current_count))
                
                # Initialize new word
                current_word = word
                current_count = count
        
        # Handle the last word
        if current_word:
            word_counts.append((current_word, current_count))
        
        # Sort by count in descending order
        word_counts.sort(key=lambda x: x[1], reverse=True)
        
        return word_counts
        
    except Exception as e:
        logger.error(f"Error in reducer: {e}")
        return []

def main():
    """Main reducer function"""
    try:
        # Process word counts
        word_counts = reduce_word_counts()
        
        if not word_counts:
            logger.warning("No word counts to process")
            return
        
        # Output results
        for word, count in word_counts:
            print_word_count(word, count)
            
        logger.info(f"Processed {len(word_counts)} unique words")
        
    except KeyboardInterrupt:
        logger.info("Reducer interrupted by user")
        sys.exit(0)
    except Exception as e:
        logger.error(f"Error in reducer: {e}")
        sys.exit(1)

if __name__ == "__main__":
    try:
        main()
    except BrokenPipeError:
        handle_broken_pipe()