wehub-resource-sync
301 lines
12 KiB
Python
Executable File
301 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Link Suggester for Obsidian Vault
|
|
Identifies potential connections between notes based on content analysis.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from collections import defaultdict, Counter
|
|
import argparse
|
|
import json
|
|
|
|
class LinkSuggester:
|
|
def __init__(self, vault_path):
|
|
self.vault_path = Path(vault_path)
|
|
self.notes = {}
|
|
self.entity_mentions = defaultdict(set)
|
|
self.potential_links = []
|
|
|
|
# Common entities to look for
|
|
self.entities = {
|
|
'technologies': [
|
|
'langchain', 'langgraph', 'mcp', 'model context protocol',
|
|
'graphrag', 'openai', 'anthropic', 'claude', 'gpt', 'llm',
|
|
'ollama', 'huggingface', 'github', 'python', 'javascript',
|
|
'cloudflare', 'supabase', 'vector database', 'embedding',
|
|
'ai agent', 'autonomous agent', 'rag', 'retrieval augmented'
|
|
],
|
|
'concepts': [
|
|
'machine learning', 'deep learning', 'neural network',
|
|
'transformer', 'attention mechanism', 'fine-tuning',
|
|
'prompt engineering', 'chain of thought', 'reasoning',
|
|
'multimodal', 'text generation', 'code generation',
|
|
'tool use', 'function calling', 'api integration'
|
|
],
|
|
'companies': [
|
|
'google', 'microsoft', 'amazon', 'meta', 'apple',
|
|
'nvidia', 'intel', 'amd', 'tesla', 'stripe',
|
|
'y combinator', 'techstars', 'propel', 'dental'
|
|
],
|
|
'people': [
|
|
'andrew ng', 'geoffrey hinton', 'yann lecun', 'ilya sutskever',
|
|
'sam altman', 'dario amodei', 'demis hassabis', 'jensen huang'
|
|
]
|
|
}
|
|
|
|
# Flatten entities for easier searching
|
|
self.all_entities = []
|
|
for category, entities in self.entities.items():
|
|
self.all_entities.extend(entities)
|
|
|
|
def load_notes(self):
|
|
"""Load all markdown files and their content."""
|
|
skip_dirs = {'.obsidian', '.trash', 'System_Files', '.git'}
|
|
|
|
for file_path in self.vault_path.rglob('*.md'):
|
|
if any(skip_dir in file_path.parts for skip_dir in skip_dirs):
|
|
continue
|
|
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# Extract title
|
|
title_match = re.search(r'^#\s+(.+)$', content, re.MULTILINE)
|
|
title = title_match.group(1) if title_match else file_path.stem
|
|
|
|
# Extract existing links
|
|
existing_links = set(re.findall(r'\[\[([^\]]+)\]\]', content))
|
|
|
|
self.notes[file_path] = {
|
|
'title': title,
|
|
'content': content.lower(),
|
|
'existing_links': existing_links,
|
|
'word_count': len(content.split())
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error reading {file_path}: {e}")
|
|
|
|
def find_entity_mentions(self):
|
|
"""Find mentions of entities across all notes."""
|
|
for file_path, note_data in self.notes.items():
|
|
content = note_data['content']
|
|
|
|
for entity in self.all_entities:
|
|
if entity in content:
|
|
self.entity_mentions[entity].add(file_path)
|
|
|
|
def suggest_links_by_entities(self):
|
|
"""Suggest links based on common entity mentions."""
|
|
suggestions = []
|
|
|
|
for entity, files in self.entity_mentions.items():
|
|
if len(files) >= 2: # Entity mentioned in at least 2 files
|
|
file_list = list(files)
|
|
|
|
for i, file1 in enumerate(file_list):
|
|
for file2 in file_list[i+1:]:
|
|
# Check if files don't already link to each other
|
|
note1 = self.notes[file1]
|
|
note2 = self.notes[file2]
|
|
|
|
if (note2['title'] not in note1['existing_links'] and
|
|
note1['title'] not in note2['existing_links']):
|
|
|
|
suggestions.append({
|
|
'file1': file1,
|
|
'file2': file2,
|
|
'title1': note1['title'],
|
|
'title2': note2['title'],
|
|
'common_entity': entity,
|
|
'type': 'entity_mention',
|
|
'confidence': len(files) / 10 # Simple confidence score
|
|
})
|
|
|
|
return suggestions
|
|
|
|
def suggest_links_by_keywords(self):
|
|
"""Suggest links based on keyword overlap."""
|
|
suggestions = []
|
|
|
|
# Extract keywords from titles and content
|
|
for file_path, note_data in self.notes.items():
|
|
if note_data['word_count'] < 100: # Skip very short notes
|
|
continue
|
|
|
|
# Get keywords from title
|
|
title_words = set(re.findall(r'\b\w{4,}\b', note_data['title'].lower()))
|
|
|
|
# Find other notes with similar keywords
|
|
for other_path, other_data in self.notes.items():
|
|
if file_path == other_path:
|
|
continue
|
|
|
|
other_title_words = set(re.findall(r'\b\w{4,}\b', other_data['title'].lower()))
|
|
|
|
# Check for keyword overlap
|
|
common_words = title_words.intersection(other_title_words)
|
|
if len(common_words) >= 2: # At least 2 common significant words
|
|
|
|
# Check if files don't already link to each other
|
|
if (other_data['title'] not in note_data['existing_links'] and
|
|
note_data['title'] not in other_data['existing_links']):
|
|
|
|
suggestions.append({
|
|
'file1': file_path,
|
|
'file2': other_path,
|
|
'title1': note_data['title'],
|
|
'title2': other_data['title'],
|
|
'common_words': list(common_words),
|
|
'type': 'keyword_overlap',
|
|
'confidence': len(common_words) / 5
|
|
})
|
|
|
|
return suggestions
|
|
|
|
def find_orphaned_notes(self):
|
|
"""Find notes with no incoming or outgoing links."""
|
|
orphaned = []
|
|
|
|
for file_path, note_data in self.notes.items():
|
|
if len(note_data['existing_links']) == 0:
|
|
# Check if any other notes link to this one
|
|
mentioned_in = []
|
|
for other_path, other_data in self.notes.items():
|
|
if note_data['title'] in other_data['existing_links']:
|
|
mentioned_in.append(other_path)
|
|
|
|
if not mentioned_in:
|
|
orphaned.append({
|
|
'file': file_path,
|
|
'title': note_data['title'],
|
|
'word_count': note_data['word_count']
|
|
})
|
|
|
|
return orphaned
|
|
|
|
def analyze_vault(self):
|
|
"""Perform complete analysis of the vault."""
|
|
print("Loading notes...")
|
|
self.load_notes()
|
|
print(f"Loaded {len(self.notes)} notes")
|
|
|
|
print("Finding entity mentions...")
|
|
self.find_entity_mentions()
|
|
|
|
print("Generating link suggestions...")
|
|
entity_suggestions = self.suggest_links_by_entities()
|
|
keyword_suggestions = self.suggest_links_by_keywords()
|
|
orphaned_notes = self.find_orphaned_notes()
|
|
|
|
return {
|
|
'entity_suggestions': entity_suggestions,
|
|
'keyword_suggestions': keyword_suggestions,
|
|
'orphaned_notes': orphaned_notes,
|
|
'stats': {
|
|
'total_notes': len(self.notes),
|
|
'entity_suggestions': len(entity_suggestions),
|
|
'keyword_suggestions': len(keyword_suggestions),
|
|
'orphaned_notes': len(orphaned_notes)
|
|
}
|
|
}
|
|
|
|
def generate_report(self, results, output_file=None):
|
|
"""Generate a human-readable report."""
|
|
report = []
|
|
|
|
report.append("# Link Suggestions Report")
|
|
report.append(f"Generated for vault: {self.vault_path}")
|
|
report.append(f"Total notes analyzed: {results['stats']['total_notes']}")
|
|
report.append("")
|
|
|
|
# Entity-based suggestions
|
|
report.append("## Entity-Based Link Suggestions")
|
|
report.append(f"Found {len(results['entity_suggestions'])} potential connections")
|
|
report.append("")
|
|
|
|
# Group by entity
|
|
entity_groups = defaultdict(list)
|
|
for suggestion in results['entity_suggestions']:
|
|
entity_groups[suggestion['common_entity']].append(suggestion)
|
|
|
|
for entity, suggestions in sorted(entity_groups.items()):
|
|
report.append(f"### {entity.title()}")
|
|
for suggestion in suggestions[:5]: # Top 5 per entity
|
|
report.append(f"- [[{suggestion['title1']}]] ↔ [[{suggestion['title2']}]]")
|
|
report.append("")
|
|
|
|
# Keyword-based suggestions
|
|
report.append("## Keyword-Based Link Suggestions")
|
|
report.append(f"Found {len(results['keyword_suggestions'])} potential connections")
|
|
report.append("")
|
|
|
|
# Sort by confidence
|
|
sorted_keywords = sorted(results['keyword_suggestions'],
|
|
key=lambda x: x['confidence'], reverse=True)
|
|
|
|
for suggestion in sorted_keywords[:20]: # Top 20
|
|
common_words = ', '.join(suggestion['common_words'])
|
|
report.append(f"- [[{suggestion['title1']}]] ↔ [[{suggestion['title2']}]]")
|
|
report.append(f" Common words: {common_words}")
|
|
report.append("")
|
|
|
|
# Orphaned notes
|
|
report.append("## Orphaned Notes (No Links)")
|
|
report.append(f"Found {len(results['orphaned_notes'])} notes with no connections")
|
|
report.append("")
|
|
|
|
# Sort by word count (longer notes first)
|
|
sorted_orphaned = sorted(results['orphaned_notes'],
|
|
key=lambda x: x['word_count'], reverse=True)
|
|
|
|
for note in sorted_orphaned[:30]: # Top 30
|
|
report.append(f"- [[{note['title']}]] ({note['word_count']} words)")
|
|
|
|
report_text = '\n'.join(report)
|
|
|
|
if output_file:
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
f.write(report_text)
|
|
print(f"Report saved to: {output_file}")
|
|
|
|
return report_text
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Suggest links for Obsidian vault')
|
|
parser.add_argument('--vault', default='/Users/cam/VAULT01',
|
|
help='Path to Obsidian vault')
|
|
parser.add_argument('--output',
|
|
default='/Users/cam/VAULT01/System_Files/Link_Suggestions_Report.md',
|
|
help='Output file for report')
|
|
parser.add_argument('--json',
|
|
help='Output JSON file for programmatic use')
|
|
|
|
args = parser.parse_args()
|
|
|
|
suggester = LinkSuggester(args.vault)
|
|
results = suggester.analyze_vault()
|
|
|
|
# Generate report
|
|
report = suggester.generate_report(results, args.output)
|
|
|
|
# Save JSON if requested
|
|
if args.json:
|
|
with open(args.json, 'w') as f:
|
|
json.dump(results, f, indent=2, default=str)
|
|
print(f"JSON data saved to: {args.json}")
|
|
|
|
# Print summary
|
|
print("\n" + "="*50)
|
|
print("LINK SUGGESTIONS SUMMARY")
|
|
print("="*50)
|
|
print(f"Total notes: {results['stats']['total_notes']}")
|
|
print(f"Entity-based suggestions: {results['stats']['entity_suggestions']}")
|
|
print(f"Keyword-based suggestions: {results['stats']['keyword_suggestions']}")
|
|
print(f"Orphaned notes: {results['stats']['orphaned_notes']}")
|
|
|
|
if __name__ == '__main__':
|
|
main() |