reducer_tfidf.py

#!/usr/bin/env python3
"""
TF-IDF Reducer Module
Improved version with better error handling and output formatting
"""

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

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

def reduce_tfidf_scores() -> List[Tuple[str, float]]:
    """
    Reduce TF-IDF scores from mapper output
    
    Returns:
        List of (word, avg_tfidf) tuples sorted by score descending
    """
    current_word = None
    current_tfidf_sum = 0.0
    document_count = 0
    word_tfidfs = []

    try:
        # Read input from stdin
        for line in sys.stdin:
            parsed = safe_parse_line(line, expected_parts=3)
            if not parsed:
                continue
                
            word, doc_id, tfidf_str = parsed
            
            try:
                tfidf = float(tfidf_str)
            except ValueError:
                logger.warning(f"Invalid TF-IDF value: {tfidf_str}")
                continue

            # Sum up TF-IDF scores for the same word
            if current_word == word:
                current_tfidf_sum += tfidf
                document_count += 1
            else:
                # Store previous word's average TF-IDF if exists
                if current_word:
                    avg_tfidf = current_tfidf_sum / document_count
                    word_tfidfs.append((current_word, avg_tfidf))
                
                # Initialize new word
                current_word = word
                current_tfidf_sum = tfidf
                document_count = 1

        # Handle the last word
        if current_word:
            avg_tfidf = current_tfidf_sum / document_count
            word_tfidfs.append((current_word, avg_tfidf))

        # Sort by average TF-IDF score in descending order
        word_tfidfs.sort(key=lambda x: x[1], reverse=True)
        
        return word_tfidfs
        
    except Exception as e:
        logger.error(f"Error in TF-IDF reducer: {e}")
        return []

def print_header() -> None:
    """Print formatted header"""
    print(f"{'Word':<20} {'Avg_TF-IDF':<15}")
    print("-" * 35)

def main():
    """Main TF-IDF reducer function"""
    try:
        # Process TF-IDF scores
        word_tfidfs = reduce_tfidf_scores()
        
        if not word_tfidfs:
            logger.warning("No TF-IDF scores to process")
            return
        
        # Print header
        print_header()
        
        # Output results
        for word, avg_tfidf in word_tfidfs:
            print(format_output(word, avg_tfidf, width=20))
            
        logger.info(f"Processed {len(word_tfidfs)} unique words with TF-IDF scores")
        
    except KeyboardInterrupt:
        logger.info("TF-IDF reducer interrupted by user")
        sys.exit(0)
    except Exception as e:
        logger.error(f"Error in TF-IDF reducer: {e}")
        sys.exit(1)

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