# 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]: """ Parcourt les descendants et ascendants avec le fork OpenGDS. Le plugin personnalisé ajoute au résultat standard ``nodeIds`` une liste ``depths`` alignée. La profondeur minimale de chaque nœud est ainsi calculée pendant le BFS, sans second parcours Cypher. Returns: L'identifiant et le label de la source, ainsi que les résultats GDS des parcours descendant et ascendant. """ 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: print(f"Le modèle ou dataset avec le nom '{source_name}' n'a pas été trouvé.") return None source_label = source_id_result["label"].iloc[0][0] if source_label == "Author" and not expert: print(f"Le modèle ou dataset avec le nom '{source_name}' n'a pas été trouvé.") return None source_node_id = int(source_id_result["id"].iloc[0]) except Exception as e: print(f"Erreur lors de la recherche de l'ID du nœud source pour '{source_name}': {e}") return None bfs_params = {"sourceNode": source_node_id} if max_depth is not None: bfs_params["maxDepth"] = max_depth g_natural = gds.graph.get(natural_graph_name) 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) _validate_depths_result(desc_df) print("BFS descendants terminé avec les profondeurs.") 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) _validate_depths_result(asc_df) print("BFS ascendants terminé avec les profondeurs.") return { "source_node": source_node_id, "source_label": source_label, "descendant": desc_df, "ascendant": asc_df, } def _validate_depths_result(result: pd.DataFrame) -> None: """Fail explicitly if Neo4j did not load the custom OpenGDS plugin.""" if result.empty: return if "depths" not in result.columns: raise RuntimeError( "Le plugin OpenGDS personnalisé n'est pas chargé : " "gds.bfs.stream ne renvoie pas la colonne depths." ) node_ids = result["nodeIds"].iloc[0] depths = result["depths"].iloc[0] if len(node_ids) != len(depths): raise RuntimeError( "Résultat BFS invalide : nodeIds et depths n'ont pas la même taille." ) 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 }