File size: 4,246 Bytes
376ee15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | # tools/concept_store.py
import uuid
from typing import Dict, List, Optional
from models import GraphExport
class Concept:
def __init__(
self,
id: Optional[str] = None,
label: str = "",
description: str = "",
tags: Optional[List[str]] = None
):
self.id = id or str(uuid.uuid4())
self.label = label
self.description = description
self.tags = tags or []
def to_dict(self):
return {
"id": self.id,
"label": self.label,
"description": self.description,
"tags": self.tags
}
@staticmethod
def from_dict(data):
return Concept(
id=data.get("id"),
label=data.get("label", ""),
description=data.get("description", ""),
tags=data.get("tags", [])
)
class Edge:
def __init__(self, source: str, target: str, relation: str = "related_to"):
self.source = source
self.target = target
self.relation = relation
def to_dict(self):
return {
"source": self.source,
"target": self.target,
"relation": self.relation
}
@staticmethod
def from_dict(data):
return Edge(
source=data["source"],
target=data["target"],
relation=data.get("relation", "related_to")
)
class ConceptStore:
def __init__(self):
self.concepts: Dict[str, Concept] = {}
self.edges: List[Edge] = []
def add(self, concept: Concept):
self.concepts[concept.id] = concept
def get(self, concept_id: str) -> Optional[dict]:
concept = self.concepts.get(concept_id)
return concept.to_dict() if concept else None
def add_edge(self, edge: Edge):
if not any(
e.source == edge.source and
e.target == edge.target and
e.relation == edge.relation
for e in self.edges
):
self.edges.append(edge)
def find_by_label(self, label: str) -> Optional[Concept]:
for c in self.concepts.values():
if c.label == label:
return c
return None
def import_from_json(self, data: dict) -> dict:
nodes = data.get("nodes", [])
edges = data.get("edges", [])
if not isinstance(nodes, list) or not isinstance(edges, list):
raise ValueError("Invalid format: 'nodes' and 'edges' must be lists")
imported_nodes = 0
imported_edges = 0
for c in nodes:
try:
concept = Concept.from_dict(c)
self.add(concept)
imported_nodes += 1
except Exception as e:
print(f"Failed to import concept: {e}")
for e in edges:
try:
edge = Edge.from_dict(e)
self.add_edge(edge)
imported_edges += 1
except Exception as e:
print(f"Failed to import edge: {e}")
return {
"status": "ok",
"imported_nodes": imported_nodes,
"imported_edges": imported_edges
}
def export_as_json(self) -> GraphExport:
return GraphExport(
nodes=[c.to_dict() for c in self.concepts.values()],
edges=[e.to_dict() for e in self.edges]
)
def all_concepts(self):
return [c.to_dict() for c in self.concepts.values()]
def all_edges(self):
return [e.to_dict() for e in self.edges]
def remove_concept(self, concept_id: str):
if concept_id in self.concepts:
del self.concepts[concept_id]
self.edges = [e for e in self.edges if e.source != concept_id and e.target != concept_id]
def remove_edge(self, source_id: str, target_id: str, relation: str):
self.edges = [
e for e in self.edges
if not (e.source == source_id and e.target == target_id and e.relation == relation)
]
def debug_print(self):
print("Concepts:")
for c in self.concepts.values():
print(c.to_dict())
print("\nEdges:")
for e in self.edges:
print(e.to_dict())
|