from pyspark.sql.functions import udf, col
from pyspark.ml.feature import Tokenizer, StopWordsRemover, HashingTF, IDF
from pyspark.sql.types import DoubleType
import numpy as np
import re
import logging

class EnhancedMonteCarloTextCategorizerSpark:
    def __init__(self, categories_df, tfidf_num_features=500, n_simulations=1000, confidence_threshold=0.6):
        self.categories_df = categories_df  # Espera-se que já seja um DataFrame do PySpark
        self.tfidf_num_features = tfidf_num_features
        self.n_simulations = n_simulations
        self.confidence_threshold = confidence_threshold

    @staticmethod
    def preprocess_text(text):
        """Pré-processa o texto para minúsculas e remove caracteres não desejados."""
        return ' '.join(re.findall(r'\w+', text.lower()))

    def create_tfidf_features(self, texts_df, input_col, output_col):
        """Cria features TF-IDF usando tokenização e remoção de stopwords."""
        tokenizer = Tokenizer(inputCol=input_col, outputCol="words")
        words_data = tokenizer.transform(texts_df)
        
        remover = StopWordsRemover(inputCol="words", outputCol="filtered")
        filtered_words_data = remover.transform(words_data)
        
        hashing_tf = HashingTF(inputCol="filtered", outputCol="rawFeatures", numFeatures=self.tfidf_num_features)
        featurized_data = hashing_tf.transform(filtered_words_data)
        
        idf = IDF(inputCol="rawFeatures", outputCol=output_col)
        idf_model = idf.fit(featurized_data)
        rescaled_data = idf_model.transform(featurized_data)
        
        return rescaled_data

    def categorize_texts(self, text_df):
        """Categoriza cada texto no DataFrame."""
        prepared_categories = self.create_tfidf_features(self.categories_df, "palavras_chave", "tfidfFeaturesCategory")

        text_tfidf_results = self.create_tfidf_features(text_df, "tx_trsc_segm", "tfidfFeaturesText")
        
        category_joined_df = text_tfidf_results.crossJoin(prepared_categories)

        def monte_carlo_udf(cat_vector, text_vector):
            """Calcula a similaridade TF-IDF entre texto e categorias."""
            similarities = float(text_vector.dot(cat_vector))
            return similarities  # Retorna a similaridade como um valor float

        similarity_udf = udf(monte_carlo_udf, DoubleType())
        
        labeled_df = category_joined_df.withColumn(
            "similarity", 
            similarity_udf(col("tfidfFeaturesCategory"), col("tfidfFeaturesText"))
        )

        from pyspark.sql.window import Window
        from pyspark.sql.functions import row_number
        
        window_spec = Window.partitionBy("tx_trsc_segm").orderBy(col("similarity").desc())

        ranked_df = labeled_df.withColumn("rank", row_number().over(window_spec))
        
        result_df = ranked_df.filter(col("rank") == 1).select("tx_trsc_segm", "categoria")

        return result_df

# Supõe-se que categories_df e sampled_df sejam DataFrames do PySpark em um ambiente que fornece uma SparkSession

categorizer = EnhancedMonteCarloTextCategorizerSpark(categories_df)
final_df = categorizer.categorize_texts(sampled_df)

logging.info("Resultados finais:")
final_df.show() use cach para tudo