PlanOpticon

planopticon / video_processor / agent / skills / wiki_generator.py
Source Blame History 315 lines
0981a08… noreply 1 """Skill: Generate a GitHub wiki from knowledge graph and artifacts."""
0981a08… noreply 2
0981a08… noreply 3 import json
0981a08… noreply 4 import logging
0981a08… noreply 5 import subprocess
0981a08… noreply 6 from pathlib import Path
0981a08… noreply 7 from typing import Dict, List, Optional
0981a08… noreply 8
0981a08… noreply 9 from video_processor.agent.skills.base import (
0981a08… noreply 10 AgentContext,
0981a08… noreply 11 Artifact,
0981a08… noreply 12 Skill,
0981a08… noreply 13 register_skill,
0981a08… noreply 14 )
0981a08… noreply 15
0981a08… noreply 16 logger = logging.getLogger(__name__)
0981a08… noreply 17
0981a08… noreply 18
0981a08… noreply 19 def _sanitize_filename(name: str) -> str:
0981a08… noreply 20 """Convert entity name to a wiki-safe filename."""
0981a08… noreply 21 return name.replace("/", "-").replace("\\", "-").replace(" ", "-").replace(".", "-")
0981a08… noreply 22
0981a08… noreply 23
0981a08… noreply 24 def _wiki_link(name: str) -> str:
0981a08… noreply 25 """Create a GitHub wiki-style markdown link."""
0981a08… noreply 26 safe = _sanitize_filename(name)
0981a08… noreply 27 return f"[{name}]({safe})"
0981a08… noreply 28
0981a08… noreply 29
0981a08… noreply 30 def generate_wiki(
0981a08… noreply 31 kg_data: dict,
0981a08… noreply 32 artifacts: Optional[List[Artifact]] = None,
0981a08… noreply 33 title: str = "Knowledge Base",
0981a08… noreply 34 ) -> Dict[str, str]:
0981a08… noreply 35 """Generate a dict of {filename: markdown_content} for a GitHub wiki.
0981a08… noreply 36
0981a08… noreply 37 Returns pages for: Home, _Sidebar, entity type indexes, individual
0981a08… noreply 38 entity pages, and any planning artifacts.
0981a08… noreply 39 """
0981a08… noreply 40 pages: Dict[str, str] = {}
0981a08… noreply 41 artifacts = artifacts or []
0981a08… noreply 42
0981a08… noreply 43 nodes = kg_data.get("nodes", [])
0981a08… noreply 44 relationships = kg_data.get("relationships", [])
0981a08… noreply 45
0981a08… noreply 46 # Group entities by type
0981a08… noreply 47 by_type: Dict[str, list] = {}
0981a08… noreply 48 node_lookup: Dict[str, dict] = {}
0981a08… noreply 49 for node in nodes:
0981a08… noreply 50 name = node.get("name", node.get("id", ""))
0981a08… noreply 51 ntype = node.get("type", "concept")
0981a08… noreply 52 by_type.setdefault(ntype, []).append(node)
0981a08… noreply 53 node_lookup[name.lower()] = node
0981a08… noreply 54
0981a08… noreply 55 # Build relationship index (outgoing and incoming per entity)
0981a08… noreply 56 outgoing: Dict[str, list] = {}
0981a08… noreply 57 incoming: Dict[str, list] = {}
0981a08… noreply 58 for rel in relationships:
0981a08… noreply 59 src = rel.get("source", "")
0981a08… noreply 60 tgt = rel.get("target", "")
0981a08… noreply 61 rtype = rel.get("type", "related_to")
0981a08… noreply 62 outgoing.setdefault(src, []).append((tgt, rtype))
0981a08… noreply 63 incoming.setdefault(tgt, []).append((src, rtype))
0981a08… noreply 64
0981a08… noreply 65 # --- Home page ---
0981a08… noreply 66 home_parts = [
0981a08… noreply 67 f"# {title}",
0981a08… noreply 68 "",
0981a08… noreply 69 f"**{len(nodes)}** entities | **{len(relationships)}** relationships",
0981a08… noreply 70 "",
0981a08… noreply 71 "## Entity Types",
0981a08… noreply 72 "",
0981a08… noreply 73 ]
0981a08… noreply 74 for etype, elist in sorted(by_type.items()):
0981a08… noreply 75 home_parts.append(f"- {_wiki_link(etype.title())} ({len(elist)})")
0981a08… noreply 76
0981a08… noreply 77 if artifacts:
0981a08… noreply 78 home_parts.append("")
0981a08… noreply 79 home_parts.append("## Planning Artifacts")
0981a08… noreply 80 home_parts.append("")
0981a08… noreply 81 for art in artifacts:
0981a08… noreply 82 safe = _sanitize_filename(art.name)
0981a08… noreply 83 home_parts.append(f"- [{art.name}]({safe})")
0981a08… noreply 84
0981a08… noreply 85 pages["Home"] = "\n".join(home_parts)
0981a08… noreply 86
0981a08… noreply 87 # --- Sidebar ---
0981a08… noreply 88 sidebar_parts = [f"**{title}**", "", "**Navigation**", "", "- [Home](Home)", ""]
0981a08… noreply 89 sidebar_parts.append("**Entity Types**")
0981a08… noreply 90 sidebar_parts.append("")
0981a08… noreply 91 for etype in sorted(by_type.keys()):
0981a08… noreply 92 sidebar_parts.append(f"- {_wiki_link(etype.title())}")
0981a08… noreply 93
0981a08… noreply 94 if artifacts:
0981a08… noreply 95 sidebar_parts.append("")
0981a08… noreply 96 sidebar_parts.append("**Artifacts**")
0981a08… noreply 97 sidebar_parts.append("")
0981a08… noreply 98 for art in artifacts:
0981a08… noreply 99 safe = _sanitize_filename(art.name)
0981a08… noreply 100 sidebar_parts.append(f"- [{art.name}]({safe})")
0981a08… noreply 101
0981a08… noreply 102 pages["_Sidebar"] = "\n".join(sidebar_parts)
0981a08… noreply 103
0981a08… noreply 104 # --- Type index pages ---
0981a08… noreply 105 for etype, elist in sorted(by_type.items()):
0981a08… noreply 106 page_name = _sanitize_filename(etype.title())
0981a08… noreply 107 parts = [
0981a08… noreply 108 f"# {etype.title()}",
0981a08… noreply 109 "",
0981a08… noreply 110 f"{len(elist)} entities of type **{etype}**.",
0981a08… noreply 111 "",
0981a08… noreply 112 "| Entity | Descriptions |",
0981a08… noreply 113 "|--------|-------------|",
0981a08… noreply 114 ]
0981a08… noreply 115 for node in sorted(elist, key=lambda n: n.get("name", "")):
0981a08… noreply 116 name = node.get("name", "")
0981a08… noreply 117 descs = node.get("descriptions", [])
0981a08… noreply 118 desc_text = "; ".join(descs[:2]) if descs else "—"
0981a08… noreply 119 parts.append(f"| {_wiki_link(name)} | {desc_text} |")
0981a08… noreply 120
0981a08… noreply 121 pages[page_name] = "\n".join(parts)
0981a08… noreply 122
0981a08… noreply 123 # --- Individual entity pages ---
0981a08… noreply 124 for node in nodes:
0981a08… noreply 125 name = node.get("name", "")
0981a08… noreply 126 if not name:
0981a08… noreply 127 continue
0981a08… noreply 128 ntype = node.get("type", "concept")
0981a08… noreply 129 descs = node.get("descriptions", [])
0981a08… noreply 130 page_name = _sanitize_filename(name)
0981a08… noreply 131
0981a08… noreply 132 parts = [
0981a08… noreply 133 f"# {name}",
0981a08… noreply 134 "",
0981a08… noreply 135 f"**Type:** {ntype}",
0981a08… noreply 136 "",
0981a08… noreply 137 ]
0981a08… noreply 138
0981a08… noreply 139 if descs:
0981a08… noreply 140 parts.append("## Descriptions")
0981a08… noreply 141 parts.append("")
0981a08… noreply 142 for d in descs:
0981a08… noreply 143 parts.append(f"- {d}")
0981a08… noreply 144 parts.append("")
0981a08… noreply 145
0981a08… noreply 146 # Outgoing relationships
0981a08… noreply 147 outs = outgoing.get(name, [])
0981a08… noreply 148 if outs:
0981a08… noreply 149 parts.append("## Relationships")
0981a08… noreply 150 parts.append("")
0981a08… noreply 151 parts.append("| Target | Type |")
0981a08… noreply 152 parts.append("|--------|------|")
0981a08… noreply 153 for tgt, rtype in outs:
0981a08… noreply 154 parts.append(f"| {_wiki_link(tgt)} | {rtype} |")
0981a08… noreply 155 parts.append("")
0981a08… noreply 156
0981a08… noreply 157 # Incoming relationships
0981a08… noreply 158 ins = incoming.get(name, [])
0981a08… noreply 159 if ins:
0981a08… noreply 160 parts.append("## Referenced By")
0981a08… noreply 161 parts.append("")
0981a08… noreply 162 parts.append("| Source | Type |")
0981a08… noreply 163 parts.append("|--------|------|")
0981a08… noreply 164 for src, rtype in ins:
0981a08… noreply 165 parts.append(f"| {_wiki_link(src)} | {rtype} |")
0981a08… noreply 166 parts.append("")
0981a08… noreply 167
0981a08… noreply 168 # Occurrences / sources
0981a08… noreply 169 occs = node.get("occurrences", [])
0981a08… noreply 170 if occs:
0981a08… noreply 171 parts.append("## Sources")
0981a08… noreply 172 parts.append("")
0981a08… noreply 173 for occ in occs:
0981a08… noreply 174 src = occ.get("source", "unknown")
0981a08… noreply 175 ts = occ.get("timestamp", "")
0981a08… noreply 176 text = occ.get("text", "")
0981a08… noreply 177 line = f"- **{src}**"
0981a08… noreply 178 if ts:
0981a08… noreply 179 line += f" @ {ts}"
0981a08… noreply 180 if text:
0981a08… noreply 181 line += f": _{text}_"
0981a08… noreply 182 parts.append(line)
0981a08… noreply 183 parts.append("")
0981a08… noreply 184
0981a08… noreply 185 pages[page_name] = "\n".join(parts)
0981a08… noreply 186
0981a08… noreply 187 # --- Artifact pages ---
0981a08… noreply 188 for art in artifacts:
0981a08… noreply 189 page_name = _sanitize_filename(art.name)
0981a08… noreply 190 if art.format == "json":
0981a08… noreply 191 try:
0981a08… noreply 192 data = json.loads(art.content)
0981a08… noreply 193 content = f"```json\n{json.dumps(data, indent=2)}\n```"
0981a08… noreply 194 except json.JSONDecodeError:
0981a08… noreply 195 content = art.content
0981a08… noreply 196 else:
0981a08… noreply 197 content = art.content
0981a08… noreply 198
0981a08… noreply 199 pages[page_name] = f"# {art.name}\n\n{content}"
0981a08… noreply 200
0981a08… noreply 201 return pages
0981a08… noreply 202
0981a08… noreply 203
0981a08… noreply 204 def write_wiki(pages: Dict[str, str], output_dir: Path) -> List[Path]:
0981a08… noreply 205 """Write wiki pages to a directory as .md files."""
0981a08… noreply 206 output_dir.mkdir(parents=True, exist_ok=True)
0981a08… noreply 207 paths = []
0981a08… noreply 208 for name, content in pages.items():
0981a08… noreply 209 path = output_dir / f"{name}.md"
0981a08… noreply 210 path.write_text(content, encoding="utf-8")
0981a08… noreply 211 paths.append(path)
0981a08… noreply 212 return paths
0981a08… noreply 213
0981a08… noreply 214
0981a08… noreply 215 def push_wiki(wiki_dir: Path, repo: str, message: str = "Update wiki") -> bool:
0981a08… noreply 216 """Push wiki pages to a GitHub wiki repo.
0981a08… noreply 217
0981a08… noreply 218 Clones the wiki repo, copies pages, commits and pushes.
0981a08… noreply 219 The repo should be in 'owner/repo' format.
0981a08… noreply 220 """
0981a08… noreply 221 wiki_url = f"https://github.com/{repo}.wiki.git"
0981a08… noreply 222
0981a08… noreply 223 # Clone existing wiki (or init if empty)
0981a08… noreply 224 clone_dir = wiki_dir / ".wiki_clone"
0981a08… noreply 225 if clone_dir.exists():
0981a08… noreply 226 subprocess.run(["rm", "-rf", str(clone_dir)], check=True)
0981a08… noreply 227
0981a08… noreply 228 result = subprocess.run(
0981a08… noreply 229 ["git", "clone", wiki_url, str(clone_dir)],
0981a08… noreply 230 capture_output=True,
0981a08… noreply 231 text=True,
0981a08… noreply 232 )
0981a08… noreply 233
0981a08… noreply 234 if result.returncode != 0:
0981a08… noreply 235 # Wiki might not exist yet — init a new repo
0981a08… noreply 236 clone_dir.mkdir(parents=True, exist_ok=True)
0981a08… noreply 237 subprocess.run(["git", "init"], cwd=clone_dir, capture_output=True)
0981a08… noreply 238 subprocess.run(
0981a08… noreply 239 ["git", "remote", "add", "origin", wiki_url],
0981a08… noreply 240 cwd=clone_dir,
0981a08… noreply 241 capture_output=True,
0981a08… noreply 242 )
0981a08… noreply 243
0981a08… noreply 244 # Copy wiki pages into clone
0981a08… noreply 245 for md_file in wiki_dir.glob("*.md"):
0981a08… noreply 246 if md_file.parent == wiki_dir:
0981a08… noreply 247 dest = clone_dir / md_file.name
0981a08… noreply 248 dest.write_text(md_file.read_text(encoding="utf-8"), encoding="utf-8")
0981a08… noreply 249
0981a08… noreply 250 # Commit and push
0981a08… noreply 251 subprocess.run(["git", "add", "-A"], cwd=clone_dir, capture_output=True)
0981a08… noreply 252 commit_result = subprocess.run(
0981a08… noreply 253 ["git", "commit", "-m", message],
0981a08… noreply 254 cwd=clone_dir,
0981a08… noreply 255 capture_output=True,
0981a08… noreply 256 text=True,
0981a08… noreply 257 )
0981a08… noreply 258 if commit_result.returncode != 0:
0981a08… noreply 259 logger.info("No wiki changes to commit")
0981a08… noreply 260 return True
0981a08… noreply 261
0981a08… noreply 262 push_result = subprocess.run(
0981a08… noreply 263 ["git", "push", "origin", "master"],
0981a08… noreply 264 cwd=clone_dir,
0981a08… noreply 265 capture_output=True,
0981a08… noreply 266 text=True,
0981a08… noreply 267 )
0981a08… noreply 268 if push_result.returncode != 0:
0981a08… noreply 269 # Try main branch
0981a08… noreply 270 push_result = subprocess.run(
0981a08… noreply 271 ["git", "push", "origin", "main"],
0981a08… noreply 272 cwd=clone_dir,
0981a08… noreply 273 capture_output=True,
0981a08… noreply 274 text=True,
0981a08… noreply 275 )
0981a08… noreply 276
0981a08… noreply 277 if push_result.returncode == 0:
0981a08… noreply 278 logger.info(f"Wiki pushed to {wiki_url}")
0981a08… noreply 279 return True
0981a08… noreply 280 else:
0981a08… noreply 281 logger.error(f"Wiki push failed: {push_result.stderr}")
0981a08… noreply 282 return False
0981a08… noreply 283
0981a08… noreply 284
0981a08… noreply 285 class WikiGeneratorSkill(Skill):
0981a08… noreply 286 name = "wiki_generator"
0981a08… noreply 287 description = "Generate a GitHub wiki from knowledge graph and artifacts"
0981a08… noreply 288
0981a08… noreply 289 def execute(self, context: AgentContext, **kwargs) -> Artifact:
0981a08… noreply 290 kg_data = context.knowledge_graph.to_dict()
0981a08… noreply 291 pages = generate_wiki(
0981a08… noreply 292 kg_data,
0981a08… noreply 293 artifacts=context.artifacts,
0981a08… noreply 294 title=kwargs.get("title", "Knowledge Base"),
0981a08… noreply 295 )
0981a08… noreply 296
0981a08… noreply 297 # Return a summary artifact; actual pages are written via write_wiki()
0981a08… noreply 298 page_list = sorted(pages.keys())
0981a08… noreply 299 summary_parts = [
0981a08… noreply 300 f"Generated {len(pages)} wiki pages:",
0981a08… noreply 301 "",
0981a08… noreply 302 ]
0981a08… noreply 303 for name in page_list:
0981a08… noreply 304 summary_parts.append(f"- {name}.md")
0981a08… noreply 305
0981a08… noreply 306 return Artifact(
0981a08… noreply 307 name="Wiki",
0981a08… noreply 308 content="\n".join(summary_parts),
0981a08… noreply 309 artifact_type="wiki",
0981a08… noreply 310 format="markdown",
0981a08… noreply 311 metadata={"pages": pages},
0981a08… noreply 312 )
0981a08… noreply 313
0981a08… noreply 314
0981a08… noreply 315 register_skill(WikiGeneratorSkill())

Keyboard Shortcuts

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