|
import requests |
|
import json |
|
import networkx as nx |
|
import matplotlib.pyplot as plt |
|
from fuzzywuzzy import fuzz |
|
from fuzzywuzzy import process |
|
from lib.memory import * |
|
|
|
|
|
class APIRequester: |
|
def __init__(self): |
|
pass |
|
|
|
def make_request(self, url): |
|
response = requests.get(url) |
|
if response.status_code == 200: |
|
return response.json() |
|
else: |
|
return None |
|
|
|
class Grapher: |
|
def __init__(self, memoria_nlp, threshold=70): |
|
self.threshold = threshold |
|
self.graph = nx.Graph() |
|
self.memoria_nlp = memoria_nlp |
|
|
|
def parse_json(self, data, parent=None): |
|
if isinstance(data, dict): |
|
for key, value in data.items(): |
|
if parent: |
|
self.graph.add_node(parent) |
|
self.graph.add_node(key) |
|
self.graph.add_edge(parent, key) |
|
|
|
for node in self.graph.nodes(): |
|
if node != parent and fuzz.ratio(node, key) >= self.threshold: |
|
self.graph = nx.contracted_nodes(self.graph, node, key, self_loops=False) |
|
|
|
self.memoria_nlp.agregar_concepto("keys", [(key, 1.0)]) |
|
|
|
if isinstance(value, (dict, list)): |
|
self.parse_json(value, key) |
|
else: |
|
self.memoria_nlp.agregar_concepto("values", [(str(value), 1.0)]) |
|
if parent: |
|
self.graph.add_node(value) |
|
self.graph.add_edge(key, value) |
|
|
|
for node in self.graph.nodes(): |
|
if node != value and fuzz.ratio(node, value) >= self.threshold: |
|
self.graph = nx.contracted_nodes(self.graph, node, value, self_loops=False) |
|
elif isinstance(data, list): |
|
for item in data: |
|
self.parse_json(item, parent) |
|
|
|
def draw_graph(self): |
|
pos = nx.spring_layout(self.graph, seed=42) |
|
nx.draw(self.graph, pos, with_labels=True, node_size=700, node_color='skyblue', font_size=10, font_weight='bold') |
|
plt.title("JSON Graph") |
|
plt.show() |
|
|
|
def guardar_en_memoria(self): |
|
keys = self.memoria_nlp.obtener_conceptos_acotados(100) |
|
with open("memoria.json", "w") as file: |
|
json.dump(keys, file) |
|
|
|
def buscar_nodo(self, nodo): |
|
return process.extractOne(nodo, self.graph.nodes())[0] |
|
|
|
def eliminar_nodo(self, nodo): |
|
self.graph.remove_node(nodo) |
|
|
|
def agregar_nodo(self, nodo): |
|
self.graph.add_node(nodo) |
|
|
|
def distancia_entre_nodos(self, nodo1, nodo2): |
|
return nx.shortest_path_length(self.graph, source=nodo1, target=nodo2) |
|
|
|
def ruta_entre_nodos(self, nodo1, nodo2): |
|
return nx.shortest_path(self.graph, source=nodo1, target=nodo2) |
|
|
|
def unir_grafos(self, otro_grafo, umbral): |
|
for nodo in otro_grafo.nodes(): |
|
nodo_similar = process.extractOne(nodo, self.graph.nodes())[0] |
|
if fuzz.ratio(nodo, nodo_similar) >= umbral: |
|
self.graph = nx.contracted_nodes(self.graph, nodo_similar, nodo, self_loops=False) |
|
else: |
|
self.graph.add_node(nodo) |
|
for vecino in otro_grafo.neighbors(nodo): |
|
self.graph.add_edge(nodo, vecino) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
|
|
memoria_nlp = MemoriaRobotNLP(max_size=100) |
|
json_parser = JSONParser(memoria_nlp) |
|
|
|
api_requester = APIRequester() |
|
url = "https://jsonplaceholder.typicode.com/posts" |
|
data = api_requester.make_request(url) |
|
|
|
if data: |
|
json_parser.parse_json(data) |
|
json_parser.draw_graph() |
|
|
|
otro_parser = JSONParser(MemoriaRobotNLP(max_size=100)) |
|
otro_parser.parse_json({"id": 101, "title": "New Title", "userId": 11}) |
|
|
|
print("Uniendo los grafos...") |
|
json_parser.unir_grafos(otro_parser.graph, umbral=80) |
|
print("Grafo unido:") |
|
json_parser.draw_graph() |
|
|
|
json_parser.guardar_en_memoria() |
|
else: |
|
print("Error al realizar la solicitud a la API.") |
|
|
|
|
