#!/usr/bin/env python3
"""
Skill library → Brain index pipeline.
Ingests all skills with name, description, category, and frontmatter metadata.
Lets semantic search find relevant procedures: "how do I fix WeasyPrint page breaks?"
"""
import json
import sys
import re
from pathlib import Path
BRAIN_DIR = Path.home() / '.hermes' / 'brain'
sys.path.insert(0, str(BRAIN_DIR))
from brain import init_brain, log_event, query_events
def parse_skill_md(skill_path):
"""Parse SKILL.md frontmatter and description."""
with open(skill_path) as f:
content = f.read()
result = {
'name': skill_path.stem,
'path': str(skill_path),
'title': '',
'description': '',
'tags': [],
'category': '',
'has_linked_files': False,
}
# Extract YAML frontmatter
if content.startswith('---'):
end = content.index('---', 3)
frontmatter = content[3:end].strip()
# Simple YAML parsing
for line in frontmatter.split('\n'):
line = line.strip()
if line.startswith('name:'):
result['name'] = line[5:].strip().strip('"').strip("'")
elif line.startswith('description:'):
result['description'] = line[12:].strip().strip('"').strip("'")
elif line.startswith('tags:'):
tags_str = line[5:].strip()
if tags_str.startswith('[') and tags_str.endswith(']'):
tags_str = tags_str[1:-1]
result['tags'] = [t.strip().strip('"').strip("'") for t in tags_str.split(',') if t.strip()]
elif line.startswith('category:'):
result['category'] = line[9:].strip().strip('"').strip("'")
# Extract title from first heading if no description
if not result['description']:
heading = re.search(r'^#\s+(.+)', content, re.MULTILINE)
if heading:
result['title'] = heading.group(1).strip()
# Check for linked files section
result['has_linked_files'] = bool(re.search(r'linked_files|references/|templates/|scripts/', content))
return result
def purge_existing():
"""Remove previously ingested skill events."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='skills_index', limit=10000)
if not events:
print(" No existing skill events to purge")
return 0
brain_dir = Path.home() / '.hermes' / 'brain'
db_path = brain_dir / 'index' / 'events.db'
event_ids = [e['id'] for e in events]
try:
import sqlite3
db = sqlite3.connect(str(db_path))
batch_size = 500
for i in range(0, len(event_ids), batch_size):
batch = event_ids[i:i+batch_size]
placeholders = ','.join(['?'] * len(batch))
db.execute(f"DELETE FROM events WHERE id IN ({placeholders})", batch)
db.commit()
db.close()
except Exception as e:
print(f" Warning: purge error: {e}")
return len(event_ids)
def ingest_skills():
"""Parse and ingest all skills."""
init_brain()
skills_base = Path.home() / '.hermes' / 'skills'
print("Purging existing skill events...")
purged = purge_existing()
if purged:
print(f" Purged {purged} events")
# Collect all skills
skills = []
for item in sorted(skills_base.iterdir()):
if not item.is_dir():
continue
# Skip hidden/metadata dirs
if item.name.startswith('.'):
continue
# Look for SKILL.md or any .md files
skill_files = list(item.glob('*.md'))
if not skill_files:
continue
category = item.name
for sf in skill_files:
if sf.name == 'SKILL.md':
data = parse_skill_md(sf)
data['category'] = category
skills.append(data)
print(f"\nIngesting {len(skills)} skills...")
count = 0
for skill in skills:
name = skill['name']
category = skill['category']
description = skill['description'] or ''
title = skill.get('title', '')
tags_list = skill.get('tags', [])
# Build tags
tags = [
'skill',
'procedure',
category,
]
tags.extend(tags_list)
if skill.get('has_linked_files'):
tags.append('has_templates')
# Build searchable context
context = {
'type': 'skill',
'name': name,
'category': category,
'description': description,
'title': title,
'path': skill['path'],
'has_linked_files': skill.get('has_linked_files', False),
'tags': tags_list,
}
log_event(
event_type='knowledge_ingest',
tool='skills_index',
args={'name': name, 'category': category},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
if count % 50 == 0 or count == len(skills):
print(f" [{count}/{len(skills)}]")
# Ingest category structure
categories = {}
for skill in skills:
cat = skill['category']
categories[cat] = categories.get(cat, 0) + 1
print(f"\nIngesting {len(categories)} skill categories...")
for cat, cat_count in sorted(categories.items()):
log_event(
event_type='knowledge_ingest',
tool='skills_index',
args={'category': cat, 'skill_count': cat_count},
context={
'type': 'skill_category',
'category': cat,
'skill_count': cat_count,
},
tags=['skill', 'category', cat],
skip_synapse=True,
)
return count
def test_queries():
"""Verify skill index with semantic queries."""
init_brain()
queries = [
("WeasyPrint page break", "weasyprint-page-break-debugging or similar"),
("freqtrade strategy", "freqtrade-strategy-development"),
("handwriting practice PDF", "handwriting-practice-generator"),
("WordPress migration", "wordpress-migration"),
("memory v2 maintenance", "memory-v2-maintenance"),
("brain system", "brain-system"),
("cron job setup", "cron-related skills"),
("seedvault product SEO", "seedvault-product-seo-api"),
]
print("\n=== TEST QUERIES ===")
for query, expected in queries:
results = query_events(
event_type='knowledge_ingest',
tool='skills_index',
full_text_search=query,
limit=3,
)
status = "✓" if results else "✗"
print(f" {status} '{query}' → {len(results)} results (expecting: {expected})")
if results:
for r in results[:2]:
ctx = r.get('context', {})
name = ctx.get('name', '?')
desc = (ctx.get('description') or '')[:60]
print(f" - {name}: {desc}")
if __name__ == '__main__':
print("=" * 60)
print("Skill Library → Brain Index Pipeline")
print("=" * 60)
count = ingest_skills()
print(f"\nTotal ingested: {count}")
test_queries()
print("\nDone.")