#!/usr/bin/env python3 """ skill-kanten-generator.py Generiert Kanten für den Skills-Journey-Graphen basierend auf: - Kanal A: Manuelle Kanten (aus Memory) - Kanal B: SKILL.md-Metadaten (gleiche Kategorie, related_skills) - Kanal C: Co-Nutzung (Session-Analyse, TODO: noch nicht implementiert) Ergebnis: journey-graph.json im Werkstatt-Skills-Verzeichnis. """ from __future__ import annotations import json import os import re from collections import defaultdict from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Set, Tuple # Pfade HERMES_WORKSPACE = Path.home() / ".hermes" / "profiles" / "werkstatt" SKILLS_DIR = HERMES_WORKSPACE / "skills" USAGE_FILE = SKILLS_DIR / ".usage.json" MEMORY_FILE = HERMES_WORKSPACE / "memories" / "MEMORY.md" JOURNEY_GRAPH_FILE = SKILLS_DIR / "journey-graph.json" # Thresholds für Kanal C (Co-Nutzung) MIN_CO_USE = 2 MIN_USE_COUNT = 3 JACCARD_THRESHOLD = 0.3 # Gewichte für Priorisierung WEIGHT_MANUAL = 1.0 WEIGHT_RELATED_DECLARED = 0.9 WEIGHT_CATEGORY_MATCH = 0.7 WEIGHT_TAG_OVERLAP = 0.5 @dataclass class Edge: """Repräsentiert eine Kante im Journey-Graphen.""" skill_a: str skill_b: str weight: float edge_type: str source: str def key(self) -> Tuple[str, str]: return (self.skill_a, self.skill_b) if self.skill_a < self.skill_b else (self.skill_b, self.skill_a) def to_dict(self) -> Dict: return { "skill_a": self.skill_a, "skill_b": self.skill_b, "weight": self.weight, "type": self.edge_type, "source": self.source, } @dataclass class SkillInfo: """Zusammengeführte Informationen zu einem Skill.""" name: str category: Optional[str] = None related_skills: List[str] = field(default_factory=list) tags: List[str] = field(default_factory=list) use_count: int = 0 state: str = "active" def load_manual_edges_from_memory() -> List[Edge]: """Liest manuelle Kanten aus dem MEMORY.md-Eintrag 'Skills-Kanten (Journey-Graph)'.""" edges: List[Edge] = [] if not MEMORY_FILE.exists(): print(f"Warnung: MEMORY.md nicht gefunden unter {MEMORY_FILE}") return edges content = MEMORY_FILE.read_text(encoding="utf-8") # Suche nach dem Eintrag "Skills-Kanten (Journey-Graph)" pattern = r"Skills-Kanten \(Journey-Graph\):\s*(.+?)(?=\n\n|\Z)" match = re.search(pattern, content, re.DOTALL) if not match: print("Keine manuellen Skills-Kanten im MEMORY.md gefunden.") return edges kanten_text = match.group(1).strip() # Parso die Kanten: "skill-a ↔ skill-b (Begründung)" # Trennzeichen ist " | " für mehrere Kanten kanten_liste = re.split(r"\s*\|\s*", kanten_text) for kante in kanten_liste: kante = kante.strip() if not kante: continue # Format: skill-a ↔ skill-b (Begründung) parts = kante.split("↔") if len(parts) != 2: print(f"Warnung: Ungültiges Kanten-Format: {kante}") continue skill_a = parts[0].strip() skill_b_and_reason = parts[1].strip() # Trenne Skill-B und Begründung (in Klammern) reason_match = re.match(r"(.+?)\s*\((.+?)\)\s*$", skill_b_and_reason) if reason_match: skill_b = reason_match.group(1).strip() # reason = reason_match.group(2).strip() # nicht verwendet, aber verfügbar else: skill_b = skill_b_and_reason print(f"Warnung: Keine Begründung in Kante: {kante}") edges.append(Edge( skill_a=skill_a, skill_b=skill_b, weight=WEIGHT_MANUAL, edge_type="manual", source="memory" )) print(f"Manuelle Kanten geladen: {len(edges)}") return edges def build_skill_index() -> Dict[str, SkillInfo]: """Erstellt einen Index aller Skills aus .usage.json und SKILL.md-Dateien.""" skills: Dict[str, SkillInfo] = {} # 1. Lade .usage.json für use_count und state if not USAGE_FILE.exists(): print(f"Warnung: .usage.json nicht gefunden unter {USAGE_FILE}") return skills usage_data = json.loads(USAGE_FILE.read_text(encoding="utf-8")) # Extrahiere Skill-Namen aus keys (es gibt auch "Mission Control Model Management" etc.) for skill_name, info in usage_data.items(): # Normalisiere Namen (ersetze Leerzeichen, Sonderzeichen für Verzeichnisnamen) normalized_name = skill_name.lower().replace(" ", "-").replace("_", "-") skill = SkillInfo( name=skill_name, use_count=info.get("use_count", 0), state=info.get("state", "active") ) skills[normalized_name] = skill # 2. Lade SKILL.md Metadaten für Kategorien, Tags, related_skills # Suche in allen Kategorien-Verzeichnissen for category_dir in SKILLS_DIR.iterdir(): if not category_dir.is_dir() or category_dir.name.startswith("."): continue for skill_dir in category_dir.iterdir(): if not skill_dir.is_dir(): continue skill_md = skill_dir / "SKILL.md" if not skill_md.exists(): continue content = skill_md.read_text(encoding="utf-8") # YAML-Frontmatter extrahieren (alles vor der ersten leeren Zeile nach ---) frontmatter_match = re.match(r"---\s*(.+?)\s*---", content, re.DOTALL) if not frontmatter_match: continue frontmatter = frontmatter_match.group(1) # Parse YAML-Felder (einfache Regex-basierte Parsers) category_match = re.search(r"category:\s*(['\"]?)(\w+)\1", frontmatter) tags_match = re.search(r"tags:\s*\[(.*?)\]", frontmatter, re.DOTALL) related_match = re.search(r"related_skills:\s*\[(.*?)\]", frontmatter, re.DOTALL) # Extrahiere den Skill-Namen aus dem Frontmatter name_match = re.search(r"name:\s*['\"]?([^\n'\"]+)['\"]?", frontmatter) if name_match: skill_name = name_match.group(1).strip().lower().replace(" ", "-") if skill_name in skills: if category_match: skills[skill_name].category = category_match.group(2).lower() if tags_match: tags_str = tags_match.group(1) tags = [t.strip().strip("'\"") for t in tags_str.split(",") if t.strip()] skills[skill_name].tags.extend(tags) if related_match: related_str = related_match.group(1) related = [r.strip().strip("'\"") for r in related_str.split(",") if r.strip()] skills[skill_name].related_skills.extend(related) print(f"Skill-Index erstellt: {len(skills)} Skills") return skills def generate_metadata_edges(skills: Dict[str, SkillInfo]) -> List[Edge]: """Generiert Kanten aus SKILL.md-Metadaten (Kategorie, related_skills, Tags).""" edges: List[Edge] = [] skill_list = list(skills.values()) # Kategorien und Tags für schnellen Zugriff category_map: Dict[str, List[str]] = defaultdict(list) tag_map: Dict[str, List[str]] = defaultdict(list) for skill in skill_list: if skill.category: category_map[skill.category].append(skill.name) for tag in skill.tags: tag_map[tag].append(skill.name) # 1. Kategorie-basierte Kanten for category, skill_names in category_map.items(): if len(skill_names) < 2: continue for i, skill_a in enumerate(skill_names): for skill_b in skill_names[i + 1:]: if skill_a != skill_b: edges.append(Edge( skill_a=skill_a, skill_b=skill_b, weight=WEIGHT_CATEGORY_MATCH, edge_type="category", source="category_match" )) # 2. Related skills Kanten (gerichtet, aber als ungerichtete Kante gespeichert) for skill in skill_list: for related in skill.related_skills: if related in skills: edges.append(Edge( skill_a=skill.name, skill_b=related, weight=WEIGHT_RELATED_DECLARED, edge_type="declared", source="related_declared" )) # 3. Tag-Überlappung (mindestens 2 gemeinsame Tags) for skill_a in skill_list: for skill_b in skill_list: if skill_a.name >= skill_b.name: # Vermeide Doppelt-Generierung continue common_tags = set(skill_a.tags) & set(skill_b.tags) if len(common_tags) >= 2: edges.append(Edge( skill_a=skill_a.name, skill_b=skill_b.name, weight=WEIGHT_TAG_OVERLAP, edge_type="tag_overlap", source="tag_overlap" )) print(f"Metadaten-Kanten generiert: {len(edges)}") return edges def generate_co_use_edges(skills: Dict[str, SkillInfo]) -> List[Edge]: """ Generiert Kanten basierend auf Co-Nutzungsmustern aus Sessions. HINWEIS: Aktuell nicht implementiert, da Sessions keine Skill-IDs speichern. Gibt eine leere Liste zurück. """ print("Co-Nutzung-Analyse: SKIPPED (Sessions speichern keine Skill-IDs)") return [] def deduplicate_edges(edges: List[Edge]) -> List[Edge]: """ Entfernt doppelte Kanten zwischen denselben Skills. Wenn mehrere Kanten zwischen selbem Paar existieren, behält die mit höchstem Gewicht. PRIORITY: manual > declared > category > co_use """ # Gruppiere nach Schlüssel (ungereichtes Paar) edge_groups: Dict[Tuple[str, str], List[Edge]] = defaultdict(list) for edge in edges: edge_groups[edge.key()].append(edge) # Wähle die beste Kante pro Gruppe result: List[Edge] = [] for key, group in edge_groups.items(): # Sortiere nach Gewicht (höchste zuerst), dann nach Typ-Priorität type_priority = {"manual": 4, "declared": 3, "category": 2, "tag_overlap": 1, "co_use": 0} group.sort(key=lambda e: (e.weight, type_priority.get(e.edge_type, -1)), reverse=True) result.append(group[0]) print(f"Nach Deduplizierung: {len(result)} eindeutige Kanten") return result def generate_journey_graph() -> Dict: """Generiert den Journey-Graphen mit Knoten und Kanten.""" # 1. Manuelle Kanten laden manual_edges = load_manual_edges_from_memory() # 2. Skill-Index bauen skills = build_skill_index() # 3. Metadaten-Kanten generieren metadata_edges = generate_metadata_edges(skills) # 4. Co-Nutzung (zurzeit leer) co_use_edges = generate_co_use_edges(skills) # 5. Alle Kanten zusammenfassen all_edges = manual_edges + metadata_edges + co_use_edges # 6. Deduplizieren final_edges = deduplicate_edges(all_edges) # 7. Graph strukturieren graph = { "generated_at": None, # Wird unten gesetzt "version": "1.0.0", "nodes": [ { "id": skill.name, "label": skill.name.replace("-", " ").title(), "category": skill.category, "state": skill.state, } for skill in skills.values() if skill.state == "active" ], "edges": [edge.to_dict() for edge in final_edges], } # 8. Timestamp hinzufügen from datetime import datetime, timezone graph["generated_at"] = datetime.now(timezone.utc).isoformat() return graph def main() -> int: """Hauptfunktion.""" print("=== Skill-Kanten-Generator ===") print() # Graph generieren graph = generate_journey_graph() # Ausgabe print() print(f"Knoten: {len(graph['nodes'])}") print(f"Kanten: {len(graph['edges'])}") # Protokoll edge_types = defaultdict(int) for edge in graph["edges"]: edge_types[edge["type"]] += 1 print() print("Kanten nach Typ:") for t, count in sorted(edge_types.items(), key=lambda x: -x[1]): print(f" {t}: {count}") # Speichern JOURNEY_GRAPH_FILE.write_text( json.dumps(graph, indent=2, ensure_ascii=False), encoding="utf-8" ) print() print(f"Graph gespeichert unter: {JOURNEY_GRAPH_FILE}") return 0 if __name__ == "__main__": import sys sys.exit(main())