PlanOpticon

planopticon / video_processor / exporters / markdown.py
Source Blame History 500 lines
0981a08… noreply 1 """Generate structured markdown documents from knowledge graphs.
0981a08… noreply 2
0981a08… noreply 3 No LLM required — pure template-based generation from KG data.
0981a08… noreply 4 Produces federated, curated notes suitable for Obsidian, Notion,
0981a08… noreply 5 GitHub, or any markdown-based workflow.
0981a08… noreply 6 """
0981a08… noreply 7
0981a08… noreply 8 import csv
0981a08… noreply 9 import io
0981a08… noreply 10 import logging
0981a08… noreply 11 from datetime import datetime
0981a08… noreply 12 from pathlib import Path
0981a08… noreply 13 from typing import Dict, List, Optional
0981a08… noreply 14
0981a08… noreply 15 logger = logging.getLogger(__name__)
0981a08… noreply 16
0981a08… noreply 17
0981a08… noreply 18 def _heading(text: str, level: int = 1) -> str:
0981a08… noreply 19 return f"{'#' * level} {text}"
0981a08… noreply 20
0981a08… noreply 21
0981a08… noreply 22 def _table(headers: List[str], rows: List[List[str]]) -> str:
0981a08… noreply 23 lines = ["| " + " | ".join(headers) + " |"]
0981a08… noreply 24 lines.append("| " + " | ".join("---" for _ in headers) + " |")
0981a08… noreply 25 for row in rows:
0981a08… noreply 26 lines.append("| " + " | ".join(str(c) for c in row) + " |")
0981a08… noreply 27 return "\n".join(lines)
0981a08… noreply 28
0981a08… noreply 29
0981a08… noreply 30 def _badge(label: str, value: str) -> str:
0981a08… noreply 31 return f"**{label}:** {value}"
0981a08… noreply 32
0981a08… noreply 33
0981a08… noreply 34 # ---------------------------------------------------------------------------
0981a08… noreply 35 # Individual document generators
0981a08… noreply 36 # ---------------------------------------------------------------------------
0981a08… noreply 37
0981a08… noreply 38
0981a08… noreply 39 def generate_entity_brief(entity: dict, relationships: list) -> str:
0981a08… noreply 40 """Generate a one-pager markdown brief for a single entity."""
0981a08… noreply 41 name = entity.get("name", "Untitled")
0981a08… noreply 42 etype = entity.get("type", "concept")
0981a08… noreply 43 descs = entity.get("descriptions", [])
0981a08… noreply 44 occs = entity.get("occurrences", [])
0981a08… noreply 45
0981a08… noreply 46 outgoing = [(r["target"], r["type"]) for r in relationships if r.get("source") == name]
0981a08… noreply 47 incoming = [(r["source"], r["type"]) for r in relationships if r.get("target") == name]
0981a08… noreply 48
0981a08… noreply 49 parts = [
0981a08… noreply 50 _heading(name),
0981a08… noreply 51 "",
0981a08… noreply 52 _badge("Type", etype),
0981a08… noreply 53 "",
0981a08… noreply 54 ]
0981a08… noreply 55
0981a08… noreply 56 if descs:
0981a08… noreply 57 parts.append(_heading("Summary", 2))
0981a08… noreply 58 parts.append("")
0981a08… noreply 59 for d in descs:
0981a08… noreply 60 parts.append(f"- {d}")
0981a08… noreply 61 parts.append("")
0981a08… noreply 62
0981a08… noreply 63 if outgoing:
0981a08… noreply 64 parts.append(_heading("Relates To", 2))
0981a08… noreply 65 parts.append("")
0981a08… noreply 66 parts.append(_table(["Entity", "Relationship"], [[t, r] for t, r in outgoing]))
0981a08… noreply 67 parts.append("")
0981a08… noreply 68
0981a08… noreply 69 if incoming:
0981a08… noreply 70 parts.append(_heading("Referenced By", 2))
0981a08… noreply 71 parts.append("")
0981a08… noreply 72 parts.append(_table(["Entity", "Relationship"], [[s, r] for s, r in incoming]))
0981a08… noreply 73 parts.append("")
0981a08… noreply 74
0981a08… noreply 75 if occs:
0981a08… noreply 76 parts.append(_heading("Sources", 2))
0981a08… noreply 77 parts.append("")
0981a08… noreply 78 for occ in occs:
0981a08… noreply 79 src = occ.get("source", "unknown")
0981a08… noreply 80 ts = occ.get("timestamp", "")
0981a08… noreply 81 text = occ.get("text", "")
0981a08… noreply 82 line = f"- **{src}**"
0981a08… noreply 83 if ts:
0981a08… noreply 84 line += f" ({ts})"
0981a08… noreply 85 if text:
0981a08… noreply 86 line += f" — {text}"
0981a08… noreply 87 parts.append(line)
0981a08… noreply 88 parts.append("")
0981a08… noreply 89
0981a08… noreply 90 return "\n".join(parts)
0981a08… noreply 91
0981a08… noreply 92
0981a08… noreply 93 def generate_executive_summary(kg_data: dict) -> str:
0981a08… noreply 94 """Generate a high-level executive summary from the KG."""
0981a08… noreply 95 nodes = kg_data.get("nodes", [])
0981a08… noreply 96 rels = kg_data.get("relationships", [])
0981a08… noreply 97
0981a08… noreply 98 by_type: Dict[str, list] = {}
0981a08… noreply 99 for n in nodes:
0981a08… noreply 100 t = n.get("type", "concept")
0981a08… noreply 101 by_type.setdefault(t, []).append(n)
0981a08… noreply 102
0981a08… noreply 103 parts = [
0981a08… noreply 104 _heading("Executive Summary"),
0981a08… noreply 105 "",
0981a08… noreply 106 f"Knowledge base contains **{len(nodes)} entities** "
0981a08… noreply 107 f"and **{len(rels)} relationships** across "
0981a08… noreply 108 f"**{len(by_type)} categories**.",
0981a08… noreply 109 "",
0981a08… noreply 110 _heading("Entity Breakdown", 2),
0981a08… noreply 111 "",
0981a08… noreply 112 _table(
0981a08… noreply 113 ["Type", "Count", "Examples"],
0981a08… noreply 114 [
0981a08… noreply 115 [
0981a08… noreply 116 etype,
0981a08… noreply 117 str(len(elist)),
0981a08… noreply 118 ", ".join(e.get("name", "") for e in elist[:3]),
0981a08… noreply 119 ]
0981a08… noreply 120 for etype, elist in sorted(by_type.items(), key=lambda x: -len(x[1]))
0981a08… noreply 121 ],
0981a08… noreply 122 ),
0981a08… noreply 123 "",
0981a08… noreply 124 ]
0981a08… noreply 125
0981a08… noreply 126 # Top connected entities
0981a08… noreply 127 degree: Dict[str, int] = {}
0981a08… noreply 128 for r in rels:
0981a08… noreply 129 degree[r.get("source", "")] = degree.get(r.get("source", ""), 0) + 1
0981a08… noreply 130 degree[r.get("target", "")] = degree.get(r.get("target", ""), 0) + 1
0981a08… noreply 131
0981a08… noreply 132 top = sorted(degree.items(), key=lambda x: -x[1])[:10]
0981a08… noreply 133 if top:
0981a08… noreply 134 parts.append(_heading("Key Entities (by connections)", 2))
0981a08… noreply 135 parts.append("")
0981a08… noreply 136 parts.append(
0981a08… noreply 137 _table(
0981a08… noreply 138 ["Entity", "Connections"],
0981a08… noreply 139 [[name, str(deg)] for name, deg in top],
0981a08… noreply 140 )
0981a08… noreply 141 )
0981a08… noreply 142 parts.append("")
0981a08… noreply 143
0981a08… noreply 144 # Relationship type breakdown
0981a08… noreply 145 rel_types: Dict[str, int] = {}
0981a08… noreply 146 for r in rels:
0981a08… noreply 147 rt = r.get("type", "related_to")
0981a08… noreply 148 rel_types[rt] = rel_types.get(rt, 0) + 1
0981a08… noreply 149
0981a08… noreply 150 if rel_types:
0981a08… noreply 151 parts.append(_heading("Relationship Types", 2))
0981a08… noreply 152 parts.append("")
0981a08… noreply 153 parts.append(
0981a08… noreply 154 _table(
0981a08… noreply 155 ["Type", "Count"],
0981a08… noreply 156 [[rt, str(c)] for rt, c in sorted(rel_types.items(), key=lambda x: -x[1])],
0981a08… noreply 157 )
0981a08… noreply 158 )
0981a08… noreply 159 parts.append("")
0981a08… noreply 160
0981a08… noreply 161 return "\n".join(parts)
0981a08… noreply 162
0981a08… noreply 163
0981a08… noreply 164 def generate_meeting_notes(kg_data: dict, title: Optional[str] = None) -> str:
0981a08… noreply 165 """Generate meeting notes format from KG data."""
0981a08… noreply 166 nodes = kg_data.get("nodes", [])
0981a08… noreply 167 rels = kg_data.get("relationships", [])
0981a08… noreply 168 title = title or "Meeting Notes"
0981a08… noreply 169
0981a08… noreply 170 # Categorize by planning-relevant types
0981a08… noreply 171 decisions = [n for n in nodes if n.get("type") in ("decision", "constraint")]
0981a08… noreply 172 actions = [n for n in nodes if n.get("type") in ("goal", "feature", "milestone")]
0981a08… noreply 173 people = [n for n in nodes if n.get("type") == "person"]
0981a08… noreply 174 topics = [n for n in nodes if n.get("type") in ("concept", "technology", "topic")]
0981a08… noreply 175
0981a08… noreply 176 parts = [
0981a08… noreply 177 _heading(title),
0981a08… noreply 178 "",
0981a08… noreply 179 f"*Generated {datetime.now().strftime('%Y-%m-%d %H:%M')}*",
0981a08… noreply 180 "",
0981a08… noreply 181 ]
0981a08… noreply 182
0981a08… noreply 183 if topics:
0981a08… noreply 184 parts.append(_heading("Discussion Topics", 2))
0981a08… noreply 185 parts.append("")
0981a08… noreply 186 for t in topics:
0981a08… noreply 187 descs = t.get("descriptions", [])
0981a08… noreply 188 desc = descs[0] if descs else ""
0981a08… noreply 189 parts.append(f"- **{t['name']}**: {desc}")
0981a08… noreply 190 parts.append("")
0981a08… noreply 191
0981a08… noreply 192 if people:
0981a08… noreply 193 parts.append(_heading("Participants", 2))
0981a08… noreply 194 parts.append("")
0981a08… noreply 195 for p in people:
0981a08… noreply 196 parts.append(f"- {p['name']}")
0981a08… noreply 197 parts.append("")
0981a08… noreply 198
0981a08… noreply 199 if decisions:
0981a08… noreply 200 parts.append(_heading("Decisions & Constraints", 2))
0981a08… noreply 201 parts.append("")
0981a08… noreply 202 for d in decisions:
0981a08… noreply 203 descs = d.get("descriptions", [])
0981a08… noreply 204 desc = descs[0] if descs else ""
0981a08… noreply 205 parts.append(f"- **{d['name']}**: {desc}")
0981a08… noreply 206 parts.append("")
0981a08… noreply 207
0981a08… noreply 208 if actions:
0981a08… noreply 209 parts.append(_heading("Action Items", 2))
0981a08… noreply 210 parts.append("")
0981a08… noreply 211 for a in actions:
0981a08… noreply 212 descs = a.get("descriptions", [])
0981a08… noreply 213 desc = descs[0] if descs else ""
0981a08… noreply 214 # Find who it's related to
0981a08… noreply 215 owners = [
0981a08… noreply 216 r["target"]
0981a08… noreply 217 for r in rels
0981a08… noreply 218 if r.get("source") == a["name"] and r.get("type") in ("assigned_to", "owned_by")
0981a08… noreply 219 ]
0981a08… noreply 220 owner_str = f" (@{', '.join(owners)})" if owners else ""
0981a08… noreply 221 parts.append(f"- [ ] **{a['name']}**{owner_str}: {desc}")
0981a08… noreply 222 parts.append("")
0981a08… noreply 223
0981a08… noreply 224 # Open questions (entities without many relationships)
0981a08… noreply 225 degree_map: Dict[str, int] = {}
0981a08… noreply 226 for r in rels:
0981a08… noreply 227 degree_map[r.get("source", "")] = degree_map.get(r.get("source", ""), 0) + 1
0981a08… noreply 228 degree_map[r.get("target", "")] = degree_map.get(r.get("target", ""), 0) + 1
0981a08… noreply 229
0981a08… noreply 230 orphans = [n for n in nodes if degree_map.get(n.get("name", ""), 0) <= 1 and n not in people]
0981a08… noreply 231 if orphans:
0981a08… noreply 232 parts.append(_heading("Open Questions / Loose Ends", 2))
0981a08… noreply 233 parts.append("")
0981a08… noreply 234 for o in orphans[:10]:
0981a08… noreply 235 parts.append(f"- {o['name']}")
0981a08… noreply 236 parts.append("")
0981a08… noreply 237
0981a08… noreply 238 return "\n".join(parts)
0981a08… noreply 239
0981a08… noreply 240
0981a08… noreply 241 def generate_glossary(kg_data: dict) -> str:
0981a08… noreply 242 """Generate a glossary/dictionary of all entities."""
0981a08… noreply 243 nodes = sorted(kg_data.get("nodes", []), key=lambda n: n.get("name", "").lower())
0981a08… noreply 244
0981a08… noreply 245 parts = [
0981a08… noreply 246 _heading("Glossary"),
0981a08… noreply 247 "",
0981a08… noreply 248 ]
0981a08… noreply 249
0981a08… noreply 250 for node in nodes:
0981a08… noreply 251 name = node.get("name", "")
0981a08… noreply 252 etype = node.get("type", "concept")
0981a08… noreply 253 descs = node.get("descriptions", [])
0981a08… noreply 254 desc = descs[0] if descs else "No description available."
0981a08… noreply 255 parts.append(f"**{name}** *({etype})*")
0981a08… noreply 256 parts.append(f": {desc}")
0981a08… noreply 257 parts.append("")
0981a08… noreply 258
0981a08… noreply 259 return "\n".join(parts)
0981a08… noreply 260
0981a08… noreply 261
0981a08… noreply 262 def generate_relationship_map(kg_data: dict) -> str:
0981a08… noreply 263 """Generate a relationship map as a markdown document with Mermaid diagram."""
0981a08… noreply 264 rels = kg_data.get("relationships", [])
0981a08… noreply 265 nodes = kg_data.get("nodes", [])
0981a08… noreply 266
0981a08… noreply 267 parts = [
0981a08… noreply 268 _heading("Relationship Map"),
0981a08… noreply 269 "",
0981a08… noreply 270 f"*{len(nodes)} entities, {len(rels)} relationships*",
0981a08… noreply 271 "",
0981a08… noreply 272 ]
0981a08… noreply 273
0981a08… noreply 274 # Group by relationship type
0981a08… noreply 275 by_type: Dict[str, list] = {}
0981a08… noreply 276 for r in rels:
0981a08… noreply 277 rt = r.get("type", "related_to")
0981a08… noreply 278 by_type.setdefault(rt, []).append(r)
0981a08… noreply 279
0981a08… noreply 280 for rt, rlist in sorted(by_type.items()):
0981a08… noreply 281 parts.append(_heading(rt.replace("_", " ").title(), 2))
0981a08… noreply 282 parts.append("")
0981a08… noreply 283 parts.append(
0981a08… noreply 284 _table(
0981a08… noreply 285 ["Source", "Target"],
0981a08… noreply 286 [[r.get("source", ""), r.get("target", "")] for r in rlist],
0981a08… noreply 287 )
0981a08… noreply 288 )
0981a08… noreply 289 parts.append("")
0981a08… noreply 290
0981a08… noreply 291 # Mermaid diagram (top 20 nodes by degree)
0981a08… noreply 292 degree: Dict[str, int] = {}
0981a08… noreply 293 for r in rels:
0981a08… noreply 294 degree[r.get("source", "")] = degree.get(r.get("source", ""), 0) + 1
0981a08… noreply 295 degree[r.get("target", "")] = degree.get(r.get("target", ""), 0) + 1
0981a08… noreply 296
0981a08… noreply 297 top_nodes = {name for name, _ in sorted(degree.items(), key=lambda x: -x[1])[:20]}
0981a08… noreply 298
0981a08… noreply 299 if top_nodes:
0981a08… noreply 300 parts.append(_heading("Visual Map", 2))
0981a08… noreply 301 parts.append("")
0981a08… noreply 302 parts.append("```mermaid")
0981a08… noreply 303 parts.append("graph LR")
0981a08… noreply 304
0981a08… noreply 305 def safe(s):
0981a08… noreply 306 return "".join(c if c.isalnum() or c == "_" else "_" for c in s)
0981a08… noreply 307
0981a08… noreply 308 seen = set()
0981a08… noreply 309 for r in rels:
0981a08… noreply 310 src, tgt = r.get("source", ""), r.get("target", "")
0981a08… noreply 311 if src in top_nodes and tgt in top_nodes:
0981a08… noreply 312 key = (src, tgt)
0981a08… noreply 313 if key not in seen:
0981a08… noreply 314 parts.append(
0981a08… noreply 315 f' {safe(src)}["{src}"] -->|{r.get("type", "")}| {safe(tgt)}["{tgt}"]'
0981a08… noreply 316 )
0981a08… noreply 317 seen.add(key)
0981a08… noreply 318 parts.append("```")
0981a08… noreply 319 parts.append("")
0981a08… noreply 320
0981a08… noreply 321 return "\n".join(parts)
0981a08… noreply 322
0981a08… noreply 323
0981a08… noreply 324 def generate_status_report(kg_data: dict, title: Optional[str] = None) -> str:
0981a08… noreply 325 """Generate a project status report from KG data."""
0981a08… noreply 326 nodes = kg_data.get("nodes", [])
0981a08… noreply 327 rels = kg_data.get("relationships", [])
0981a08… noreply 328 title = title or "Status Report"
0981a08… noreply 329
0981a08… noreply 330 milestones = [n for n in nodes if n.get("type") == "milestone"]
0981a08… noreply 331 features = [n for n in nodes if n.get("type") == "feature"]
0981a08… noreply 332 risks = [n for n in nodes if n.get("type") in ("risk", "constraint")]
0981a08… noreply 333 requirements = [n for n in nodes if n.get("type") == "requirement"]
0981a08… noreply 334
0981a08… noreply 335 parts = [
0981a08… noreply 336 _heading(title),
0981a08… noreply 337 "",
0981a08… noreply 338 f"*Generated {datetime.now().strftime('%Y-%m-%d %H:%M')}*",
0981a08… noreply 339 "",
0981a08… noreply 340 ]
0981a08… noreply 341
0981a08… noreply 342 parts.append(_heading("Overview", 2))
0981a08… noreply 343 parts.append("")
0981a08… noreply 344 parts.append(f"- **Entities:** {len(nodes)}")
0981a08… noreply 345 parts.append(f"- **Relationships:** {len(rels)}")
0981a08… noreply 346 parts.append(f"- **Features:** {len(features)}")
0981a08… noreply 347 parts.append(f"- **Milestones:** {len(milestones)}")
0981a08… noreply 348 parts.append(f"- **Requirements:** {len(requirements)}")
0981a08… noreply 349 parts.append(f"- **Risks/Constraints:** {len(risks)}")
0981a08… noreply 350 parts.append("")
0981a08… noreply 351
0981a08… noreply 352 if milestones:
0981a08… noreply 353 parts.append(_heading("Milestones", 2))
0981a08… noreply 354 parts.append("")
0981a08… noreply 355 for m in milestones:
0981a08… noreply 356 descs = m.get("descriptions", [])
0981a08… noreply 357 parts.append(f"- **{m['name']}**: {descs[0] if descs else 'TBD'}")
0981a08… noreply 358 parts.append("")
0981a08… noreply 359
0981a08… noreply 360 if features:
0981a08… noreply 361 parts.append(_heading("Features", 2))
0981a08… noreply 362 parts.append("")
0981a08… noreply 363 parts.append(
0981a08… noreply 364 _table(
0981a08… noreply 365 ["Feature", "Description"],
0981a08… noreply 366 [[f["name"], (f.get("descriptions") or [""])[0][:60]] for f in features],
0981a08… noreply 367 )
0981a08… noreply 368 )
0981a08… noreply 369 parts.append("")
0981a08… noreply 370
0981a08… noreply 371 if risks:
0981a08… noreply 372 parts.append(_heading("Risks & Constraints", 2))
0981a08… noreply 373 parts.append("")
0981a08… noreply 374 for r in risks:
0981a08… noreply 375 descs = r.get("descriptions", [])
0981a08… noreply 376 parts.append(f"- **{r['name']}**: {descs[0] if descs else ''}")
0981a08… noreply 377 parts.append("")
0981a08… noreply 378
0981a08… noreply 379 return "\n".join(parts)
0981a08… noreply 380
0981a08… noreply 381
0981a08… noreply 382 def generate_entity_index(kg_data: dict) -> str:
0981a08… noreply 383 """Generate a master index of all entities grouped by type."""
0981a08… noreply 384 nodes = kg_data.get("nodes", [])
0981a08… noreply 385
0981a08… noreply 386 by_type: Dict[str, list] = {}
0981a08… noreply 387 for n in nodes:
0981a08… noreply 388 t = n.get("type", "concept")
0981a08… noreply 389 by_type.setdefault(t, []).append(n)
0981a08… noreply 390
0981a08… noreply 391 parts = [
0981a08… noreply 392 _heading("Entity Index"),
0981a08… noreply 393 "",
0981a08… noreply 394 f"*{len(nodes)} entities across {len(by_type)} types*",
0981a08… noreply 395 "",
0981a08… noreply 396 ]
0981a08… noreply 397
0981a08… noreply 398 for etype, elist in sorted(by_type.items()):
0981a08… noreply 399 parts.append(_heading(f"{etype.title()} ({len(elist)})", 2))
0981a08… noreply 400 parts.append("")
0981a08… noreply 401 for e in sorted(elist, key=lambda x: x.get("name", "")):
0981a08… noreply 402 descs = e.get("descriptions", [])
0981a08… noreply 403 desc = f" — {descs[0]}" if descs else ""
0981a08… noreply 404 parts.append(f"- **{e['name']}**{desc}")
0981a08… noreply 405 parts.append("")
0981a08… noreply 406
0981a08… noreply 407 return "\n".join(parts)
0981a08… noreply 408
0981a08… noreply 409
0981a08… noreply 410 def generate_csv_export(kg_data: dict) -> str:
0981a08… noreply 411 """Generate CSV of entities for spreadsheet import."""
0981a08… noreply 412 nodes = kg_data.get("nodes", [])
0981a08… noreply 413 rels = kg_data.get("relationships", [])
0981a08… noreply 414
0981a08… noreply 415 # Build adjacency info
0981a08… noreply 416 related: Dict[str, list] = {}
0981a08… noreply 417 for r in rels:
0981a08… noreply 418 src = r.get("source", "")
0981a08… noreply 419 tgt = r.get("target", "")
0981a08… noreply 420 related.setdefault(src, []).append(tgt)
0981a08… noreply 421
0981a08… noreply 422 output = io.StringIO()
0981a08… noreply 423 writer = csv.writer(output)
0981a08… noreply 424 writer.writerow(["Name", "Type", "Description", "Related To", "Source"])
0981a08… noreply 425
0981a08… noreply 426 for n in sorted(nodes, key=lambda x: x.get("name", "")):
0981a08… noreply 427 name = n.get("name", "")
0981a08… noreply 428 etype = n.get("type", "")
0981a08… noreply 429 descs = n.get("descriptions", [])
0981a08… noreply 430 desc = descs[0] if descs else ""
0981a08… noreply 431 rels_str = "; ".join(related.get(name, []))
0981a08… noreply 432 sources = n.get("occurrences", [])
0981a08… noreply 433 src_str = sources[0].get("source", "") if sources else ""
0981a08… noreply 434 writer.writerow([name, etype, desc, rels_str, src_str])
0981a08… noreply 435
0981a08… noreply 436 return output.getvalue()
0981a08… noreply 437
0981a08… noreply 438
0981a08… noreply 439 # ---------------------------------------------------------------------------
0981a08… noreply 440 # Document types registry
0981a08… noreply 441 # ---------------------------------------------------------------------------
0981a08… noreply 442
0981a08… noreply 443 DOCUMENT_TYPES = {
0981a08… noreply 444 "summary": ("Executive Summary", generate_executive_summary),
0981a08… noreply 445 "meeting-notes": ("Meeting Notes", generate_meeting_notes),
0981a08… noreply 446 "glossary": ("Glossary", generate_glossary),
0981a08… noreply 447 "relationship-map": ("Relationship Map", generate_relationship_map),
0981a08… noreply 448 "status-report": ("Status Report", generate_status_report),
0981a08… noreply 449 "entity-index": ("Entity Index", generate_entity_index),
0981a08… noreply 450 "csv": ("CSV Export", generate_csv_export),
0981a08… noreply 451 }
0981a08… noreply 452
0981a08… noreply 453
0981a08… noreply 454 def generate_all(
0981a08… noreply 455 kg_data: dict,
0981a08… noreply 456 output_dir: Path,
0981a08… noreply 457 doc_types: Optional[List[str]] = None,
0981a08… noreply 458 title: Optional[str] = None,
0981a08… noreply 459 ) -> List[Path]:
0981a08… noreply 460 """Generate multiple document types and write to output directory.
0981a08… noreply 461
0981a08… noreply 462 If doc_types is None, generates all available types.
0981a08… noreply 463 Returns list of created file paths.
0981a08… noreply 464 """
0981a08… noreply 465 output_dir.mkdir(parents=True, exist_ok=True)
0981a08… noreply 466 types_to_generate = doc_types or list(DOCUMENT_TYPES.keys())
0981a08… noreply 467 created = []
0981a08… noreply 468
0981a08… noreply 469 for dtype in types_to_generate:
0981a08… noreply 470 if dtype not in DOCUMENT_TYPES:
0981a08… noreply 471 logger.warning(f"Unknown document type: {dtype}")
0981a08… noreply 472 continue
0981a08… noreply 473
0981a08… noreply 474 label, generator = DOCUMENT_TYPES[dtype]
0981a08… noreply 475 try:
0981a08… noreply 476 content = generator(kg_data)
0981a08… noreply 477 ext = ".csv" if dtype == "csv" else ".md"
0981a08… noreply 478 filename = f"{dtype}{ext}"
0981a08… noreply 479 path = output_dir / filename
0981a08… noreply 480 path.write_text(content, encoding="utf-8")
0981a08… noreply 481 created.append(path)
0981a08… noreply 482 logger.info(f"Generated {label} → {path}")
0981a08… noreply 483 except Exception as e:
0981a08… noreply 484 logger.error(f"Failed to generate {label}: {e}")
0981a08… noreply 485
0981a08… noreply 486 # Also generate individual entity briefs
0981a08… noreply 487 briefs_dir = output_dir / "entities"
0981a08… noreply 488 briefs_dir.mkdir(exist_ok=True)
0981a08… noreply 489 rels = kg_data.get("relationships", [])
0981a08… noreply 490 for node in kg_data.get("nodes", []):
0981a08… noreply 491 name = node.get("name", "")
0981a08… noreply 492 if not name:
0981a08… noreply 493 continue
0981a08… noreply 494 safe = name.replace("/", "-").replace("\\", "-").replace(" ", "-")
0981a08… noreply 495 brief = generate_entity_brief(node, rels)
0981a08… noreply 496 path = briefs_dir / f"{safe}.md"
0981a08… noreply 497 path.write_text(brief, encoding="utf-8")
0981a08… noreply 498 created.append(path)
0981a08… noreply 499
0981a08… noreply 500 return created

Keyboard Shortcuts

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