def _build_networkx_graph()

in identity-resolution/notebooks/identity-graph/nepytune/usecase/user_summary.py [0:0]

• 45 lines of code
• 11 McCabe index (conditional complexity)

def _build_networkx_graph(g, identity_group_id):
    def get_attributes(attribute_list):
        attrs = {}
        for attr_name, value in attribute_list:
            attr_name = str(attr_name)

            if isinstance(value, Iterable) and not isinstance(value, str):
                for i, single_val in enumerate(value):
                    attrs[f"{attr_name}-{i}"] = single_val
            else:
                if '.' in attr_name:
                    attr_name = attr_name.split('.')[-1]
                attrs[attr_name] = value

        return attrs

    graph = nx.Graph()

    for ig_node in _get_subgraph(g, identity_group_id).toList():
        ig_id = ig_node["props"][T.id]

        graph.add_node(
            ig_id,
            **get_attributes(ig_node["props"].items())
        )

        for persistent_node in ig_node["persistent_ids"]:
            p_id = persistent_node["props"][T.id]
            graph.add_node(
                p_id,
                **get_attributes(persistent_node["props"].items())
            )
            graph.add_edge(ig_id, p_id, label="member")

            for transient_node in persistent_node["transient_ids"]:
                transient_node_map = transient_node["props"]
                transient_id = transient_node_map[T.id]
                graph.add_node(
                    transient_id,
                    **get_attributes(transient_node_map.items())
                )
                graph.add_edge(transient_id, p_id, label="has_identity")

                for ip_location_node in transient_node["ip_location"]:
                    ip_location_id = ip_location_node[T.id]
                    graph.add_node(ip_location_id, **get_attributes(ip_location_node.items()))
                    graph.add_edge(ip_location_id, transient_id, label="uses")

                for visited_link, root_url in transient_node["random_website_paths"]:
                    graph.add_node(visited_link["url"][0], label="website", **get_attributes(visited_link.items()))
                    graph.add_node(root_url["url"][0], label="websiteGroup", **get_attributes(root_url.items()))
                    graph.add_edge(transient_id, visited_link["url"][0], label="visits")
                    graph.add_edge(visited_link["url"][0], root_url["url"][0], label="links_to")
    return graph