# -*- coding: utf-8 -*-

# les mêmes librairies
import os
import time
import pandas
from email import policy
from email.parser import BytesParser
#import nltk
from nltk.sentiment.vader import SentimentIntensityAnalyzer
import argparse
from itertools import chain

# avec ray en plus
import ray

# ----------- Configuration -----------

EMAIL_FOLDER = "emails"  # sous-dossier contenant les fichiers d'e-mails

# pour éviter le warning
os.environ["RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO"] = "0"

# nécessaire une seule fois dans l'environnement
#nltk.download('vader_lexicon')


# ----------- Classe de calcul des sentiments --------
# -- on aura autant de worker que de processus demandé
# cf. la notion d'Actors de Ray (Ray Core / Key Concepts)

@ray.remote
class SentimentWorker:
    # constructeur - initialisation
    def __init__(self):
        #objet pour calculer le sentiment
        self.sentiment_analyzer = SentimentIntensityAnalyzer()
        
    #elements constitutifs du mail extrait d'un fichier
    def parse_email_file(self, file_path):
        """
        Parse un fichier email et extrait les champs demandés.
        """
        with open(file_path, "rb") as f:
            msg = BytesParser(policy=policy.default).parse(f)
    
        message_id = msg.get("Message-ID", "")
        date = msg.get("Date", "")
        sender = msg.get("From", "")
        body = self.extract_email_body(msg)
    
        return {
            "message_id": message_id,
            "date": date,
            "sender": sender,
            "body": body
        }
        
    #extraction du corps du mail
    def extract_email_body(self, msg):
        """
        Extrait le corps texte d'un email.
        """
        return msg.get_payload(decode=True).decode(errors="ignore")

    #analyser un message - score de sentiment
    def analyze_sentiment(self, email_data):
        # récuperation du corps du message
        text = email_data["body"]

        # si vide
        if not text.strip():
            #score de positivité nul
            result = 0
        else:    
            # appel de l'objet de calcul des sentimens
            result = self.sentiment_analyzer.polarity_scores(text)["pos"]

        # integration des resultats dans le mail
        email_data["score"] = result
        
        return email_data
    
    #analyser une liste représentant des noms de fichiers
    def analyze_sentiment_liste(self,lst_fichiers):
    
        # liste des e-mails à constituer
        lst_mails = []
    
        # pour chaque nom de fichier de la liste
        for nom_fichier in lst_fichiers:
            #récupérer les parties du mail
            mail = self.parse_email_file(nom_fichier)
            # analyse des sentiments
            self.analyze_sentiment(mail)
            #ajouter dans la sortie
            lst_mails.append(mail)
            
        #renvoie le tout
        return lst_mails


# ----------- Traitement principal -----------
# paramétré par le nombre de "workers" invoqués
# l'appréhension' des fichiers dans le dossier n'est pas parralélisé
# ce sont les calculs des sentiments qui le sont
# ----------------------------------------------

def process_email_folder_parallel(folder_path, num_workers):

    # Création des workers
    # c.-à-d. autant d'objet de calcul de sentiment que de workers
    workers = [SentimentWorker.remote() for _ in range(num_workers)]

    # liste des listes de mails - répartition en blocs
    lst_lst_mails = [[] for _ in range(num_workers)]
    
    # pour dispatcher aux workers les appels
    worker_index = 0

    #pour chaque élément dans le dossier à traiter
    for filename in os.listdir(folder_path):
        
        # nom complet : chemin + nom du fichier
        file_path = os.path.join(folder_path, filename)

        # On vérifie juste que c’est un fichier (pas un dossier)
        if os.path.isfile(file_path):
        
            try:
                file_path = os.path.join(folder_path, filename)

                # ajouter dans la liste des fichier à traiter pour un worker
                lst_lst_mails[worker_index].append(file_path)
                
                # numéro du worker suivant à solliciter
                # si worker_index > num_workers, retour à 0
                worker_index = (worker_index + 1) % num_workers
                
            except Exception as e:
                print(f"Erreur avec {filename}: {e}")                

    #tailles des blocs
    #[print(len(lst_lst_mails[id])) for id in range(num_workers)]

    # préparer le liste des calculs
    futures = [workers[id].analyze_sentiment_liste.remote(lst_lst_mails[id]) for id in range(num_workers)]
                                                          
    # lance effectivement les calculs
    # et récupère les résultats
    results = ray.get(futures)
    
    # renvoyer le tout
    return results



# programme principal
if __name__ == "__main__":
    
    # paramètre en ligne de commande
    parser = argparse.ArgumentParser()
    parser.add_argument("--n", type=int,default=4)
    
    # récupération des paramètres
    args = parser.parse_args()

    #nombre de cores
    num_cores = args.n
    print(f"Nombre de coeurs a solliciter = {num_cores}")

    # Initialisation de Ray
    ray.init(ignore_reinit_error=True,num_cpus=num_cores)
    

    print("Debut du traitement...")
       
    #debut chrono
    debut = time.time()  
    
    # lacement du processus avec "num_cores" workers
    # on obtient une liste de listes de dictionnaires
    # ou chaque dictionnaire représente un e-mail
    all_results = process_email_folder_parallel(EMAIL_FOLDER,num_workers=num_cores)
    
   
    # fin chrono
    duree = time.time() - debut

    print(f"\nFin du traitement, duree = {duree} sec.\n")
    
    # Fin de ray
    ray.shutdown()
    
    #nombre de listes dans la liste globale
    print(f"Nombre de blocs de resultats = {len(all_results)}")
            
    # mettre sous la forme d'un data frame
    # la liste de listes de dictionnaires
    df = pandas.DataFrame(list(chain.from_iterable(all_results)))
    #affichage des infos
    #print(df.info())
      
    #trier selon l'id
    df = df.sort_values("message_id")
    # et affichage
    print(df[['date','score']].head(10))
    
    
