| |
|
|
| 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()) |
|
|
