feat(tagging-system): Add comprehensive tagging system implementation
- Add tagging system Python package with modular architecture (config, core, impl) - Create core interfaces for file discovery, content analysis, and tag generation - Implement content analyzer with language detection and topic extraction - Implement file discovery engine with directory scanning and filtering - Implement tag generator with hierarchical tag creation and consolidation - Add configuration management with example config and validation - Create comprehensive design and requirements documentation in .kiro/specs - Add pytest test suite with unit tests for models, interfaces, and config - Add setup.py, requirements.txt, and pytest.ini for package management - Add README.md with project overview and usage instructions - Update workspace.json with new project structure - Add Excalidraw diagram for system architecture visualization - Establish foundation for automated vault tagging and metadata management
This commit is contained in:
windyboy
@@ -0,0 +1,41 @@
|
||||
"""
|
||||
Comprehensive Tagging System for Obsidian Vault
|
||||
|
||||
A Python-based system for analyzing files, generating appropriate tags based on
|
||||
directory structure and content analysis, and updating frontmatter while
|
||||
preserving existing data.
|
||||
"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__author__ = "Tagging System"
|
||||
|
||||
from .core.models import (
|
||||
FileInfo,
|
||||
ContentAnalysis,
|
||||
TagStructure,
|
||||
FrontmatterData,
|
||||
ContentType,
|
||||
LanguageInfo,
|
||||
ValidationResult
|
||||
)
|
||||
|
||||
from .core.interfaces import (
|
||||
FileDiscovery,
|
||||
ContentAnalyzer,
|
||||
TagGenerator,
|
||||
FrontmatterManager
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"FileInfo",
|
||||
"ContentAnalysis",
|
||||
"TagStructure",
|
||||
"FrontmatterData",
|
||||
"ContentType",
|
||||
"LanguageInfo",
|
||||
"ValidationResult",
|
||||
"FileDiscovery",
|
||||
"ContentAnalyzer",
|
||||
"TagGenerator",
|
||||
"FrontmatterManager"
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,73 @@
|
||||
"""Command-line interface for the tagging system."""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .config import load_config
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point for the CLI."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Comprehensive Tagging System for Obsidian Vaults"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"vault_path",
|
||||
type=str,
|
||||
help="Path to the Obsidian vault directory"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
type=str,
|
||||
help="Path to configuration file (YAML or JSON)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Show what would be done without making changes"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
"-v",
|
||||
action="store_true",
|
||||
help="Enable verbose output"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Validate vault path
|
||||
vault_path = Path(args.vault_path)
|
||||
if not vault_path.exists():
|
||||
print(f"Error: Vault path '{vault_path}' does not exist")
|
||||
sys.exit(1)
|
||||
|
||||
if not vault_path.is_dir():
|
||||
print(f"Error: Vault path '{vault_path}' is not a directory")
|
||||
sys.exit(1)
|
||||
|
||||
# Load configuration
|
||||
config = load_config(args.config)
|
||||
|
||||
if args.verbose:
|
||||
print(f"Vault path: {vault_path}")
|
||||
print(f"Configuration: {args.config or 'default'}")
|
||||
print(f"Dry run: {args.dry_run}")
|
||||
|
||||
# TODO: Implement actual tagging logic in future tasks
|
||||
print("Tagging system setup complete!")
|
||||
print("Note: Core implementation will be added in subsequent tasks.")
|
||||
|
||||
if args.dry_run:
|
||||
print("Dry run mode - no files would be modified")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Configuration system for the tagging system."""
|
||||
|
||||
from .config import TaggingConfig, DirectoryMapping, TagHierarchy, SensitivePatterns, load_config, save_config
|
||||
|
||||
__all__ = [
|
||||
"TaggingConfig",
|
||||
"DirectoryMapping",
|
||||
"TagHierarchy",
|
||||
"SensitivePatterns",
|
||||
"load_config",
|
||||
"save_config"
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,223 @@
|
||||
"""Configuration system for tag hierarchies and rules."""
|
||||
|
||||
import json
|
||||
import yaml
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Any, Union
|
||||
|
||||
|
||||
@dataclass
|
||||
class DirectoryMapping:
|
||||
"""Configuration for directory-based tag mapping."""
|
||||
pattern: str # Directory pattern to match
|
||||
primary_tag: str # Primary tag to assign
|
||||
hierarchical_tags: List[str] = field(default_factory=list) # Additional hierarchical tags
|
||||
exclude_patterns: List[str] = field(default_factory=list) # Patterns to exclude
|
||||
|
||||
|
||||
@dataclass
|
||||
class TagHierarchy:
|
||||
"""Configuration for hierarchical tag structures."""
|
||||
root: str # Root tag name
|
||||
children: Dict[str, 'TagHierarchy'] = field(default_factory=dict) # Child hierarchies
|
||||
aliases: List[str] = field(default_factory=list) # Alternative names
|
||||
|
||||
def get_full_path(self, child_path: str = "") -> str:
|
||||
"""Get full hierarchical path."""
|
||||
if child_path:
|
||||
return f"{self.root}/{child_path}"
|
||||
return self.root
|
||||
|
||||
|
||||
@dataclass
|
||||
class SensitivePatterns:
|
||||
"""Configuration for sensitive content detection."""
|
||||
credential_patterns: List[str] = field(default_factory=lambda: [
|
||||
r'api[_-]?key',
|
||||
r'secret[_-]?key',
|
||||
r'password',
|
||||
r'token',
|
||||
r'auth[_-]?token'
|
||||
])
|
||||
personal_patterns: List[str] = field(default_factory=lambda: [
|
||||
r'\b\d{3}-\d{2}-\d{4}\b', # SSN pattern
|
||||
r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', # Credit card pattern
|
||||
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' # Email pattern
|
||||
])
|
||||
financial_patterns: List[str] = field(default_factory=lambda: [
|
||||
r'bank[_-]?account',
|
||||
r'routing[_-]?number',
|
||||
r'credit[_-]?card',
|
||||
r'social[_-]?security'
|
||||
])
|
||||
|
||||
|
||||
@dataclass
|
||||
class TaggingConfig:
|
||||
"""Main configuration for the tagging system."""
|
||||
|
||||
# Directory mappings
|
||||
directory_mappings: List[DirectoryMapping] = field(default_factory=lambda: [
|
||||
DirectoryMapping("100-project", "project"),
|
||||
DirectoryMapping("200-area", "area"),
|
||||
DirectoryMapping("300-resources", "resource"),
|
||||
DirectoryMapping("400-archive", "archive"),
|
||||
DirectoryMapping("Clippings", "clipping"),
|
||||
DirectoryMapping("ReadItLater Inbox", "clipping"),
|
||||
DirectoryMapping("000-inbox", "inbox")
|
||||
])
|
||||
|
||||
# Tag hierarchies
|
||||
tag_hierarchies: Dict[str, TagHierarchy] = field(default_factory=lambda: {
|
||||
"tech": TagHierarchy("tech", {
|
||||
"ai": TagHierarchy("ai", {
|
||||
"llm": TagHierarchy("llm"),
|
||||
"ml": TagHierarchy("ml"),
|
||||
"nlp": TagHierarchy("nlp")
|
||||
}),
|
||||
"infrastructure": TagHierarchy("infrastructure", {
|
||||
"docker": TagHierarchy("docker"),
|
||||
"kubernetes": TagHierarchy("kubernetes"),
|
||||
"cloud": TagHierarchy("cloud")
|
||||
}),
|
||||
"development": TagHierarchy("development", {
|
||||
"python": TagHierarchy("python"),
|
||||
"javascript": TagHierarchy("javascript"),
|
||||
"typescript": TagHierarchy("typescript")
|
||||
})
|
||||
}),
|
||||
"personal": TagHierarchy("personal", {
|
||||
"productivity": TagHierarchy("productivity", {
|
||||
"gtd": TagHierarchy("gtd"),
|
||||
"pkm": TagHierarchy("pkm")
|
||||
}),
|
||||
"health": TagHierarchy("health", {
|
||||
"cycling": TagHierarchy("cycling"),
|
||||
"fitness": TagHierarchy("fitness")
|
||||
}),
|
||||
"finance": TagHierarchy("finance")
|
||||
}),
|
||||
"work": TagHierarchy("work", {
|
||||
"government": TagHierarchy("government"),
|
||||
"enterprise": TagHierarchy("enterprise"),
|
||||
"consulting": TagHierarchy("consulting")
|
||||
})
|
||||
})
|
||||
|
||||
# Sensitive content patterns
|
||||
sensitive_patterns: SensitivePatterns = field(default_factory=SensitivePatterns)
|
||||
|
||||
# File processing settings
|
||||
excluded_directories: List[str] = field(default_factory=lambda: [
|
||||
".obsidian",
|
||||
".git",
|
||||
".smart-env",
|
||||
"node_modules",
|
||||
"__pycache__"
|
||||
])
|
||||
|
||||
excluded_file_patterns: List[str] = field(default_factory=lambda: [
|
||||
"*.pyc",
|
||||
"*.log",
|
||||
"*.tmp",
|
||||
".DS_Store"
|
||||
])
|
||||
|
||||
# Tag formatting rules
|
||||
tag_format_rules: Dict[str, Any] = field(default_factory=lambda: {
|
||||
"case": "kebab", # kebab-case for tags
|
||||
"max_length": 50,
|
||||
"allowed_chars": "abcdefghijklmnopqrstuvwxyz0123456789-/",
|
||||
"hierarchy_separator": "/"
|
||||
})
|
||||
|
||||
# Language detection settings
|
||||
language_detection: Dict[str, Any] = field(default_factory=lambda: {
|
||||
"chinese_threshold": 0.1, # Minimum ratio of Chinese characters
|
||||
"mixed_threshold": 0.3, # Threshold for mixed language detection
|
||||
"min_content_length": 50 # Minimum content length for reliable detection
|
||||
})
|
||||
|
||||
def get_directory_mapping(self, directory_path: str) -> Optional[DirectoryMapping]:
|
||||
"""Get directory mapping for a given path."""
|
||||
for mapping in self.directory_mappings:
|
||||
if mapping.pattern in directory_path:
|
||||
return mapping
|
||||
return None
|
||||
|
||||
def get_tag_hierarchy(self, root_tag: str) -> Optional[TagHierarchy]:
|
||||
"""Get tag hierarchy for a root tag."""
|
||||
return self.tag_hierarchies.get(root_tag)
|
||||
|
||||
def is_excluded_directory(self, directory: str) -> bool:
|
||||
"""Check if directory should be excluded."""
|
||||
return any(excluded in directory for excluded in self.excluded_directories)
|
||||
|
||||
def is_excluded_file(self, filename: str) -> bool:
|
||||
"""Check if file should be excluded based on patterns."""
|
||||
import fnmatch
|
||||
return any(fnmatch.fnmatch(filename, pattern) for pattern in self.excluded_file_patterns)
|
||||
|
||||
|
||||
def load_config(config_path: Optional[Union[str, Path]] = None) -> TaggingConfig:
|
||||
"""Load configuration from file or return default configuration."""
|
||||
if config_path is None:
|
||||
return TaggingConfig()
|
||||
|
||||
config_path = Path(config_path)
|
||||
if not config_path.exists():
|
||||
# Create default config file
|
||||
default_config = TaggingConfig()
|
||||
save_config(default_config, config_path)
|
||||
return default_config
|
||||
|
||||
try:
|
||||
with open(config_path, 'r', encoding='utf-8') as f:
|
||||
if config_path.suffix.lower() == '.json':
|
||||
data = json.load(f)
|
||||
else: # Assume YAML
|
||||
data = yaml.safe_load(f)
|
||||
|
||||
# Convert dict to TaggingConfig (simplified conversion)
|
||||
# In a full implementation, you'd want more robust deserialization
|
||||
config = TaggingConfig()
|
||||
|
||||
# Update with loaded data
|
||||
if 'excluded_directories' in data:
|
||||
config.excluded_directories = data['excluded_directories']
|
||||
if 'excluded_file_patterns' in data:
|
||||
config.excluded_file_patterns = data['excluded_file_patterns']
|
||||
if 'tag_format_rules' in data:
|
||||
config.tag_format_rules.update(data['tag_format_rules'])
|
||||
if 'language_detection' in data:
|
||||
config.language_detection.update(data['language_detection'])
|
||||
|
||||
return config
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error loading config from {config_path}: {e}")
|
||||
return TaggingConfig()
|
||||
|
||||
|
||||
def save_config(config: TaggingConfig, config_path: Union[str, Path]) -> None:
|
||||
"""Save configuration to file."""
|
||||
config_path = Path(config_path)
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Convert to dict for serialization (simplified)
|
||||
config_dict = {
|
||||
'excluded_directories': config.excluded_directories,
|
||||
'excluded_file_patterns': config.excluded_file_patterns,
|
||||
'tag_format_rules': config.tag_format_rules,
|
||||
'language_detection': config.language_detection
|
||||
}
|
||||
|
||||
try:
|
||||
with open(config_path, 'w', encoding='utf-8') as f:
|
||||
if config_path.suffix.lower() == '.json':
|
||||
json.dump(config_dict, f, indent=2)
|
||||
else: # Save as YAML
|
||||
yaml.dump(config_dict, f, default_flow_style=False, allow_unicode=True)
|
||||
except Exception as e:
|
||||
print(f"Error saving config to {config_path}: {e}")
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Core components for the tagging system."""
|
||||
|
||||
from .models import (
|
||||
FileInfo,
|
||||
ContentAnalysis,
|
||||
TagStructure,
|
||||
FrontmatterData,
|
||||
ContentType,
|
||||
LanguageInfo,
|
||||
ValidationResult
|
||||
)
|
||||
|
||||
from .interfaces import (
|
||||
FileDiscovery,
|
||||
ContentAnalyzer,
|
||||
TagGenerator,
|
||||
FrontmatterManager
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"FileInfo",
|
||||
"ContentAnalysis",
|
||||
"TagStructure",
|
||||
"FrontmatterData",
|
||||
"ContentType",
|
||||
"LanguageInfo",
|
||||
"ValidationResult",
|
||||
"FileDiscovery",
|
||||
"ContentAnalyzer",
|
||||
"TagGenerator",
|
||||
"FrontmatterManager"
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,167 @@
|
||||
"""Core interfaces and protocols for the tagging system."""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import List, Protocol, runtime_checkable
|
||||
from .models import FileInfo, ContentAnalysis, TagStructure, FrontmatterData, ValidationResult
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class FileDiscovery(Protocol):
|
||||
"""Protocol for file discovery operations."""
|
||||
|
||||
def scan_directory(self, path: str) -> List[FileInfo]:
|
||||
"""Recursively scan directory and return file information."""
|
||||
...
|
||||
|
||||
def filter_by_type(self, files: List[FileInfo], types: List[str]) -> List[FileInfo]:
|
||||
"""Filter files by extension types."""
|
||||
...
|
||||
|
||||
def exclude_sensitive(self, files: List[FileInfo]) -> List[FileInfo]:
|
||||
"""Exclude sensitive directories and files."""
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ContentAnalyzer(Protocol):
|
||||
"""Protocol for content analysis operations."""
|
||||
|
||||
def analyze_content(self, content: str) -> ContentAnalysis:
|
||||
"""Analyze file content and return analysis results."""
|
||||
...
|
||||
|
||||
def detect_language(self, content: str) -> str:
|
||||
"""Detect the primary language of the content."""
|
||||
...
|
||||
|
||||
def extract_topics(self, content: str) -> List[str]:
|
||||
"""Extract topics from content."""
|
||||
...
|
||||
|
||||
def classify_content_type(self, content: str, filename: str) -> str:
|
||||
"""Classify the type of content."""
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class TagGenerator(Protocol):
|
||||
"""Protocol for tag generation operations."""
|
||||
|
||||
def generate_directory_tags(self, filepath: str) -> List[str]:
|
||||
"""Generate tags based on directory structure."""
|
||||
...
|
||||
|
||||
def generate_content_tags(self, analysis: ContentAnalysis) -> List[str]:
|
||||
"""Generate tags based on content analysis."""
|
||||
...
|
||||
|
||||
def generate_hierarchical_tags(self, topics: List[str]) -> List[str]:
|
||||
"""Generate hierarchical tags from topics."""
|
||||
...
|
||||
|
||||
def consolidate_tags(self, tags: List[str]) -> List[str]:
|
||||
"""Consolidate and deduplicate tags."""
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class FrontmatterManager(Protocol):
|
||||
"""Protocol for frontmatter management operations."""
|
||||
|
||||
def parse_frontmatter(self, content: str) -> FrontmatterData:
|
||||
"""Parse YAML frontmatter from content."""
|
||||
...
|
||||
|
||||
def update_frontmatter(self, content: str, updates: FrontmatterData) -> str:
|
||||
"""Update frontmatter in content while preserving existing data."""
|
||||
...
|
||||
|
||||
def validate_frontmatter(self, data: FrontmatterData) -> ValidationResult:
|
||||
"""Validate frontmatter structure and content."""
|
||||
...
|
||||
|
||||
|
||||
class BaseFileDiscovery(ABC):
|
||||
"""Abstract base class for file discovery implementations."""
|
||||
|
||||
@abstractmethod
|
||||
def scan_directory(self, path: str) -> List[FileInfo]:
|
||||
"""Recursively scan directory and return file information."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def filter_by_type(self, files: List[FileInfo], types: List[str]) -> List[FileInfo]:
|
||||
"""Filter files by extension types."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def exclude_sensitive(self, files: List[FileInfo]) -> List[FileInfo]:
|
||||
"""Exclude sensitive directories and files."""
|
||||
pass
|
||||
|
||||
|
||||
class BaseContentAnalyzer(ABC):
|
||||
"""Abstract base class for content analyzer implementations."""
|
||||
|
||||
@abstractmethod
|
||||
def analyze_content(self, content: str) -> ContentAnalysis:
|
||||
"""Analyze file content and return analysis results."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def detect_language(self, content: str) -> str:
|
||||
"""Detect the primary language of the content."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def extract_topics(self, content: str) -> List[str]:
|
||||
"""Extract topics from content."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def classify_content_type(self, content: str, filename: str) -> str:
|
||||
"""Classify the type of content."""
|
||||
pass
|
||||
|
||||
|
||||
class BaseTagGenerator(ABC):
|
||||
"""Abstract base class for tag generator implementations."""
|
||||
|
||||
@abstractmethod
|
||||
def generate_directory_tags(self, filepath: str) -> List[str]:
|
||||
"""Generate tags based on directory structure."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def generate_content_tags(self, analysis: ContentAnalysis) -> List[str]:
|
||||
"""Generate tags based on content analysis."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def generate_hierarchical_tags(self, topics: List[str]) -> List[str]:
|
||||
"""Generate hierarchical tags from topics."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def consolidate_tags(self, tags: List[str]) -> List[str]:
|
||||
"""Consolidate and deduplicate tags."""
|
||||
pass
|
||||
|
||||
|
||||
class BaseFrontmatterManager(ABC):
|
||||
"""Abstract base class for frontmatter manager implementations."""
|
||||
|
||||
@abstractmethod
|
||||
def parse_frontmatter(self, content: str) -> FrontmatterData:
|
||||
"""Parse YAML frontmatter from content."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update_frontmatter(self, content: str, updates: FrontmatterData) -> str:
|
||||
"""Update frontmatter in content while preserving existing data."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def validate_frontmatter(self, data: FrontmatterData) -> ValidationResult:
|
||||
"""Validate frontmatter structure and content."""
|
||||
pass
|
||||
@@ -0,0 +1,158 @@
|
||||
"""Core data models for the tagging system."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import List, Dict, Optional, Any
|
||||
|
||||
|
||||
class ContentType(Enum):
|
||||
"""Content type classifications."""
|
||||
HUB = "hub"
|
||||
NOTE = "note"
|
||||
CLIPPING = "clipping"
|
||||
DAILY_NOTE = "daily-note"
|
||||
MEETING = "meeting"
|
||||
DOCUMENTATION = "documentation"
|
||||
TUTORIAL = "tutorial"
|
||||
REFERENCE = "reference"
|
||||
DRAFT = "draft"
|
||||
PROJECT = "project"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class LanguageInfo(Enum):
|
||||
"""Language detection results."""
|
||||
ENGLISH = "en"
|
||||
CHINESE = "zh"
|
||||
MIXED = "mixed"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
@dataclass
|
||||
class FileInfo:
|
||||
"""Information about a file in the vault."""
|
||||
path: str
|
||||
name: str
|
||||
directory: str
|
||||
extension: str
|
||||
size: int
|
||||
created: datetime
|
||||
modified: datetime
|
||||
content: Optional[str] = None
|
||||
|
||||
@property
|
||||
def relative_path(self) -> str:
|
||||
"""Get the relative path from vault root."""
|
||||
return self.path
|
||||
|
||||
@property
|
||||
def is_markdown(self) -> bool:
|
||||
"""Check if file is a markdown file."""
|
||||
return self.extension.lower() in ['.md', '.markdown']
|
||||
|
||||
|
||||
@dataclass
|
||||
class ContentAnalysis:
|
||||
"""Results of content analysis."""
|
||||
language: LanguageInfo
|
||||
content_type: ContentType
|
||||
topics: List[str]
|
||||
mentions: Dict[str, List[str]] = field(default_factory=lambda: {
|
||||
'tools': [],
|
||||
'technologies': [],
|
||||
'people': [],
|
||||
'organizations': []
|
||||
})
|
||||
sentiment: Optional[str] = None
|
||||
complexity: str = "basic"
|
||||
|
||||
def __post_init__(self):
|
||||
"""Validate complexity level."""
|
||||
if self.complexity not in ['basic', 'intermediate', 'advanced']:
|
||||
self.complexity = 'basic'
|
||||
|
||||
|
||||
@dataclass
|
||||
class TagStructure:
|
||||
"""Structured representation of tags."""
|
||||
primary: List[str] = field(default_factory=list) # Main category tags
|
||||
hierarchical: List[str] = field(default_factory=list) # Topic/subtopic/detail tags
|
||||
content: List[str] = field(default_factory=list) # Content-derived tags
|
||||
meta: List[str] = field(default_factory=list) # Metadata tags (language, type, etc.)
|
||||
custom: List[str] = field(default_factory=list) # Manually added tags to preserve
|
||||
|
||||
def all_tags(self) -> List[str]:
|
||||
"""Get all tags as a flat list."""
|
||||
all_tags = []
|
||||
all_tags.extend(self.primary)
|
||||
all_tags.extend(self.hierarchical)
|
||||
all_tags.extend(self.content)
|
||||
all_tags.extend(self.meta)
|
||||
all_tags.extend(self.custom)
|
||||
return list(set(all_tags)) # Remove duplicates
|
||||
|
||||
|
||||
@dataclass
|
||||
class FrontmatterData:
|
||||
"""YAML frontmatter data structure."""
|
||||
title: Optional[str] = None
|
||||
tags: List[str] = field(default_factory=list)
|
||||
created: Optional[str] = None
|
||||
updated: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
lang: Optional[str] = None
|
||||
source: Optional[str] = None
|
||||
aliases: List[str] = field(default_factory=list)
|
||||
description: Optional[str] = None
|
||||
custom_fields: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Convert to dictionary for YAML serialization."""
|
||||
result = {}
|
||||
|
||||
if self.title:
|
||||
result['title'] = self.title
|
||||
if self.tags:
|
||||
result['tags'] = self.tags
|
||||
if self.created:
|
||||
result['created'] = self.created
|
||||
if self.updated:
|
||||
result['updated'] = self.updated
|
||||
if self.type:
|
||||
result['type'] = self.type
|
||||
if self.lang:
|
||||
result['lang'] = self.lang
|
||||
if self.source:
|
||||
result['source'] = self.source
|
||||
if self.aliases:
|
||||
result['aliases'] = self.aliases
|
||||
if self.description:
|
||||
result['description'] = self.description
|
||||
|
||||
# Add custom fields
|
||||
result.update(self.custom_fields)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@dataclass
|
||||
class ValidationResult:
|
||||
"""Result of validation operations."""
|
||||
is_valid: bool
|
||||
errors: List[str] = field(default_factory=list)
|
||||
warnings: List[str] = field(default_factory=list)
|
||||
suggestions: List[str] = field(default_factory=list)
|
||||
|
||||
def add_error(self, message: str):
|
||||
"""Add an error message."""
|
||||
self.errors.append(message)
|
||||
self.is_valid = False
|
||||
|
||||
def add_warning(self, message: str):
|
||||
"""Add a warning message."""
|
||||
self.warnings.append(message)
|
||||
|
||||
def add_suggestion(self, message: str):
|
||||
"""Add a suggestion message."""
|
||||
self.suggestions.append(message)
|
||||
@@ -0,0 +1,7 @@
|
||||
"""Implementation modules for the tagging system."""
|
||||
|
||||
from .file_discovery import VaultFileDiscovery
|
||||
from .content_analyzer import VaultContentAnalyzer
|
||||
from .tag_generator import TagGeneratorImpl
|
||||
|
||||
__all__ = ['VaultFileDiscovery', 'VaultContentAnalyzer', 'TagGeneratorImpl']
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,303 @@
|
||||
"""Content analysis implementation for extracting topics, language, and content type."""
|
||||
|
||||
import re
|
||||
from collections import Counter
|
||||
from typing import List, Dict, Set, Tuple
|
||||
from ..core.interfaces import BaseContentAnalyzer
|
||||
from ..core.models import ContentAnalysis, ContentType, LanguageInfo
|
||||
|
||||
|
||||
class VaultContentAnalyzer(BaseContentAnalyzer):
|
||||
"""Implementation of content analyzer for Obsidian vault content."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the content analyzer with patterns and keywords."""
|
||||
|
||||
# Language detection patterns
|
||||
self.chinese_chars = re.compile(r'[\u4e00-\u9fff]')
|
||||
self.english_chars = re.compile(r'[a-zA-Z]')
|
||||
|
||||
# Content type patterns
|
||||
self.content_type_patterns = {
|
||||
ContentType.HUB: [
|
||||
r'# .+\n\n.*(?:index|hub|overview|contents?)',
|
||||
r'## (?:Projects?|Areas?|Resources?|Archive)',
|
||||
r'dataview\s*```',
|
||||
r'!\[\[.*\]\].*!\[\[.*\]\]', # Multiple embeds
|
||||
],
|
||||
ContentType.CLIPPING: [
|
||||
r'source:\s*https?://',
|
||||
r'clipped from:',
|
||||
r'saved from:',
|
||||
r'ReadItLater',
|
||||
r'# .+ - .+\.com',
|
||||
],
|
||||
ContentType.DAILY_NOTE: [
|
||||
r'^\d{4}-\d{2}-\d{2}',
|
||||
r'# \d{4}-\d{2}-\d{2}',
|
||||
r'## Daily Notes?',
|
||||
r'## Today',
|
||||
],
|
||||
ContentType.MEETING: [
|
||||
r'# Meeting:',
|
||||
r'## Attendees?',
|
||||
r'## Action Items?',
|
||||
r'## Minutes',
|
||||
r'meeting notes?',
|
||||
],
|
||||
ContentType.DOCUMENTATION: [
|
||||
r'# (?:API|Documentation|Guide|Manual)',
|
||||
r'## Installation',
|
||||
r'## Usage',
|
||||
r'## Configuration',
|
||||
r'```(?:bash|shell|cmd)',
|
||||
],
|
||||
ContentType.TUTORIAL: [
|
||||
r'# (?:How to|Tutorial|Step by Step)',
|
||||
r'## Step \d+',
|
||||
r'### Prerequisites?',
|
||||
r'## Getting Started',
|
||||
],
|
||||
ContentType.REFERENCE: [
|
||||
r'# (?:Reference|Cheat ?Sheet|Quick Reference)',
|
||||
r'## Commands?',
|
||||
r'## Syntax',
|
||||
r'## Examples?',
|
||||
],
|
||||
ContentType.DRAFT: [
|
||||
r'=Draft=',
|
||||
r'# Draft:',
|
||||
r'status:\s*draft',
|
||||
r'TODO:',
|
||||
r'FIXME:',
|
||||
],
|
||||
}
|
||||
|
||||
# Technology and tool keywords
|
||||
self.tech_keywords = {
|
||||
'ai': ['gpt', 'llm', 'chatgpt', 'openai', 'claude', 'anthropic', 'deepseek', 'ollama'],
|
||||
'infrastructure': ['docker', 'kubernetes', 'aws', 'azure', 'gcp', 'terraform', 'ansible'],
|
||||
'development': ['python', 'javascript', 'typescript', 'react', 'vue', 'node', 'npm', 'yarn'],
|
||||
'database': ['mysql', 'postgresql', 'mongodb', 'redis', 'sqlite', 'oracle'],
|
||||
'web': ['html', 'css', 'http', 'api', 'rest', 'graphql', 'json', 'xml'],
|
||||
'devops': ['ci/cd', 'jenkins', 'github actions', 'gitlab', 'git', 'version control'],
|
||||
'network': ['vpn', 'proxy', 'nginx', 'apache', 'dns', 'ssl', 'tls'],
|
||||
'security': ['encryption', 'authentication', 'authorization', 'oauth', 'jwt', 'ssl'],
|
||||
'mobile': ['ios', 'android', 'react native', 'flutter', 'swift', 'kotlin'],
|
||||
'home-automation': ['home assistant', 'zigbee', 'mqtt', 'esphome', 'tuya'],
|
||||
}
|
||||
|
||||
# Topic extraction patterns
|
||||
self.topic_patterns = {
|
||||
'project_management': ['project', 'task', 'milestone', 'deadline', 'planning'],
|
||||
'productivity': ['gtd', 'productivity', 'workflow', 'automation', 'efficiency'],
|
||||
'health': ['health', 'fitness', 'exercise', 'cycling', 'nutrition'],
|
||||
'finance': ['budget', 'investment', 'money', 'financial', 'ynab'],
|
||||
'cooking': ['recipe', 'cooking', 'ingredient', 'meal', 'food'],
|
||||
'travel': ['travel', 'trip', 'vacation', 'hotel', 'flight'],
|
||||
'gaming': ['game', 'gaming', 'steam', 'console', 'multiplayer'],
|
||||
'writing': ['blog', 'article', 'writing', 'content', 'publish'],
|
||||
}
|
||||
|
||||
# Entity extraction patterns
|
||||
self.entity_patterns = {
|
||||
'tools': re.compile(r'\b(?:obsidian|notion|vscode|cursor|docker|kubernetes|git|npm|yarn|pip)\b', re.IGNORECASE),
|
||||
'technologies': re.compile(r'\b(?:python|javascript|typescript|react|vue|node|html|css|sql|json|yaml)\b', re.IGNORECASE),
|
||||
'organizations': re.compile(r'\b(?:google|microsoft|apple|amazon|meta|openai|anthropic|github|gitlab)\b', re.IGNORECASE),
|
||||
'people': re.compile(r'@[a-zA-Z0-9_]+|(?:by|from|author:)\s+([A-Z][a-z]+\s+[A-Z][a-z]+)'),
|
||||
}
|
||||
|
||||
def analyze_content(self, content: str) -> ContentAnalysis:
|
||||
"""Analyze file content and return comprehensive analysis results."""
|
||||
if not content or not content.strip():
|
||||
return ContentAnalysis(
|
||||
language=LanguageInfo.UNKNOWN,
|
||||
content_type=ContentType.UNKNOWN,
|
||||
topics=[],
|
||||
mentions={'tools': [], 'technologies': [], 'people': [], 'organizations': []},
|
||||
complexity='basic'
|
||||
)
|
||||
|
||||
# Detect language
|
||||
language = self._detect_language_enum(content)
|
||||
|
||||
# Classify content type
|
||||
content_type = self._classify_content_type_enum(content, "")
|
||||
|
||||
# Extract topics
|
||||
topics = self.extract_topics(content)
|
||||
|
||||
# Extract mentions
|
||||
mentions = self._extract_mentions(content)
|
||||
|
||||
# Determine complexity
|
||||
complexity = self._determine_complexity(content)
|
||||
|
||||
return ContentAnalysis(
|
||||
language=language,
|
||||
content_type=content_type,
|
||||
topics=topics,
|
||||
mentions=mentions,
|
||||
complexity=complexity
|
||||
)
|
||||
|
||||
def detect_language(self, content: str) -> str:
|
||||
"""Detect the primary language of the content."""
|
||||
return self._detect_language_enum(content).value
|
||||
|
||||
def extract_topics(self, content: str) -> List[str]:
|
||||
"""Extract topics from content using keyword analysis."""
|
||||
if not content:
|
||||
return []
|
||||
|
||||
content_lower = content.lower()
|
||||
topics = []
|
||||
|
||||
# Check technology topics
|
||||
for tech_category, keywords in self.tech_keywords.items():
|
||||
if any(keyword in content_lower for keyword in keywords):
|
||||
topics.append(f'tech/{tech_category}')
|
||||
|
||||
# Check general topics
|
||||
for topic, keywords in self.topic_patterns.items():
|
||||
if any(keyword in content_lower for keyword in keywords):
|
||||
topics.append(topic)
|
||||
|
||||
# Extract topics from headers
|
||||
header_topics = self._extract_header_topics(content)
|
||||
topics.extend(header_topics)
|
||||
|
||||
# Remove duplicates and return
|
||||
return list(set(topics))
|
||||
|
||||
def classify_content_type(self, content: str, filename: str) -> str:
|
||||
"""Classify the type of content."""
|
||||
return self._classify_content_type_enum(content, filename).value
|
||||
|
||||
def _detect_language_enum(self, content: str) -> LanguageInfo:
|
||||
"""Detect language and return LanguageInfo enum."""
|
||||
if not content:
|
||||
return LanguageInfo.UNKNOWN
|
||||
|
||||
# Count character types
|
||||
chinese_count = len(self.chinese_chars.findall(content))
|
||||
english_count = len(self.english_chars.findall(content))
|
||||
total_chars = chinese_count + english_count
|
||||
|
||||
if total_chars == 0:
|
||||
return LanguageInfo.UNKNOWN
|
||||
|
||||
chinese_ratio = chinese_count / total_chars
|
||||
english_ratio = english_count / total_chars
|
||||
|
||||
# Determine language based on ratios
|
||||
if chinese_ratio > 0.3 and english_ratio > 0.3:
|
||||
return LanguageInfo.MIXED
|
||||
elif chinese_ratio > 0.1:
|
||||
return LanguageInfo.CHINESE
|
||||
elif english_ratio > 0.5:
|
||||
return LanguageInfo.ENGLISH
|
||||
else:
|
||||
return LanguageInfo.UNKNOWN
|
||||
|
||||
def _classify_content_type_enum(self, content: str, filename: str) -> ContentType:
|
||||
"""Classify content type and return ContentType enum."""
|
||||
if not content:
|
||||
return ContentType.UNKNOWN
|
||||
|
||||
# Check filename patterns first
|
||||
filename_lower = filename.lower()
|
||||
if re.match(r'\d{4}-\d{2}-\d{2}', filename_lower):
|
||||
return ContentType.DAILY_NOTE
|
||||
|
||||
# Check content patterns
|
||||
for content_type, patterns in self.content_type_patterns.items():
|
||||
for pattern in patterns:
|
||||
if re.search(pattern, content, re.IGNORECASE | re.MULTILINE):
|
||||
return content_type
|
||||
|
||||
# Default classification based on content characteristics
|
||||
if len(content.split('\n')) < 10:
|
||||
return ContentType.NOTE
|
||||
elif '```' in content and ('##' in content or '###' in content):
|
||||
return ContentType.DOCUMENTATION
|
||||
elif content.count('#') > 3:
|
||||
return ContentType.REFERENCE
|
||||
else:
|
||||
return ContentType.NOTE
|
||||
|
||||
def _extract_mentions(self, content: str) -> Dict[str, List[str]]:
|
||||
"""Extract mentions of tools, technologies, people, and organizations."""
|
||||
mentions = {
|
||||
'tools': [],
|
||||
'technologies': [],
|
||||
'people': [],
|
||||
'organizations': []
|
||||
}
|
||||
|
||||
for entity_type, pattern in self.entity_patterns.items():
|
||||
matches = pattern.findall(content)
|
||||
if entity_type == 'people':
|
||||
# Special handling for people mentions
|
||||
people = []
|
||||
for match in matches:
|
||||
if isinstance(match, tuple):
|
||||
people.extend([m for m in match if m])
|
||||
else:
|
||||
people.append(match)
|
||||
mentions[entity_type] = list(set(people))
|
||||
else:
|
||||
mentions[entity_type] = list(set(matches))
|
||||
|
||||
return mentions
|
||||
|
||||
def _extract_header_topics(self, content: str) -> List[str]:
|
||||
"""Extract topics from markdown headers."""
|
||||
topics = []
|
||||
|
||||
# Find all headers
|
||||
header_pattern = re.compile(r'^#+\s+(.+)$', re.MULTILINE)
|
||||
headers = header_pattern.findall(content)
|
||||
|
||||
for header in headers:
|
||||
header_lower = header.lower().strip()
|
||||
|
||||
# Skip common header words
|
||||
skip_words = {'introduction', 'overview', 'conclusion', 'summary', 'notes', 'todo', 'done'}
|
||||
if header_lower in skip_words:
|
||||
continue
|
||||
|
||||
# Extract meaningful topics from headers
|
||||
words = re.findall(r'\b[a-zA-Z]{3,}\b', header_lower)
|
||||
for word in words:
|
||||
if word not in skip_words and len(word) > 3:
|
||||
topics.append(word)
|
||||
|
||||
return topics[:5] # Limit to top 5 header topics
|
||||
|
||||
def _determine_complexity(self, content: str) -> str:
|
||||
"""Determine content complexity based on various factors."""
|
||||
if not content:
|
||||
return 'basic'
|
||||
|
||||
# Count various complexity indicators
|
||||
code_blocks = content.count('```')
|
||||
links = content.count('http')
|
||||
technical_terms = sum(1 for category in self.tech_keywords.values()
|
||||
for term in category if term in content.lower())
|
||||
word_count = len(content.split())
|
||||
|
||||
# Calculate complexity score
|
||||
complexity_score = 0
|
||||
complexity_score += min(code_blocks * 2, 10) # Code blocks add complexity
|
||||
complexity_score += min(links, 5) # External links add complexity
|
||||
complexity_score += min(technical_terms, 15) # Technical terms add complexity
|
||||
complexity_score += min(word_count // 500, 10) # Length adds complexity
|
||||
|
||||
# Classify based on score
|
||||
if complexity_score >= 20:
|
||||
return 'advanced'
|
||||
elif complexity_score >= 10:
|
||||
return 'intermediate'
|
||||
else:
|
||||
return 'basic'
|
||||
@@ -0,0 +1,234 @@
|
||||
"""File discovery implementation for vault scanning."""
|
||||
|
||||
import os
|
||||
import chardet
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Set
|
||||
from ..core.interfaces import BaseFileDiscovery
|
||||
from ..core.models import FileInfo
|
||||
|
||||
|
||||
class VaultFileDiscovery(BaseFileDiscovery):
|
||||
"""Implementation of file discovery for Obsidian vault scanning."""
|
||||
|
||||
def __init__(self, vault_root: str):
|
||||
"""Initialize with vault root directory."""
|
||||
self.vault_root = Path(vault_root).resolve()
|
||||
|
||||
# Sensitive directories to exclude
|
||||
self.sensitive_dirs = {
|
||||
'.obsidian',
|
||||
'.git',
|
||||
'.smart-env',
|
||||
'.pytest_cache',
|
||||
'__pycache__',
|
||||
'node_modules',
|
||||
'.vscode',
|
||||
'.idea',
|
||||
'400-archive/security-sensitive' # From the vault structure
|
||||
}
|
||||
|
||||
# File extensions to include (primarily text files)
|
||||
self.allowed_extensions = {
|
||||
'.md', '.markdown', '.txt', '.yaml', '.yml', '.json',
|
||||
'.py', '.js', '.ts', '.html', '.css', '.xml'
|
||||
}
|
||||
|
||||
# File patterns to exclude
|
||||
self.excluded_patterns = {
|
||||
'.DS_Store',
|
||||
'Thumbs.db',
|
||||
'.gitignore',
|
||||
'.gitkeep'
|
||||
}
|
||||
|
||||
def scan_directory(self, path: str) -> List[FileInfo]:
|
||||
"""Recursively scan directory and return file information."""
|
||||
scan_path = Path(path)
|
||||
if not scan_path.is_absolute():
|
||||
scan_path = self.vault_root / scan_path
|
||||
|
||||
files = []
|
||||
|
||||
try:
|
||||
for root, dirs, filenames in os.walk(scan_path):
|
||||
root_path = Path(root)
|
||||
|
||||
# Filter out sensitive directories
|
||||
dirs[:] = [d for d in dirs if not self._is_sensitive_dir(root_path / d)]
|
||||
|
||||
for filename in filenames:
|
||||
file_path = root_path / filename
|
||||
|
||||
# Skip excluded patterns
|
||||
if filename in self.excluded_patterns:
|
||||
continue
|
||||
|
||||
# Check if file extension is allowed
|
||||
if not self._is_allowed_file(file_path):
|
||||
continue
|
||||
|
||||
try:
|
||||
file_info = self._create_file_info(file_path)
|
||||
if file_info:
|
||||
files.append(file_info)
|
||||
except (OSError, PermissionError) as e:
|
||||
# Log error but continue processing
|
||||
print(f"Warning: Could not process file {file_path}: {e}")
|
||||
continue
|
||||
|
||||
except (OSError, PermissionError) as e:
|
||||
print(f"Error scanning directory {scan_path}: {e}")
|
||||
|
||||
return files
|
||||
|
||||
def filter_by_type(self, files: List[FileInfo], types: List[str]) -> List[FileInfo]:
|
||||
"""Filter files by extension types."""
|
||||
if not types:
|
||||
return files
|
||||
|
||||
# Normalize extensions (ensure they start with .)
|
||||
normalized_types = set()
|
||||
for ext in types:
|
||||
if not ext.startswith('.'):
|
||||
ext = '.' + ext
|
||||
normalized_types.add(ext.lower())
|
||||
|
||||
return [f for f in files if f.extension.lower() in normalized_types]
|
||||
|
||||
def exclude_sensitive(self, files: List[FileInfo]) -> List[FileInfo]:
|
||||
"""Exclude sensitive directories and files."""
|
||||
filtered_files = []
|
||||
|
||||
for file_info in files:
|
||||
file_path = Path(file_info.path)
|
||||
|
||||
# Check if file is in sensitive directory
|
||||
if self._is_in_sensitive_dir(file_path):
|
||||
continue
|
||||
|
||||
# Check for sensitive content patterns in filename
|
||||
if self._has_sensitive_filename(file_info.name):
|
||||
continue
|
||||
|
||||
filtered_files.append(file_info)
|
||||
|
||||
return filtered_files
|
||||
|
||||
def _create_file_info(self, file_path: Path) -> FileInfo:
|
||||
"""Create FileInfo object from file path."""
|
||||
try:
|
||||
stat = file_path.stat()
|
||||
|
||||
# Get relative path from vault root
|
||||
try:
|
||||
relative_path = file_path.relative_to(self.vault_root)
|
||||
except ValueError:
|
||||
# File is outside vault root, use absolute path
|
||||
relative_path = file_path
|
||||
|
||||
# Read content for text files
|
||||
content = None
|
||||
if file_path.suffix.lower() in {'.md', '.markdown', '.txt', '.yaml', '.yml'}:
|
||||
content = self._read_file_content(file_path)
|
||||
|
||||
return FileInfo(
|
||||
path=str(relative_path),
|
||||
name=file_path.name,
|
||||
directory=str(relative_path.parent) if relative_path.parent != Path('.') else '',
|
||||
extension=file_path.suffix,
|
||||
size=stat.st_size,
|
||||
created=datetime.fromtimestamp(stat.st_ctime),
|
||||
modified=datetime.fromtimestamp(stat.st_mtime),
|
||||
content=content
|
||||
)
|
||||
|
||||
except (OSError, PermissionError):
|
||||
return None
|
||||
|
||||
def _read_file_content(self, file_path: Path) -> str:
|
||||
"""Read file content with encoding detection."""
|
||||
try:
|
||||
# First try UTF-8
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
return f.read()
|
||||
except UnicodeDecodeError:
|
||||
try:
|
||||
# Detect encoding
|
||||
with open(file_path, 'rb') as f:
|
||||
raw_data = f.read()
|
||||
|
||||
detected = chardet.detect(raw_data)
|
||||
encoding = detected.get('encoding', 'utf-8')
|
||||
|
||||
# Try detected encoding
|
||||
return raw_data.decode(encoding, errors='replace')
|
||||
|
||||
except Exception:
|
||||
# Fallback to reading as binary and replacing errors
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
|
||||
return f.read()
|
||||
except Exception:
|
||||
return ""
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
def _is_sensitive_dir(self, dir_path: Path) -> bool:
|
||||
"""Check if directory should be excluded as sensitive."""
|
||||
dir_name = dir_path.name
|
||||
|
||||
# Check exact matches
|
||||
if dir_name in self.sensitive_dirs:
|
||||
return True
|
||||
|
||||
# Check relative path matches
|
||||
try:
|
||||
relative_path = dir_path.relative_to(self.vault_root)
|
||||
if str(relative_path) in self.sensitive_dirs:
|
||||
return True
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Check for hidden directories (starting with .)
|
||||
if dir_name.startswith('.') and dir_name not in {'.kiro'}:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _is_in_sensitive_dir(self, file_path: Path) -> bool:
|
||||
"""Check if file is in a sensitive directory."""
|
||||
try:
|
||||
relative_path = file_path.relative_to(self.vault_root)
|
||||
path_parts = relative_path.parts
|
||||
|
||||
for part in path_parts[:-1]: # Exclude filename
|
||||
if part in self.sensitive_dirs:
|
||||
return True
|
||||
if part.startswith('.') and part not in {'.kiro'}:
|
||||
return True
|
||||
|
||||
# Check full directory path
|
||||
dir_path = str(relative_path.parent)
|
||||
if dir_path in self.sensitive_dirs:
|
||||
return True
|
||||
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
def _is_allowed_file(self, file_path: Path) -> bool:
|
||||
"""Check if file extension is allowed."""
|
||||
return file_path.suffix.lower() in self.allowed_extensions
|
||||
|
||||
def _has_sensitive_filename(self, filename: str) -> bool:
|
||||
"""Check if filename indicates sensitive content."""
|
||||
sensitive_patterns = {
|
||||
'password', 'secret', 'key', 'token', 'credential',
|
||||
'private', 'confidential', 'sensitive'
|
||||
}
|
||||
|
||||
filename_lower = filename.lower()
|
||||
return any(pattern in filename_lower for pattern in sensitive_patterns)
|
||||
@@ -0,0 +1,331 @@
|
||||
"""Tag generation implementation for the comprehensive tagging system."""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Set
|
||||
from ..core.interfaces import BaseTagGenerator
|
||||
from ..core.models import ContentAnalysis, TagStructure
|
||||
|
||||
|
||||
class TagGeneratorImpl(BaseTagGenerator):
|
||||
"""Implementation of tag generation based on directory structure and content analysis."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the tag generator with predefined mappings and patterns."""
|
||||
# Directory-based tag mappings
|
||||
self.directory_mappings = {
|
||||
'100-project': 'project',
|
||||
'200-area': 'area',
|
||||
'300-resources': 'resource',
|
||||
'400-archive': 'archive',
|
||||
'clippings': 'clipping',
|
||||
'readitlater inbox': 'clipping'
|
||||
}
|
||||
|
||||
# Hierarchical topic mappings
|
||||
self.topic_hierarchies = {
|
||||
# Technology hierarchies
|
||||
'ai': 'tech/ai',
|
||||
'llm': 'tech/ai/llm',
|
||||
'machine learning': 'tech/ai/ml',
|
||||
'chatgpt': 'tech/ai/llm',
|
||||
'openai': 'tech/ai/llm',
|
||||
'claude': 'tech/ai/llm',
|
||||
'docker': 'tech/infrastructure/docker',
|
||||
'kubernetes': 'tech/infrastructure/k8s',
|
||||
'python': 'tech/development/python',
|
||||
'javascript': 'tech/development/javascript',
|
||||
'typescript': 'tech/development/typescript',
|
||||
'react': 'tech/development/react',
|
||||
'vue': 'tech/development/vue',
|
||||
'node': 'tech/development/nodejs',
|
||||
'api': 'tech/development/api',
|
||||
'database': 'tech/infrastructure/database',
|
||||
'mysql': 'tech/infrastructure/database',
|
||||
'postgresql': 'tech/infrastructure/database',
|
||||
'mongodb': 'tech/infrastructure/database',
|
||||
'redis': 'tech/infrastructure/database',
|
||||
'nginx': 'tech/infrastructure/web',
|
||||
'apache': 'tech/infrastructure/web',
|
||||
'aws': 'tech/infrastructure/cloud',
|
||||
'azure': 'tech/infrastructure/cloud',
|
||||
'gcp': 'tech/infrastructure/cloud',
|
||||
'linux': 'tech/infrastructure/os',
|
||||
'ubuntu': 'tech/infrastructure/os',
|
||||
'centos': 'tech/infrastructure/os',
|
||||
|
||||
# Personal hierarchies
|
||||
'productivity': 'personal/productivity',
|
||||
'gtd': 'personal/productivity/gtd',
|
||||
'health': 'personal/health',
|
||||
'cycling': 'personal/health/cycling',
|
||||
'fitness': 'personal/health/fitness',
|
||||
'finance': 'personal/finance',
|
||||
'investment': 'personal/finance/investment',
|
||||
'budget': 'personal/finance/budget',
|
||||
'cooking': 'personal/cooking',
|
||||
'recipe': 'personal/cooking/recipe',
|
||||
|
||||
# Work hierarchies
|
||||
'government': 'work/government',
|
||||
'enterprise': 'work/enterprise',
|
||||
'airport': 'work/airport',
|
||||
'ali': 'work/ali',
|
||||
|
||||
# Home automation hierarchies
|
||||
'home assistant': 'home-automation/hass',
|
||||
'esphome': 'home-automation/esphome',
|
||||
'zigbee': 'home-automation/zigbee',
|
||||
'mqtt': 'home-automation/mqtt',
|
||||
'sensor': 'home-automation/sensor',
|
||||
'automation': 'home-automation/automation'
|
||||
}
|
||||
|
||||
# Sensitive content patterns
|
||||
self.sensitive_patterns = {
|
||||
'credentials': [
|
||||
r'password\s*[:=]\s*["\']?[\w\-@#$%^&*()]+["\']?',
|
||||
r'api[_\-]?key\s*[:=]\s*["\']?[\w\-]+["\']?',
|
||||
r'secret\s*[:=]\s*["\']?[\w\-]+["\']?',
|
||||
r'token\s*[:=]\s*["\']?[\w\-\.]+["\']?',
|
||||
r'auth[_\-]?token\s*[:=]\s*["\']?[\w\-\.]+["\']?',
|
||||
r'access[_\-]?key\s*[:=]\s*["\']?[\w\-]+["\']?',
|
||||
r'private[_\-]?key',
|
||||
r'ssh[_\-]?key',
|
||||
r'-----BEGIN.*PRIVATE KEY-----'
|
||||
],
|
||||
'personal': [
|
||||
r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', # Credit card numbers
|
||||
r'\b\d{3}-\d{2}-\d{4}\b', # SSN format
|
||||
r'\b\d{11}\b', # Phone numbers (simplified)
|
||||
r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', # Email addresses
|
||||
r'\b(?:home|personal|private)\s+(?:address|phone|email)',
|
||||
r'\bbirthdate\b|\bdate\s+of\s+birth\b'
|
||||
],
|
||||
'financial': [
|
||||
r'\b(?:salary|income|wage)\s*[:=]\s*[\$¥€£]?[\d,]+',
|
||||
r'\b(?:bank|account)\s+(?:number|details)',
|
||||
r'\b(?:routing|swift)\s+(?:number|code)',
|
||||
r'\b(?:tax|invoice|receipt)\s+(?:id|number)',
|
||||
r'\b(?:budget|expense|cost)\s*[:=]\s*[\$¥€£]?[\d,]+',
|
||||
r'\b(?:investment|portfolio|stock)\s+(?:value|amount)'
|
||||
],
|
||||
'legal': [
|
||||
r'\b(?:contract|agreement|legal)\s+(?:document|file)',
|
||||
r'\b(?:confidential|proprietary|classified)',
|
||||
r'\b(?:copyright|trademark|patent)\s+(?:notice|info)',
|
||||
r'\b(?:license|licensing)\s+(?:agreement|terms)',
|
||||
r'\bnda\b|\bnon[_\-]?disclosure\b'
|
||||
]
|
||||
}
|
||||
|
||||
# Tag validation patterns
|
||||
self.valid_tag_pattern = re.compile(r'^[a-z0-9]+(?:[-/][a-z0-9]+)*$')
|
||||
|
||||
def generate_directory_tags(self, filepath: str) -> List[str]:
|
||||
"""Generate tags based on directory structure."""
|
||||
tags = []
|
||||
path = Path(filepath)
|
||||
parts = [p.lower() for p in path.parts]
|
||||
|
||||
# Generate primary directory tags
|
||||
for part in parts:
|
||||
if part in self.directory_mappings:
|
||||
primary_tag = self.directory_mappings[part]
|
||||
tags.append(primary_tag)
|
||||
|
||||
# Add hierarchical subdirectory tags
|
||||
try:
|
||||
part_index = parts.index(part)
|
||||
if part_index + 1 < len(parts):
|
||||
subdirs = parts[part_index + 1:-1] # Exclude filename
|
||||
for subdir in subdirs:
|
||||
# Clean and validate subdirectory name
|
||||
clean_subdir = self._clean_tag_name(subdir)
|
||||
if clean_subdir:
|
||||
hierarchical_tag = f"{primary_tag}/{clean_subdir}"
|
||||
tags.append(hierarchical_tag)
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
# Handle special cases
|
||||
if any('clipping' in part for part in parts):
|
||||
tags.append('clipping')
|
||||
|
||||
return list(set(tags)) # Remove duplicates
|
||||
|
||||
def generate_content_tags(self, analysis: ContentAnalysis) -> List[str]:
|
||||
"""Generate tags based on content analysis."""
|
||||
tags = []
|
||||
|
||||
# Add language tag
|
||||
if analysis.language:
|
||||
tags.append(f"lang/{analysis.language.value}")
|
||||
|
||||
# Add content type tag
|
||||
if analysis.content_type:
|
||||
tags.append(f"type/{analysis.content_type.value}")
|
||||
|
||||
# Add complexity tag if not basic
|
||||
if analysis.complexity and analysis.complexity != 'basic':
|
||||
tags.append(f"complexity/{analysis.complexity}")
|
||||
|
||||
# Add sentiment tag if available
|
||||
if analysis.sentiment and analysis.sentiment != 'neutral':
|
||||
tags.append(f"sentiment/{analysis.sentiment}")
|
||||
|
||||
# Add mention-based tags
|
||||
for category, items in analysis.mentions.items():
|
||||
for item in items:
|
||||
clean_item = self._clean_tag_name(item)
|
||||
if clean_item:
|
||||
tags.append(f"{category}/{clean_item}")
|
||||
|
||||
return tags
|
||||
|
||||
def generate_hierarchical_tags(self, topics: List[str]) -> List[str]:
|
||||
"""Generate hierarchical tags from topics."""
|
||||
hierarchical_tags = []
|
||||
|
||||
for topic in topics:
|
||||
topic_lower = topic.lower().strip()
|
||||
|
||||
# Check for direct mapping
|
||||
if topic_lower in self.topic_hierarchies:
|
||||
hierarchical_tags.append(self.topic_hierarchies[topic_lower])
|
||||
else:
|
||||
# Try partial matching for compound topics
|
||||
for key, hierarchy in self.topic_hierarchies.items():
|
||||
if key in topic_lower or topic_lower in key:
|
||||
hierarchical_tags.append(hierarchy)
|
||||
break
|
||||
else:
|
||||
# Create a generic hierarchical tag
|
||||
clean_topic = self._clean_tag_name(topic_lower)
|
||||
if clean_topic:
|
||||
# Try to categorize based on common patterns
|
||||
if any(tech_word in topic_lower for tech_word in ['tech', 'software', 'code', 'dev', 'program']):
|
||||
hierarchical_tags.append(f"tech/{clean_topic}")
|
||||
elif any(personal_word in topic_lower for personal_word in ['personal', 'life', 'habit', 'goal']):
|
||||
hierarchical_tags.append(f"personal/{clean_topic}")
|
||||
elif any(work_word in topic_lower for work_word in ['work', 'job', 'career', 'business']):
|
||||
hierarchical_tags.append(f"work/{clean_topic}")
|
||||
else:
|
||||
hierarchical_tags.append(clean_topic)
|
||||
|
||||
return list(set(hierarchical_tags))
|
||||
|
||||
def consolidate_tags(self, tags: List[str]) -> List[str]:
|
||||
"""Consolidate and deduplicate tags."""
|
||||
if not tags:
|
||||
return []
|
||||
|
||||
# Clean and validate all tags
|
||||
cleaned_tags = []
|
||||
for tag in tags:
|
||||
clean_tag = self._clean_tag_name(tag)
|
||||
if clean_tag and self._is_valid_tag(clean_tag):
|
||||
cleaned_tags.append(clean_tag)
|
||||
|
||||
# Remove duplicates while preserving order
|
||||
seen = set()
|
||||
consolidated = []
|
||||
for tag in cleaned_tags:
|
||||
if tag not in seen:
|
||||
seen.add(tag)
|
||||
consolidated.append(tag)
|
||||
|
||||
# Apply consolidation rules
|
||||
consolidated = self._apply_consolidation_rules(consolidated)
|
||||
|
||||
# Sort tags for consistency (hierarchical tags first, then alphabetical)
|
||||
return self._sort_tags(consolidated)
|
||||
|
||||
def detect_sensitive_content(self, content: str, filepath: str) -> List[str]:
|
||||
"""Detect sensitive content and return appropriate tags."""
|
||||
sensitive_tags = []
|
||||
content_lower = content.lower()
|
||||
|
||||
# Check for sensitive patterns
|
||||
for category, patterns in self.sensitive_patterns.items():
|
||||
for pattern in patterns:
|
||||
if re.search(pattern, content, re.IGNORECASE):
|
||||
sensitive_tags.append(f"sensitive/{category}")
|
||||
break # Only add the category once
|
||||
|
||||
# Check filepath for sensitive indicators
|
||||
filepath_lower = filepath.lower()
|
||||
if any(sensitive_dir in filepath_lower for sensitive_dir in ['security-sensitive', 'private', 'confidential']):
|
||||
if 'sensitive/personal' not in sensitive_tags:
|
||||
sensitive_tags.append('sensitive/personal')
|
||||
|
||||
return list(set(sensitive_tags))
|
||||
|
||||
def _clean_tag_name(self, tag: str) -> str:
|
||||
"""Clean and normalize tag names to kebab-case."""
|
||||
if not tag:
|
||||
return ""
|
||||
|
||||
# Convert to lowercase and replace spaces/underscores with hyphens
|
||||
cleaned = re.sub(r'[_\s]+', '-', tag.lower().strip())
|
||||
|
||||
# Remove special characters except hyphens and forward slashes
|
||||
cleaned = re.sub(r'[^a-z0-9\-/]', '', cleaned)
|
||||
|
||||
# Remove multiple consecutive hyphens
|
||||
cleaned = re.sub(r'-+', '-', cleaned)
|
||||
|
||||
# Remove leading/trailing hyphens
|
||||
cleaned = cleaned.strip('-')
|
||||
|
||||
return cleaned
|
||||
|
||||
def _is_valid_tag(self, tag: str) -> bool:
|
||||
"""Validate tag format."""
|
||||
if not tag:
|
||||
return False
|
||||
|
||||
# Check against valid pattern
|
||||
if not self.valid_tag_pattern.match(tag):
|
||||
return False
|
||||
|
||||
# Additional validation rules
|
||||
if len(tag) > 50: # Reasonable length limit
|
||||
return False
|
||||
|
||||
if tag.startswith('/') or tag.endswith('/'):
|
||||
return False
|
||||
|
||||
if '//' in tag: # No empty hierarchy levels
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _apply_consolidation_rules(self, tags: List[str]) -> List[str]:
|
||||
"""Apply tag consolidation rules to remove redundancy."""
|
||||
consolidated = tags.copy()
|
||||
|
||||
# Remove redundant hierarchical tags
|
||||
# If we have both 'tech' and 'tech/ai', keep only 'tech/ai'
|
||||
hierarchical_tags = [tag for tag in consolidated if '/' in tag]
|
||||
simple_tags = [tag for tag in consolidated if '/' not in tag]
|
||||
|
||||
# Remove simple tags that are covered by hierarchical tags
|
||||
filtered_simple = []
|
||||
for simple_tag in simple_tags:
|
||||
is_covered = any(hier_tag.startswith(f"{simple_tag}/") for hier_tag in hierarchical_tags)
|
||||
if not is_covered:
|
||||
filtered_simple.append(simple_tag)
|
||||
|
||||
return filtered_simple + hierarchical_tags
|
||||
|
||||
def _sort_tags(self, tags: List[str]) -> List[str]:
|
||||
"""Sort tags with hierarchical tags first, then alphabetical."""
|
||||
hierarchical = [tag for tag in tags if '/' in tag]
|
||||
simple = [tag for tag in tags if '/' not in tag]
|
||||
|
||||
# Sort hierarchical tags by depth then alphabetically
|
||||
hierarchical.sort(key=lambda x: (x.count('/'), x))
|
||||
simple.sort()
|
||||
|
||||
return hierarchical + simple
|
||||
Reference in New Issue
Block a user