application_neo4j/app_algorithms.py

# Importation des bibliothèques nécessaires
from graphdatascience import GraphDataScience
from typing import Dict, List, Any
import pandas as pd
from translations import t

def run_gds_bfs(
    gds: GraphDataScience,
    natural_graph_name: str,
    reverse_graph_name: str,
    source_name: str,
    max_depth: int = None,
    expert=False,
    job_id: str = None,
    progress_callback=None,
) -> Dict[str, Any]:
    """
    Exécute un parcours en largeur (BFS) directionnel à l'aide de GDS pour trouver les descendants et les ascendants.

    Cette fonction nécessite deux graphes pré-projetés en mémoire GDS :
    - Un graphe "naturel" pour trouver les descendants (relations dans le sens source -> cible).
    - Un graphe "inversé" pour trouver les ascendants (relations dans le sens cible -> source).

    Args:
        gds: L'objet de connexion à la bibliothèque Graph Data Science.
        natural_graph_name: Le nom du graphe GDS projeté avec une orientation NATURELLE.
        reverse_graph_name: Le nom du graphe GDS projeté avec une orientation INVERSÉE.
        source_name: La propriété 'name' du nœud de départ de la recherche.
        max_depth: La profondeur maximale de recherche. Si None, la recherche est illimitée.

    Returns:
        Un dictionnaire contenant l'ID du nœud source et deux DataFrames pandas :
        un pour les chemins des descendants et un pour les chemins des ascendants.
    """
    # GDS fonctionne avec des identifiants de nœuds internes (des nombres), pas avec des noms.
    # La première étape est donc de trouver l'ID numérique de notre nœud de départ à partir de son nom.
    try:
        source_id_result = gds.run_cypher(
            """
            MATCH (n {name: $source_name}) 
            RETURN id(n) AS id , labels(n) as label      
            LIMIT 1                       
            """,
            {"source_name": source_name}
        )
        
        if source_id_result.empty or (source_id_result["label"][0]==["Author"] and not expert):
            print(f"Le modèle ou dataset avec le nom '{source_name}' n'a pas été trouvé.")
            return None # Retourne des DataFrames vides
        
        # On récupère l'ID de la première ligne du résultat.
        source_node_id = source_id_result['id'][0]
    except Exception as e:
        print(f"Erreur lors de la recherche de l'ID du nœud source pour '{source_name}': {e}")
        return {"source_label": source_id_result["label"][0][0],"descendant": pd.DataFrame(), "ascendant": pd.DataFrame()}

    # Préparation des paramètres pour l'algorithme BFS.
    bfs_params = {'sourceNode': source_node_id}
    print(bfs_params)
    # Si une profondeur maximale est spécifiée, on l'ajoute aux paramètres.
    if max_depth is not None:
        bfs_params['maxDepth'] = max_depth

    # --- Exécution du BFS pour trouver les DESCENDANTS sur le graphe NATUREL ---
    # On récupère l'objet graphe depuis GDS.
    g_natural = gds.graph.get(natural_graph_name)
    # On exécute l'algorithme BFS en mode `stream` pour obtenir les chemins.
    desc_job_id = f"{job_id}-descendants" if job_id else None
    if desc_job_id:
        bfs_params["jobId"] = desc_job_id
    if progress_callback:
        progress_callback("searching_descendants", desc_job_id)
    desc_df = gds.bfs.stream(g_natural, **bfs_params)
    print("BFS pour les descendants sur le graphe naturel terminé.")

    # --- Exécution du BFS pour trouver les ASCENDANTS sur le graphe INVERSÉ ---
    # Utiliser un graphe inversé est très efficace pour trouver les parents/ancêtres.
    g_reverse = gds.graph.get(reverse_graph_name)
    asc_job_id = f"{job_id}-ancestors" if job_id else None
    if asc_job_id:
        bfs_params["jobId"] = asc_job_id
    elif "jobId" in bfs_params:
        del bfs_params["jobId"]
    if progress_callback:
        progress_callback("searching_ancestors", asc_job_id)
    asc_df = gds.bfs.stream(g_reverse, **bfs_params)
    print("BFS pour les ascendants sur le graphe inversé terminé.")
    print("DESC",desc_df)
    print("ASC",asc_df)
    print(source_id_result["label"][0][0])

    # Retourne les résultats sous forme d'un dictionnaire structuré.
    return {
        "source_node": source_node_id,"source_label": source_id_result["label"][0][0],
        "descendant": desc_df,
        "ascendant": asc_df
    }





def get_genealogy_highlights(gds: "GraphDataScience", model_name: str, num_highlights: int = 2, lang: str = "fr") -> Dict:
    """
    Trouve les modèles clés dans l'ascendance et la descendance (1er/2e plus cités/téléchargés).
    
    Args:
        gds: L'instance de GraphDataScience.
        model_name: Le nom du modèle de départ.
        num_highlights: Le nombre de modèles à récupérer pour chaque catégorie (par défaut 2).

    Returns:
        Un dictionnaire contenant les listes de modèles unifiés pour l'affichage.
    """
    highlights = {
        "desc_unique_models": [],
        "asc_unique_models": []
    }

    # --- DÉFINITION CENTRALE DES BADGES ---
    # Centraliser les badges ici rend le code beaucoup plus facile à modifier.
    # Les classes CSS sont directement des classes Bootstrap 5 pour simplifier le rendu dans le template HTML.
    badges_info = {
    'desc_cited_1': {
        'text': t('badge.desc_cited_1.text', lang), 
        'class': 'bg-success',
        'title': t('badge.desc_cited_1.title', lang)
    },
    'desc_cited_2': {
        'text': t('badge.desc_cited_2.text', lang), 
        'class': 'bg-success bg-opacity-75',
        'title': t('badge.desc_cited_2.title', lang)
    },
    'desc_downloaded_1': {
        'text': t('badge.desc_downloaded_1.text', lang), 
        'class': 'beta',
        'title': t('badge.desc_downloaded_1.title', lang)
    },
    'desc_downloaded_2': {
        'text': t('badge.desc_downloaded_2.text', lang), 
        'class': 'alpha',
        'title': t('badge.desc_downloaded_2.title', lang)
    },
    
    'asc_foundation': {
        'text': t('badge.asc_foundation.text', lang), 
        'class': 'bg-warning text-dark',
        'title': t('badge.asc_foundation.title', lang)
    },
    'asc_cited_1': {
        'text': t('badge.asc_cited_1.text', lang), 
        'class': 'bg-success',
        'title': t('badge.asc_cited_1.title', lang)
    },
    'asc_cited_2': {
        'text': t('badge.asc_cited_2.text', lang), 
        'class': 'bg-success bg-opacity-75',
        'title': t('badge.asc_cited_2.title', lang)
    },
    'asc_downloaded_1': {
        'text': t('badge.asc_downloaded_1.text', lang), 
        'class': 'beta',
        'title': t('badge.asc_downloaded_1.title', lang)
    },
    'asc_downloaded_2': {
        'text': t('badge.asc_downloaded_2.text', lang), 
        'class': 'alpha',
        'title': t('badge.asc_downloaded_2.title', lang)
    },
    }
    

    def process_and_assign_badges(
        unified_dict: Dict, 
        model_list: List[Dict], 
        badge_keys: List[str]
    ):
        """
        Fonction utilitaire pour ajouter des modèles et leurs badges à un dictionnaire unifié.
        Cela évite la duplication de code pour chaque catégorie (cité, téléchargé, etc.).
        """
        for i, model in enumerate(model_list):
            if i < len(badge_keys): # S'assurer qu'on a un badge défini pour ce rang
                model_name_key = model['name']
                badge_key = badge_keys[i]
                
                # Ajouter le modèle au dictionnaire s'il n'y est pas déjà
                if model_name_key not in unified_dict:
                    unified_dict[model_name_key] = model.copy()
                    unified_dict[model_name_key]['badges'] = []
                
                # Ajouter le badge correspondant
                badge_to_add = badges_info[badge_key]
                if badge_to_add not in unified_dict[model_name_key]['badges']:
                    unified_dict[model_name_key]['badges'].append(badge_to_add)

    # ==========================================================================
    # 1. GESTION DE LA DESCENDANCE
    # ==========================================================================
    desc_downloads_query = """
        MATCH (start:Model {name: $model_name})-[:USED_IN*1..]->(descendant:Model)
        WHERE start <> descendant
        WITH descendant, size([(m:Model)<-[:USED_IN]-(descendant) | m]) AS citation_count
        RETURN descendant.name AS name, citation_count, descendant.downloads AS downloads, descendant.task AS task, descendant.license AS license, descendant.likes AS likes, descendant.createdAt AS createdAt
        ORDER BY descendant.downloads DESC, descendant.name ASC
        LIMIT $limit
    """
    desc_cited_query = """
        MATCH (start:Model {name: $model_name})-[:USED_IN*1..]->(descendant:Model)
        WHERE start <> descendant
        WITH descendant, size([(m:Model)<-[:USED_IN]-(descendant) | m]) AS citation_count
        RETURN descendant.name AS name, citation_count, descendant.task AS task, descendant.downloads AS downloads, descendant.license AS license, descendant.likes AS likes, descendant.createdAt AS createdAt
        ORDER BY citation_count DESC, descendant.name ASC
        LIMIT $limit
    """
    try:
        params = {"model_name": model_name, "limit": num_highlights}
        desc_downloaded_list = gds.run_cypher(desc_downloads_query, params).to_dict('records')
        desc_cited_list = gds.run_cypher(desc_cited_query, params).to_dict('records')

        desc_unified_models = {}
        process_and_assign_badges(desc_unified_models, desc_cited_list, ['desc_cited_1', 'desc_cited_2'])
        process_and_assign_badges(desc_unified_models, desc_downloaded_list, ['desc_downloaded_1', 'desc_downloaded_2'])
        
        highlights["desc_unique_models"] = list(desc_unified_models.values())

    except Exception as e:
        print(f"Erreur lors de la recherche des descendants: {e}")

    # ==========================================================================
    # 2. GESTION DE L'ASCENDANCE
    # ==========================================================================
    asc_downloads_query = """
        MATCH (ascendant:Model)-[:USED_IN*1..]->(start:Model {name: $model_name})
        WHERE start <> ascendant
        WITH ascendant, size([(m:Model)<-[:USED_IN]-(ascendant) | m]) AS citation_count
        RETURN ascendant.name AS name, citation_count, ascendant.downloads AS downloads, ascendant.task AS task, 
        ascendant.license AS license, ascendant.likes AS likes, ascendant.createdAt AS createdAt
        ORDER BY ascendant.downloads DESC
        LIMIT 1 // On ne veut que LE plus téléchargé
    """
    asc_cited_query = """
        MATCH (ascendant:Model)-[:USED_IN*1..]->(start:Model {name: $model_name})
        WHERE start <> ascendant
        WITH ascendant, size([(m:Model)<-[:USED_IN]-(ascendant) | m]) AS citation_count
        RETURN ascendant.name AS name, citation_count, ascendant.downloads AS downloads, ascendant.task AS task, 
        ascendant.license AS license, ascendant.likes AS likes, ascendant.createdAt AS createdAt
        ORDER BY citation_count DESC
        LIMIT 1 // On ne veut que LE plus cité
    """

    foundation_query = """
        MATCH (foundation:Model)-[:USED_IN*1..]->(start:Model {name: $model_name})
        WHERE NOT EXISTS( (:Model)-[:USED_IN]->(foundation) )
        WITH foundation, size([(m:Model)<-[:USED_IN]-(foundation) | m]) AS citation_count
        RETURN DISTINCT foundation.name AS name, citation_count, foundation.downloads AS downloads, foundation.task AS task, 
        foundation.license AS license, foundation.likes AS likes, foundation.createdAt AS createdAt
        LIMIT $limit
    """

    try:
        params = {"model_name": model_name, "limit": num_highlights}
        asc_foundation_list = gds.run_cypher(foundation_query, params).to_dict('records')
        asc_downloaded_list = gds.run_cypher(asc_downloads_query, params).to_dict('records')
        asc_cited_list = gds.run_cypher(asc_cited_query, params).to_dict('records')

        asc_unified_models = {}
        # Ordre de priorité : Racine > Cité > Téléchargé
        process_and_assign_badges(asc_unified_models, asc_foundation_list, ['asc_foundation'] * num_highlights) # Le badge racine s'applique à tous
        process_and_assign_badges(asc_unified_models, asc_cited_list, ['asc_cited_1', 'asc_cited_2'])
        process_and_assign_badges(asc_unified_models, asc_downloaded_list, ['asc_downloaded_1', 'asc_downloaded_2'])

        highlights["asc_unique_models"] = list(asc_unified_models.values())

    except Exception as e:
        print(f"Erreur lors de la recherche des ascendants: {e}")

    return highlights


def create_node_data(node_props, label):
    """
    Construit un dictionnaire de données pour chaque noeud
    à afficher dans le graphe front-end.
    """
    base_data = {
        "id": node_props.get("name", "")
    }
    
    if label == "Author":
        return {
            **base_data,
            "label": node_props.get("type", "Unknown"),
            "followers": node_props.get("followers", 1)
        }
    elif label == "Model":
        licens_ =str(node_props.get("license", t("node.unknown", "fr"))).strip("[]")
        if licens_ =="\'other\'" or pd.isna(licens_) or licens_ =="nan":
            licens_ = t("node.other", "fr")
        
        tache = node_props.get("task", "")
        if tache =="unknown": 
            tache = t("node.unknown", "fr")
        return {
            **base_data,
            "label": "Modèle",
            "downloads": node_props.get("downloads", 1),
            "likes": node_props.get("likes", 0),
            "license": licens_,
            "createdAt": node_props.get("createdAt", "inconnue"),
            "createdAt_dataset": node_props.get("createdAt_dataset", "inconnue"),
            "task": tache,
            "author": node_props.get("author", ""),"dataset": node_props.get("dataset", ""),
                "ascendantsCount": node_props.get("ascendantsCount", 0),"descendantsCount": node_props.get("descendantsCount", 0),
                "citationCount": node_props.get("citationCount", 0), "distance":node_props.get("distance", 0)
        }
    else:  # Dataset or other
        return {
            **base_data,
            "label": "Dataset",
            "downloads": node_props.get("downloads", 1),
            "createdAt_dataset": node_props.get("createdAt_dataset", "inconnue")
        }

    return { "id": node_props['name'], "label": label, **node_props }