|
1
|
""" |
|
2
|
PlanopticonPipeline — first-class pipeline for meeting recordings → knowledge graph. |
|
3
|
|
|
4
|
Orchestrates: detect input type → ingest → link to code → return stats. |
|
5
|
|
|
6
|
Issues: #7 (first-class pipeline), #18 (action items, decision timeline, auto-linking) |
|
7
|
|
|
8
|
Usage:: |
|
9
|
|
|
10
|
from navegador.planopticon_pipeline import PlanopticonPipeline |
|
11
|
|
|
12
|
pipeline = PlanopticonPipeline(store) |
|
13
|
stats = pipeline.run("planopticon-output/", source_tag="Q4 Planning") |
|
14
|
|
|
15
|
items = pipeline.extract_action_items(kg_data) |
|
16
|
timeline = pipeline.build_decision_timeline(store) |
|
17
|
linked = pipeline.auto_link_to_code(store) |
|
18
|
""" |
|
19
|
|
|
20
|
from __future__ import annotations |
|
21
|
|
|
22
|
import logging |
|
23
|
from dataclasses import dataclass |
|
24
|
from pathlib import Path |
|
25
|
from typing import Any |
|
26
|
|
|
27
|
from navegador.graph.store import GraphStore |
|
28
|
|
|
29
|
logger = logging.getLogger(__name__) |
|
30
|
|
|
31
|
|
|
32
|
# ── Data models ─────────────────────────────────────────────────────────────── |
|
33
|
|
|
34
|
|
|
35
|
@dataclass |
|
36
|
class ActionItem: |
|
37
|
"""A single action item extracted from planopticon KG data.""" |
|
38
|
|
|
39
|
action: str |
|
40
|
assignee: str = "" |
|
41
|
context: str = "" |
|
42
|
priority: str = "info" |
|
43
|
source: str = "" |
|
44
|
|
|
45
|
def to_dict(self) -> dict[str, Any]: |
|
46
|
return { |
|
47
|
"action": self.action, |
|
48
|
"assignee": self.assignee, |
|
49
|
"context": self.context, |
|
50
|
"priority": self.priority, |
|
51
|
"source": self.source, |
|
52
|
} |
|
53
|
|
|
54
|
|
|
55
|
# ── Pipeline ────────────────────────────────────────────────────────────────── |
|
56
|
|
|
57
|
|
|
58
|
class PlanopticonPipeline: |
|
59
|
""" |
|
60
|
Orchestrates the full planopticon → knowledge graph pipeline. |
|
61
|
|
|
62
|
Steps: |
|
63
|
1. Detect input type from path (manifest, interchange, batch, kg) |
|
64
|
2. Delegate to PlanopticonIngester |
|
65
|
3. Run auto-link step to connect knowledge nodes to code symbols |
|
66
|
4. Return merged stats |
|
67
|
""" |
|
68
|
|
|
69
|
def __init__(self, store: GraphStore, source_tag: str = "") -> None: |
|
70
|
self.store = store |
|
71
|
self.source_tag = source_tag |
|
72
|
|
|
73
|
# ── Main entry point ────────────────────────────────────────────────────── |
|
74
|
|
|
75
|
def run(self, input_path: str | Path, source_tag: str = "") -> dict[str, Any]: |
|
76
|
""" |
|
77
|
Ingest a planopticon output and auto-link knowledge to code. |
|
78
|
|
|
79
|
Parameters |
|
80
|
---------- |
|
81
|
input_path: |
|
82
|
Path to a manifest.json, interchange.json, batch manifest, |
|
83
|
knowledge_graph.json, or a planopticon output directory. |
|
84
|
source_tag: |
|
85
|
Optional label for provenance tracking (overrides self.source_tag). |
|
86
|
|
|
87
|
Returns |
|
88
|
------- |
|
89
|
dict with keys: nodes, edges, linked (code links created) |
|
90
|
""" |
|
91
|
from navegador.ingestion.planopticon import PlanopticonIngester |
|
92
|
|
|
93
|
tag = source_tag or self.source_tag |
|
94
|
ingester = PlanopticonIngester(self.store, source_tag=tag) |
|
95
|
|
|
96
|
p = Path(input_path) |
|
97
|
input_type, resolved = self._detect_input(p) |
|
98
|
|
|
99
|
dispatch = { |
|
100
|
"manifest": ingester.ingest_manifest, |
|
101
|
"interchange": ingester.ingest_interchange, |
|
102
|
"batch": ingester.ingest_batch, |
|
103
|
"kg": ingester.ingest_kg, |
|
104
|
} |
|
105
|
stats = dispatch[input_type](resolved) |
|
106
|
|
|
107
|
# Auto-link newly ingested knowledge nodes to code symbols |
|
108
|
linked = self.auto_link_to_code(self.store) |
|
109
|
stats["linked"] = linked |
|
110
|
|
|
111
|
logger.info( |
|
112
|
"PlanopticonPipeline.run: type=%s nodes=%d edges=%d linked=%d", |
|
113
|
input_type, |
|
114
|
stats.get("nodes", 0), |
|
115
|
stats.get("edges", 0), |
|
116
|
linked, |
|
117
|
) |
|
118
|
return stats |
|
119
|
|
|
120
|
# ── Action items ────────────────────────────────────────────────────────── |
|
121
|
|
|
122
|
@staticmethod |
|
123
|
def extract_action_items(kg_data: dict[str, Any]) -> list[ActionItem]: |
|
124
|
""" |
|
125
|
Extract action items from raw planopticon KG data dict. |
|
126
|
|
|
127
|
Looks at ``action_items`` (manifest format) as well as nodes whose |
|
128
|
planning_type is ``"task"`` or whose type is ``"action_item"``. |
|
129
|
|
|
130
|
Parameters |
|
131
|
---------- |
|
132
|
kg_data: |
|
133
|
A dict as loaded from manifest.json, interchange.json, |
|
134
|
knowledge_graph.json, or any combination that may contain an |
|
135
|
``action_items`` list or ``entities``/``nodes`` with task types. |
|
136
|
""" |
|
137
|
items: list[ActionItem] = [] |
|
138
|
|
|
139
|
source = kg_data.get("video", {}).get("title", "") or kg_data.get("project", {}).get( |
|
140
|
"name", "" |
|
141
|
) |
|
142
|
|
|
143
|
# Explicit action_items list (manifest format) |
|
144
|
for raw in kg_data.get("action_items", []): |
|
145
|
action = (raw.get("action") or "").strip() |
|
146
|
if not action: |
|
147
|
continue |
|
148
|
items.append( |
|
149
|
ActionItem( |
|
150
|
action=action, |
|
151
|
assignee=(raw.get("assignee") or "").strip(), |
|
152
|
context=raw.get("context", ""), |
|
153
|
priority=raw.get("priority", "info"), |
|
154
|
source=source, |
|
155
|
) |
|
156
|
) |
|
157
|
|
|
158
|
# Entities / nodes with task/action_item planning type |
|
159
|
for entity in kg_data.get("entities", []) + kg_data.get("nodes", []): |
|
160
|
ptype = entity.get("planning_type", "") or entity.get("type", "") |
|
161
|
if ptype not in ("task", "action_item"): |
|
162
|
continue |
|
163
|
name = (entity.get("name") or "").strip() |
|
164
|
if not name: |
|
165
|
continue |
|
166
|
items.append( |
|
167
|
ActionItem( |
|
168
|
action=name, |
|
169
|
assignee=(entity.get("assignee") or "").strip(), |
|
170
|
context=entity.get("description", ""), |
|
171
|
priority=entity.get("priority", "info"), |
|
172
|
source=source, |
|
173
|
) |
|
174
|
) |
|
175
|
|
|
176
|
return items |
|
177
|
|
|
178
|
# ── Decision timeline ───────────────────────────────────────────────────── |
|
179
|
|
|
180
|
@staticmethod |
|
181
|
def build_decision_timeline(store: GraphStore) -> list[dict[str, Any]]: |
|
182
|
""" |
|
183
|
Return all Decision nodes ordered chronologically by their ``date`` property. |
|
184
|
|
|
185
|
Nodes without a date are placed last, sorted by name. |
|
186
|
|
|
187
|
Returns |
|
188
|
------- |
|
189
|
list of dicts with keys: name, description, domain, status, rationale, date |
|
190
|
""" |
|
191
|
cypher = ( |
|
192
|
"MATCH (d:Decision) " |
|
193
|
"RETURN d.name, d.description, d.domain, d.status, d.rationale, d.date " |
|
194
|
"ORDER BY d.date, d.name" |
|
195
|
) |
|
196
|
try: |
|
197
|
result = store.query(cypher) |
|
198
|
rows = result.result_set or [] |
|
199
|
except Exception: |
|
200
|
logger.warning("build_decision_timeline: query failed", exc_info=True) |
|
201
|
return [] |
|
202
|
|
|
203
|
timeline = [] |
|
204
|
for row in rows: |
|
205
|
timeline.append( |
|
206
|
{ |
|
207
|
"name": row[0] or "", |
|
208
|
"description": row[1] or "", |
|
209
|
"domain": row[2] or "", |
|
210
|
"status": row[3] or "", |
|
211
|
"rationale": row[4] or "", |
|
212
|
"date": row[5] or "", |
|
213
|
} |
|
214
|
) |
|
215
|
return timeline |
|
216
|
|
|
217
|
# ── Auto-link to code ───────────────────────────────────────────────────── |
|
218
|
|
|
219
|
@staticmethod |
|
220
|
def auto_link_to_code(store: GraphStore) -> int: |
|
221
|
""" |
|
222
|
Match knowledge nodes to code symbols by name similarity and create |
|
223
|
ANNOTATES edges. |
|
224
|
|
|
225
|
Strategy: |
|
226
|
For each Concept / Decision / Rule node, look for Function, Class, or |
|
227
|
Method nodes whose ``name`` contains a significant word from the |
|
228
|
knowledge node's name (case-insensitive, length ≥ 4 chars). |
|
229
|
|
|
230
|
Returns |
|
231
|
------- |
|
232
|
int — number of new ANNOTATES edges created |
|
233
|
""" |
|
234
|
# Fetch all knowledge nodes |
|
235
|
knowledge_cypher = ( |
|
236
|
"MATCH (k) WHERE k:Concept OR k:Decision OR k:Rule RETURN labels(k)[0], k.name" |
|
237
|
) |
|
238
|
code_cypher = ( |
|
239
|
"MATCH (c) WHERE c:Function OR c:Class OR c:Method RETURN labels(c)[0], c.name" |
|
240
|
) |
|
241
|
|
|
242
|
try: |
|
243
|
k_result = store.query(knowledge_cypher) |
|
244
|
c_result = store.query(code_cypher) |
|
245
|
except Exception: |
|
246
|
logger.warning("auto_link_to_code: initial queries failed", exc_info=True) |
|
247
|
return 0 |
|
248
|
|
|
249
|
knowledge_nodes: list[tuple[str, str]] = [ |
|
250
|
(str(row[0]), str(row[1])) for row in (k_result.result_set or []) if row[0] and row[1] |
|
251
|
] |
|
252
|
code_nodes: list[tuple[str, str]] = [ |
|
253
|
(str(row[0]), str(row[1])) for row in (c_result.result_set or []) if row[0] and row[1] |
|
254
|
] |
|
255
|
|
|
256
|
if not knowledge_nodes or not code_nodes: |
|
257
|
return 0 |
|
258
|
|
|
259
|
linked = 0 |
|
260
|
for k_label, k_name in knowledge_nodes: |
|
261
|
# Extract significant tokens (length >= 4) from the knowledge name |
|
262
|
tokens = [ |
|
263
|
w.lower() for w in k_name.replace("_", " ").replace("-", " ").split() if len(w) >= 4 |
|
264
|
] |
|
265
|
if not tokens: |
|
266
|
continue |
|
267
|
|
|
268
|
for c_label, c_name in code_nodes: |
|
269
|
c_lower = c_name.lower() |
|
270
|
if any(tok in c_lower for tok in tokens): |
|
271
|
# Create ANNOTATES edge from knowledge node to code node |
|
272
|
cypher = ( |
|
273
|
"MATCH (k:" + k_label + " {name: $kn}), (c:" + c_label + " {name: $cn}) " |
|
274
|
"MERGE (k)-[r:ANNOTATES]->(c)" |
|
275
|
) |
|
276
|
try: |
|
277
|
store.query(cypher, {"kn": k_name, "cn": c_name}) |
|
278
|
linked += 1 |
|
279
|
except Exception: |
|
280
|
logger.debug("auto_link_to_code: could not link %s → %s", k_name, c_name) |
|
281
|
|
|
282
|
return linked |
|
283
|
|
|
284
|
# ── Helpers ─────────────────────────────────────────────────────────────── |
|
285
|
|
|
286
|
@staticmethod |
|
287
|
def _detect_input(p: Path) -> tuple[str, Path]: |
|
288
|
""" |
|
289
|
Detect input type and resolve to a concrete file path. |
|
290
|
|
|
291
|
Returns |
|
292
|
------- |
|
293
|
(input_type, resolved_path) where input_type is one of: |
|
294
|
"manifest", "interchange", "batch", "kg" |
|
295
|
""" |
|
296
|
if p.is_dir(): |
|
297
|
candidates = [ |
|
298
|
("manifest", p / "manifest.json"), |
|
299
|
("interchange", p / "interchange.json"), |
|
300
|
("batch", p / "batch_manifest.json"), |
|
301
|
("kg", p / "results" / "knowledge_graph.json"), |
|
302
|
("kg", p / "knowledge_graph.json"), |
|
303
|
] |
|
304
|
for itype, candidate in candidates: |
|
305
|
if candidate.exists(): |
|
306
|
return itype, candidate |
|
307
|
raise FileNotFoundError( |
|
308
|
f"No recognised planopticon file found in {p}. " |
|
309
|
"Expected manifest.json, interchange.json, " |
|
310
|
"batch_manifest.json, or knowledge_graph.json." |
|
311
|
) |
|
312
|
|
|
313
|
name = p.name.lower() |
|
314
|
if "manifest" in name and "batch" not in name: |
|
315
|
return "manifest", p |
|
316
|
if "interchange" in name: |
|
317
|
return "interchange", p |
|
318
|
if "batch" in name: |
|
319
|
return "batch", p |
|
320
|
# Default: treat as knowledge_graph.json |
|
321
|
return "kg", p |
|
322
|
|