Navegador

navegador / navegador / ingestion / knowledge.py
Source Blame History 270 lines
4be3595… lmata 1 """
4be3595… lmata 2 KnowledgeIngester — manual curation of business concepts, rules, decisions,
4be3595… lmata 3 people, and domains into the navegador graph.
4be3595… lmata 4
4be3595… lmata 5 These are the things that don't live in code but belong in the knowledge graph:
4be3595… lmata 6 business rules, architectural decisions, domain groupings, ownership, wiki refs.
4be3595… lmata 7 """
4be3595… lmata 8
4be3595… lmata 9 import logging
4be3595… lmata 10 from typing import Any
4be3595… lmata 11
4be3595… lmata 12 from navegador.graph.schema import EdgeType, NodeLabel
4be3595… lmata 13 from navegador.graph.store import GraphStore
4be3595… lmata 14
4be3595… lmata 15 logger = logging.getLogger(__name__)
4be3595… lmata 16
4be3595… lmata 17
4be3595… lmata 18 class KnowledgeIngester:
4be3595… lmata 19 """
4be3595… lmata 20 Writes business knowledge nodes and relationships into the graph.
4be3595… lmata 21
4be3595… lmata 22 Usage:
4be3595… lmata 23 store = GraphStore.sqlite(".navegador/graph.db")
4be3595… lmata 24 k = KnowledgeIngester(store)
4be3595… lmata 25
4be3595… lmata 26 k.add_domain("auth", description="Authentication and authorisation")
4be3595… lmata 27 k.add_concept("JWT", description="Stateless token auth", domain="auth")
4be3595… lmata 28 k.add_rule("tokens must expire", domain="auth", severity="critical",
4be3595… lmata 29 rationale="Security requirement per OWASP")
4be3595… lmata 30 k.annotate_code("validate_token", "Function", concept="JWT")
4be3595… lmata 31 k.wiki_page("JWT Auth", url="...", content="...")
4be3595… lmata 32 """
4be3595… lmata 33
4be3595… lmata 34 def __init__(self, store: GraphStore) -> None:
4be3595… lmata 35 self.store = store
4be3595… lmata 36
4be3595… lmata 37 # ── Domains ───────────────────────────────────────────────────────────────
4be3595… lmata 38
4be3595… lmata 39 def add_domain(self, name: str, description: str = "") -> None:
758c6b7… lmata 40 self.store.create_node(
758c6b7… lmata 41 NodeLabel.Domain,
758c6b7… lmata 42 {
758c6b7… lmata 43 "name": name,
758c6b7… lmata 44 "description": description,
758c6b7… lmata 45 },
758c6b7… lmata 46 )
4be3595… lmata 47 logger.info("Domain: %s", name)
4be3595… lmata 48
4be3595… lmata 49 # ── Concepts ──────────────────────────────────────────────────────────────
4be3595… lmata 50
4be3595… lmata 51 def add_concept(
4be3595… lmata 52 self,
4be3595… lmata 53 name: str,
4be3595… lmata 54 description: str = "",
4be3595… lmata 55 domain: str = "",
4be3595… lmata 56 status: str = "",
4be3595… lmata 57 rules: str = "",
4be3595… lmata 58 examples: str = "",
4be3595… lmata 59 wiki_refs: str = "",
4be3595… lmata 60 ) -> None:
758c6b7… lmata 61 self.store.create_node(
758c6b7… lmata 62 NodeLabel.Concept,
758c6b7… lmata 63 {
758c6b7… lmata 64 "name": name,
758c6b7… lmata 65 "description": description,
758c6b7… lmata 66 "domain": domain,
758c6b7… lmata 67 "status": status,
758c6b7… lmata 68 "rules": rules,
758c6b7… lmata 69 "examples": examples,
758c6b7… lmata 70 "wiki_refs": wiki_refs,
758c6b7… lmata 71 },
758c6b7… lmata 72 )
4be3595… lmata 73 if domain:
4be3595… lmata 74 self._link_to_domain(name, NodeLabel.Concept, domain)
4be3595… lmata 75 logger.info("Concept: %s", name)
4be3595… lmata 76
4be3595… lmata 77 def relate_concepts(self, a: str, b: str) -> None:
4be3595… lmata 78 """Mark two concepts as related (bidirectional intent)."""
4be3595… lmata 79 self.store.create_edge(
758c6b7… lmata 80 NodeLabel.Concept,
758c6b7… lmata 81 {"name": a},
4be3595… lmata 82 EdgeType.RELATED_TO,
758c6b7… lmata 83 NodeLabel.Concept,
758c6b7… lmata 84 {"name": b},
4be3595… lmata 85 )
4be3595… lmata 86
4be3595… lmata 87 # ── Rules ─────────────────────────────────────────────────────────────────
4be3595… lmata 88
4be3595… lmata 89 def add_rule(
4be3595… lmata 90 self,
4be3595… lmata 91 name: str,
4be3595… lmata 92 description: str = "",
4be3595… lmata 93 domain: str = "",
4be3595… lmata 94 severity: str = "info",
4be3595… lmata 95 rationale: str = "",
4be3595… lmata 96 examples: str = "",
4be3595… lmata 97 ) -> None:
758c6b7… lmata 98 self.store.create_node(
758c6b7… lmata 99 NodeLabel.Rule,
758c6b7… lmata 100 {
758c6b7… lmata 101 "name": name,
758c6b7… lmata 102 "description": description,
758c6b7… lmata 103 "domain": domain,
758c6b7… lmata 104 "severity": severity,
758c6b7… lmata 105 "rationale": rationale,
758c6b7… lmata 106 "examples": examples,
758c6b7… lmata 107 },
758c6b7… lmata 108 )
4be3595… lmata 109 if domain:
4be3595… lmata 110 self._link_to_domain(name, NodeLabel.Rule, domain)
4be3595… lmata 111 logger.info("Rule: %s", name)
4be3595… lmata 112
4be3595… lmata 113 def rule_governs(self, rule_name: str, target_name: str, target_label: NodeLabel) -> None:
4be3595… lmata 114 self.store.create_edge(
758c6b7… lmata 115 NodeLabel.Rule,
758c6b7… lmata 116 {"name": rule_name},
4be3595… lmata 117 EdgeType.GOVERNS,
758c6b7… lmata 118 target_label,
758c6b7… lmata 119 {"name": target_name},
4be3595… lmata 120 )
4be3595… lmata 121
4be3595… lmata 122 # ── Decisions ─────────────────────────────────────────────────────────────
4be3595… lmata 123
4be3595… lmata 124 def add_decision(
4be3595… lmata 125 self,
4be3595… lmata 126 name: str,
4be3595… lmata 127 description: str = "",
4be3595… lmata 128 domain: str = "",
4be3595… lmata 129 status: str = "accepted",
4be3595… lmata 130 rationale: str = "",
4be3595… lmata 131 alternatives: str = "",
4be3595… lmata 132 date: str = "",
4be3595… lmata 133 ) -> None:
758c6b7… lmata 134 self.store.create_node(
758c6b7… lmata 135 NodeLabel.Decision,
758c6b7… lmata 136 {
758c6b7… lmata 137 "name": name,
758c6b7… lmata 138 "description": description,
758c6b7… lmata 139 "domain": domain,
758c6b7… lmata 140 "status": status,
758c6b7… lmata 141 "rationale": rationale,
758c6b7… lmata 142 "alternatives": alternatives,
758c6b7… lmata 143 "date": date,
758c6b7… lmata 144 },
758c6b7… lmata 145 )
4be3595… lmata 146 if domain:
4be3595… lmata 147 self._link_to_domain(name, NodeLabel.Decision, domain)
4be3595… lmata 148 logger.info("Decision: %s", name)
4be3595… lmata 149
4be3595… lmata 150 # ── People ────────────────────────────────────────────────────────────────
4be3595… lmata 151
4be3595… lmata 152 def add_person(
4be3595… lmata 153 self,
4be3595… lmata 154 name: str,
4be3595… lmata 155 email: str = "",
4be3595… lmata 156 role: str = "",
4be3595… lmata 157 team: str = "",
4be3595… lmata 158 ) -> None:
758c6b7… lmata 159 self.store.create_node(
758c6b7… lmata 160 NodeLabel.Person,
758c6b7… lmata 161 {
758c6b7… lmata 162 "name": name,
758c6b7… lmata 163 "email": email,
758c6b7… lmata 164 "role": role,
758c6b7… lmata 165 "team": team,
758c6b7… lmata 166 },
758c6b7… lmata 167 )
4be3595… lmata 168 logger.info("Person: %s", name)
4be3595… lmata 169
4be3595… lmata 170 def assign(self, target_name: str, target_label: NodeLabel, person_name: str) -> None:
4be3595… lmata 171 """Assign a person as owner of any node."""
4be3595… lmata 172 self.store.create_edge(
758c6b7… lmata 173 target_label,
758c6b7… lmata 174 {"name": target_name},
4be3595… lmata 175 EdgeType.ASSIGNED_TO,
758c6b7… lmata 176 NodeLabel.Person,
758c6b7… lmata 177 {"name": person_name},
4be3595… lmata 178 )
4be3595… lmata 179
4be3595… lmata 180 # ── Wiki pages ────────────────────────────────────────────────────────────
4be3595… lmata 181
4be3595… lmata 182 def wiki_page(
4be3595… lmata 183 self,
4be3595… lmata 184 name: str,
4be3595… lmata 185 url: str = "",
4be3595… lmata 186 source: str = "github",
4be3595… lmata 187 content: str = "",
4be3595… lmata 188 updated_at: str = "",
4be3595… lmata 189 ) -> None:
758c6b7… lmata 190 self.store.create_node(
758c6b7… lmata 191 NodeLabel.WikiPage,
758c6b7… lmata 192 {
758c6b7… lmata 193 "name": name,
758c6b7… lmata 194 "url": url,
758c6b7… lmata 195 "source": source,
758c6b7… lmata 196 "content": content,
758c6b7… lmata 197 "updated_at": updated_at,
758c6b7… lmata 198 },
758c6b7… lmata 199 )
4be3595… lmata 200 logger.info("WikiPage: %s", name)
4be3595… lmata 201
758c6b7… lmata 202 def wiki_documents(
758c6b7… lmata 203 self,
758c6b7… lmata 204 wiki_page_name: str,
758c6b7… lmata 205 target_name: str,
758c6b7… lmata 206 target_props: dict[str, Any],
758c6b7… lmata 207 target_label: NodeLabel,
758c6b7… lmata 208 ) -> None:
4be3595… lmata 209 self.store.create_edge(
758c6b7… lmata 210 NodeLabel.WikiPage,
758c6b7… lmata 211 {"name": wiki_page_name},
4be3595… lmata 212 EdgeType.DOCUMENTS,
758c6b7… lmata 213 target_label,
758c6b7… lmata 214 target_props,
4be3595… lmata 215 )
4be3595… lmata 216
4be3595… lmata 217 # ── Code ↔ Knowledge bridges ──────────────────────────────────────────────
4be3595… lmata 218
4be3595… lmata 219 def annotate_code(
4be3595… lmata 220 self,
4be3595… lmata 221 code_name: str,
4be3595… lmata 222 code_label: str,
4be3595… lmata 223 concept: str | None = None,
4be3595… lmata 224 rule: str | None = None,
4be3595… lmata 225 ) -> None:
4be3595… lmata 226 """
4be3595… lmata 227 Link a code node to a concept or rule via ANNOTATES.
4be3595… lmata 228 code_label should be a string matching a NodeLabel value.
4be3595… lmata 229 """
4be3595… lmata 230 label = NodeLabel(code_label)
4be3595… lmata 231 if concept:
4be3595… lmata 232 self.store.create_edge(
758c6b7… lmata 233 NodeLabel.Concept,
758c6b7… lmata 234 {"name": concept},
4be3595… lmata 235 EdgeType.ANNOTATES,
758c6b7… lmata 236 label,
758c6b7… lmata 237 {"name": code_name},
4be3595… lmata 238 )
4be3595… lmata 239 if rule:
4be3595… lmata 240 self.store.create_edge(
758c6b7… lmata 241 NodeLabel.Rule,
758c6b7… lmata 242 {"name": rule},
4be3595… lmata 243 EdgeType.ANNOTATES,
758c6b7… lmata 244 label,
758c6b7… lmata 245 {"name": code_name},
4be3595… lmata 246 )
4be3595… lmata 247
4be3595… lmata 248 def code_implements(self, code_name: str, code_label: str, concept_name: str) -> None:
4be3595… lmata 249 """Mark a function/class as implementing a concept."""
4be3595… lmata 250 label = NodeLabel(code_label)
4be3595… lmata 251 self.store.create_edge(
758c6b7… lmata 252 label,
758c6b7… lmata 253 {"name": code_name},
4be3595… lmata 254 EdgeType.IMPLEMENTS,
758c6b7… lmata 255 NodeLabel.Concept,
758c6b7… lmata 256 {"name": concept_name},
4be3595… lmata 257 )
4be3595… lmata 258
4be3595… lmata 259 # ── Helpers ───────────────────────────────────────────────────────────────
4be3595… lmata 260
4be3595… lmata 261 def _link_to_domain(self, name: str, label: NodeLabel, domain: str) -> None:
4be3595… lmata 262 # Ensure domain node exists
4be3595… lmata 263 self.store.create_node(NodeLabel.Domain, {"name": domain, "description": ""})
4be3595… lmata 264 self.store.create_edge(
758c6b7… lmata 265 label,
758c6b7… lmata 266 {"name": name},
4be3595… lmata 267 EdgeType.BELONGS_TO,
758c6b7… lmata 268 NodeLabel.Domain,
758c6b7… lmata 269 {"name": domain},
4be3595… lmata 270 )

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button