1130 lines
37 KiB
Python
1130 lines
37 KiB
Python
"""
|
|
Utility functions for file operations, search, and markdown processing
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional, Tuple, Any, TypeVar, Callable
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
# ============================================================================
|
|
# Pagination Support
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class PaginationResult:
|
|
"""
|
|
Result of applying pagination to a list.
|
|
|
|
Attributes:
|
|
items: The paginated subset of items
|
|
total: Total number of items before pagination
|
|
limit: The limit that was applied (None if no pagination)
|
|
offset: The offset that was applied
|
|
has_more: Whether there are more items after this page
|
|
"""
|
|
items: List[Any]
|
|
total: int
|
|
limit: Optional[int]
|
|
offset: int
|
|
has_more: bool
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert pagination info to dict for API response."""
|
|
return {
|
|
"limit": self.limit,
|
|
"offset": self.offset,
|
|
"total": self.total,
|
|
"has_more": self.has_more
|
|
}
|
|
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
def paginate(
|
|
items: List[T],
|
|
limit: Optional[int] = None,
|
|
offset: int = 0,
|
|
sort_key: Optional[Callable[[T], Any]] = None,
|
|
sort_reverse: bool = False
|
|
) -> PaginationResult:
|
|
"""
|
|
Apply optional pagination to a list with consistent sorting.
|
|
|
|
This function is designed to be backward-compatible:
|
|
- If limit is None, returns all items (no pagination)
|
|
- If limit is provided, returns a paginated subset
|
|
|
|
Sorting is always applied (when sort_key is provided) to ensure
|
|
stable pagination across requests.
|
|
|
|
Args:
|
|
items: List of items to paginate
|
|
limit: Maximum number of items to return (None = no limit)
|
|
offset: Number of items to skip (default: 0)
|
|
sort_key: Function to extract sort key from item (e.g., lambda x: x['path'])
|
|
sort_reverse: If True, sort in descending order
|
|
|
|
Returns:
|
|
PaginationResult with items and pagination metadata
|
|
|
|
Example:
|
|
# No pagination (frontend compatibility)
|
|
result = paginate(notes)
|
|
|
|
# With pagination (MCP usage)
|
|
result = paginate(notes, limit=20, offset=0, sort_key=lambda x: x['path'])
|
|
"""
|
|
total = len(items)
|
|
|
|
# Apply sorting for consistent ordering (prevents out-of-order issues)
|
|
if sort_key is not None:
|
|
items = sorted(items, key=sort_key, reverse=sort_reverse)
|
|
|
|
# Apply pagination only if limit is specified
|
|
if limit is not None:
|
|
# Clamp offset to valid range
|
|
offset = max(0, min(offset, total))
|
|
end = offset + limit
|
|
paginated_items = items[offset:end]
|
|
has_more = end < total
|
|
else:
|
|
# No pagination - return all items
|
|
paginated_items = items
|
|
offset = 0
|
|
has_more = False
|
|
|
|
return PaginationResult(
|
|
items=paginated_items,
|
|
total=total,
|
|
limit=limit,
|
|
offset=offset,
|
|
has_more=has_more
|
|
)
|
|
|
|
|
|
# In-memory cache for parsed tags
|
|
# Format: {file_path: (mtime, tags)}
|
|
_tag_cache: Dict[str, Tuple[float, List[str]]] = {}
|
|
|
|
# Notes tree scan cache (TTL).
|
|
#
|
|
# This avoids repeated full-directory walks when multiple endpoints (or the UI)
|
|
# request indexes in quick succession.
|
|
|
|
_SCAN_WALK_CACHE_LOCK = threading.Lock()
|
|
_SCAN_WALK_CACHE_TTL_SECONDS = 1.0
|
|
# key: (resolved_notes_dir, include_media) -> (cached_at_monotonic_seconds, (notes, folders))
|
|
_SCAN_WALK_CACHE: Dict[Tuple[str, bool], Tuple[float, Tuple[List[Dict], List[str]]]] = {}
|
|
|
|
|
|
def _scan_cache_get(key: Tuple[str, bool]) -> Optional[Tuple[List[Dict], List[str]]]:
|
|
now = time.monotonic()
|
|
with _SCAN_WALK_CACHE_LOCK:
|
|
entry = _SCAN_WALK_CACHE.get(key)
|
|
if not entry:
|
|
return None
|
|
cached_at, value = entry
|
|
if (now - cached_at) > _SCAN_WALK_CACHE_TTL_SECONDS:
|
|
_SCAN_WALK_CACHE.pop(key, None)
|
|
return None
|
|
return value
|
|
|
|
|
|
def _scan_cache_set(key: Tuple[str, bool], value: Tuple[List[Dict], List[str]]) -> None:
|
|
with _SCAN_WALK_CACHE_LOCK:
|
|
_SCAN_WALK_CACHE[key] = (time.monotonic(), value)
|
|
|
|
|
|
def validate_path_security(notes_dir: str, path: Path) -> bool:
|
|
"""
|
|
Validate that a path is within the notes directory (security check).
|
|
Prevents path traversal attacks.
|
|
|
|
Args:
|
|
notes_dir: Base notes directory
|
|
path: Path to validate
|
|
|
|
Returns:
|
|
True if path is safe, False otherwise
|
|
"""
|
|
try:
|
|
path.resolve().relative_to(Path(notes_dir).resolve())
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def ensure_directories(config: dict):
|
|
"""Create necessary directories if they don't exist"""
|
|
dirs = [
|
|
config['storage']['notes_dir'],
|
|
config['storage']['plugins_dir'],
|
|
]
|
|
|
|
for dir_path in dirs:
|
|
Path(dir_path).mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def create_folder(notes_dir: str, folder_path: str) -> bool:
|
|
"""Create a new folder in the notes directory"""
|
|
full_path = Path(notes_dir) / folder_path
|
|
|
|
# Security check
|
|
if not validate_path_security(notes_dir, full_path):
|
|
return False
|
|
|
|
full_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
return True
|
|
|
|
|
|
def scan_notes_fast_walk(notes_dir: str, use_cache: bool = True, include_media: bool = False) -> Tuple[List[Dict], List[str]]:
|
|
"""Fast scanner using os.walk (pure Python + stdlib).
|
|
|
|
Args:
|
|
notes_dir: Base notes directory
|
|
"""
|
|
notes_path = Path(notes_dir)
|
|
|
|
cache_key = (str(notes_path.resolve()), include_media)
|
|
if use_cache:
|
|
cached = _scan_cache_get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
if not include_media:
|
|
media_cache_key = (str(notes_path.resolve()), True)
|
|
media_cached = _scan_cache_get(media_cache_key)
|
|
if media_cached is not None:
|
|
media_notes, media_folders = media_cached
|
|
normalized_notes = []
|
|
for note in media_notes:
|
|
if not Path(note.get("path", "")).match("*.md"):
|
|
continue
|
|
normalized_note = dict(note)
|
|
normalized_note["type"] = "note"
|
|
normalized_notes.append(normalized_note)
|
|
|
|
normalized_value = (normalized_notes, media_folders)
|
|
_scan_cache_set(cache_key, normalized_value)
|
|
return normalized_value
|
|
|
|
notes: List[Dict] = []
|
|
folders_set = set()
|
|
|
|
for root, dirnames, filenames in os.walk(notes_path):
|
|
# Skip descending into dot-dirs
|
|
dirnames[:] = [d for d in dirnames if not d.startswith('.')]
|
|
|
|
root_path = Path(root)
|
|
rel_folder = root_path.relative_to(notes_path).as_posix()
|
|
if rel_folder != "." and not rel_folder.startswith('.'):
|
|
folders_set.add(rel_folder)
|
|
|
|
for filename in filenames:
|
|
if filename.startswith('.'):
|
|
continue
|
|
|
|
full_path = root_path / filename
|
|
try:
|
|
st = full_path.stat()
|
|
except OSError:
|
|
continue
|
|
|
|
relative_path = full_path.relative_to(notes_path)
|
|
media_type = get_media_type(filename) if include_media else None
|
|
is_markdown = full_path.suffix.lower() == '.md'
|
|
should_include = is_markdown or (include_media and media_type is not None)
|
|
|
|
if not should_include:
|
|
continue
|
|
|
|
folder = relative_path.parent.as_posix()
|
|
# Get tags for this note (cached)
|
|
tags = get_tags_cached(full_path) if is_markdown else []
|
|
notes.append({
|
|
"name": full_path.stem,
|
|
"path": relative_path.as_posix(),
|
|
"folder": "" if folder == "." else folder,
|
|
"modified": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
|
|
"size": st.st_size,
|
|
"type": media_type if media_type else "note",
|
|
"tags": tags,
|
|
})
|
|
|
|
value = (sorted(notes, key=lambda x: x.get('modified', ''), reverse=True), sorted(folders_set))
|
|
if use_cache:
|
|
_scan_cache_set(cache_key, value)
|
|
return value
|
|
|
|
def move_note(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]:
|
|
"""Move a note to a different location
|
|
|
|
Returns:
|
|
Tuple of (success: bool, error_message: str)
|
|
"""
|
|
old_full_path = Path(notes_dir) / old_path
|
|
new_full_path = Path(notes_dir) / new_path
|
|
|
|
# Security checks
|
|
if not validate_path_security(notes_dir, old_full_path):
|
|
return False, "Invalid source path"
|
|
if not validate_path_security(notes_dir, new_full_path):
|
|
return False, "Invalid destination path"
|
|
|
|
if not old_full_path.exists():
|
|
return False, f"Source note does not exist: {old_path}"
|
|
|
|
# Check if target already exists (prevent overwriting)
|
|
if new_full_path.exists():
|
|
return False, f"A note already exists at: {new_path}"
|
|
|
|
# Invalidate cache for old path
|
|
old_key = str(old_full_path)
|
|
if old_key in _tag_cache:
|
|
del _tag_cache[old_key]
|
|
|
|
try:
|
|
# Create parent directory if needed
|
|
new_full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Move the file
|
|
old_full_path.rename(new_full_path)
|
|
except Exception as e:
|
|
return False, f"Failed to move file: {str(e)}"
|
|
|
|
# Note: We don't automatically delete empty folders to preserve user's folder structure
|
|
|
|
return True, ""
|
|
|
|
|
|
def move_folder(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]:
|
|
"""Move a folder to a different location
|
|
|
|
Returns:
|
|
Tuple of (success: bool, error_message: str)
|
|
"""
|
|
import shutil
|
|
|
|
old_full_path = Path(notes_dir) / old_path
|
|
new_full_path = Path(notes_dir) / new_path
|
|
|
|
# Security checks
|
|
if not validate_path_security(notes_dir, old_full_path):
|
|
return False, "Invalid source path"
|
|
if not validate_path_security(notes_dir, new_full_path):
|
|
return False, "Invalid destination path"
|
|
|
|
if not old_full_path.exists() or not old_full_path.is_dir():
|
|
return False, f"Source folder does not exist: {old_path}"
|
|
|
|
# Check if target already exists
|
|
if new_full_path.exists():
|
|
return False, f"A folder already exists at: {new_path}"
|
|
|
|
# Invalidate cache for all notes in this folder
|
|
global _tag_cache
|
|
old_path_str = str(old_full_path)
|
|
keys_to_delete = [key for key in _tag_cache.keys() if key.startswith(old_path_str)]
|
|
for key in keys_to_delete:
|
|
del _tag_cache[key]
|
|
|
|
try:
|
|
# Create parent directory if needed
|
|
new_full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Move the folder
|
|
shutil.move(str(old_full_path), str(new_full_path))
|
|
except Exception as e:
|
|
return False, f"Failed to move folder: {str(e)}"
|
|
|
|
# Note: We don't automatically delete empty folders to preserve user's folder structure
|
|
|
|
return True, ""
|
|
|
|
|
|
def rename_folder(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]:
|
|
"""Rename a folder (same as move but for clarity)"""
|
|
return move_folder(notes_dir, old_path, new_path)
|
|
|
|
|
|
def delete_folder(notes_dir: str, folder_path: str) -> bool:
|
|
"""Delete a folder and all its contents"""
|
|
try:
|
|
full_path = Path(notes_dir) / folder_path
|
|
|
|
# Security check: ensure the path is within notes_dir
|
|
if not validate_path_security(notes_dir, full_path):
|
|
print(f"Security: Path is outside notes directory: {full_path}")
|
|
return False
|
|
|
|
if not full_path.exists():
|
|
print(f"Folder does not exist: {full_path}")
|
|
return False
|
|
|
|
if not full_path.is_dir():
|
|
print(f"Path is not a directory: {full_path}")
|
|
return False
|
|
|
|
# Invalidate cache for all notes in this folder
|
|
global _tag_cache
|
|
folder_path_str = str(full_path)
|
|
keys_to_delete = [key for key in _tag_cache.keys() if key.startswith(folder_path_str)]
|
|
for key in keys_to_delete:
|
|
del _tag_cache[key]
|
|
|
|
# Delete the folder and all its contents
|
|
shutil.rmtree(full_path)
|
|
print(f"Successfully deleted folder: {full_path}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error deleting folder '{folder_path}': {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
|
|
|
|
def get_note_content(notes_dir: str, note_path: str) -> Optional[str]:
|
|
"""Get the content of a specific note"""
|
|
full_path = Path(notes_dir) / note_path
|
|
|
|
if not full_path.exists() or not full_path.is_file():
|
|
return None
|
|
|
|
# Security check: ensure the path is within notes_dir
|
|
if not validate_path_security(notes_dir, full_path):
|
|
return None
|
|
|
|
with open(full_path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
|
|
|
|
def save_note(notes_dir: str, note_path: str, content: str) -> bool:
|
|
"""Save or update a note"""
|
|
full_path = Path(notes_dir) / note_path
|
|
|
|
# Ensure .md extension
|
|
if not note_path.endswith('.md'):
|
|
full_path = full_path.with_suffix('.md')
|
|
|
|
# Security check
|
|
if not validate_path_security(notes_dir, full_path):
|
|
return False
|
|
|
|
# Create parent directories if needed
|
|
full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(full_path, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
|
|
return True
|
|
|
|
|
|
def delete_note(notes_dir: str, note_path: str) -> bool:
|
|
"""Delete a note"""
|
|
full_path = Path(notes_dir) / note_path
|
|
|
|
if not full_path.exists():
|
|
return False
|
|
|
|
# Security check
|
|
if not validate_path_security(notes_dir, full_path):
|
|
return False
|
|
|
|
# Invalidate cache for this note
|
|
file_key = str(full_path)
|
|
if file_key in _tag_cache:
|
|
del _tag_cache[file_key]
|
|
|
|
full_path.unlink()
|
|
|
|
# Note: We don't automatically delete empty folders to preserve user's folder structure
|
|
|
|
return True
|
|
|
|
|
|
def search_notes(notes_dir: str, query: str) -> List[Dict]:
|
|
"""
|
|
Full-text search through note contents only.
|
|
Does NOT search in file names, folder names, or paths - only note content.
|
|
Uses character-based context extraction with highlighted matches.
|
|
"""
|
|
from html import escape
|
|
results = []
|
|
notes, _folders = scan_notes_fast_walk(notes_dir, include_media=False)
|
|
|
|
for note in notes:
|
|
md_file = Path(notes_dir) / note["path"]
|
|
try:
|
|
with open(md_file, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# Find all matches using regex (case-insensitive)
|
|
matches = list(re.finditer(re.escape(query), content, re.IGNORECASE))
|
|
|
|
if matches:
|
|
matched_lines = []
|
|
|
|
for match in matches[:3]: # Limit to 3 matches per file
|
|
start_index = match.start()
|
|
end_index = match.end()
|
|
matched_text = match.group() # Preserve original case
|
|
|
|
# Create slice window: ±15 characters around match
|
|
context_start = max(0, start_index - 15)
|
|
context_end = min(len(content), end_index + 15)
|
|
|
|
# Extract and clean parts (newlines → spaces)
|
|
before = escape(content[context_start:start_index].replace('\n', ' '))
|
|
after = escape(content[end_index:context_end].replace('\n', ' '))
|
|
matched_clean = escape(matched_text.replace('\n', ' '))
|
|
|
|
# Build snippet with <mark> highlight (styled via CSS)
|
|
snippet = f'{before}<mark class="search-highlight">{matched_clean}</mark>{after}'
|
|
|
|
# Add ellipsis if truncated at start
|
|
if context_start > 0:
|
|
snippet = '...' + snippet
|
|
|
|
# Add ellipsis if truncated at end
|
|
if context_end < len(content):
|
|
snippet = snippet + '...'
|
|
|
|
# Calculate line number by counting newlines up to match start
|
|
line_number = content.count('\n', 0, start_index) + 1
|
|
|
|
matched_lines.append({
|
|
"line_number": line_number,
|
|
"context": snippet
|
|
})
|
|
|
|
relative_path = Path(note["path"])
|
|
results.append({
|
|
"name": md_file.stem,
|
|
"path": str(relative_path.as_posix()),
|
|
"folder": str(relative_path.parent.as_posix()) if str(relative_path.parent) != "." else "",
|
|
"matches": matched_lines
|
|
})
|
|
except Exception:
|
|
continue
|
|
|
|
return results
|
|
|
|
|
|
def create_note_metadata(notes_dir: str, note_path: str) -> Dict:
|
|
"""Get metadata for a note"""
|
|
full_path = Path(notes_dir) / note_path
|
|
|
|
if not full_path.exists() or not full_path.is_file():
|
|
return {}
|
|
|
|
# Security check: ensure the path is within notes_dir (same as get_note_content)
|
|
if not validate_path_security(notes_dir, full_path):
|
|
return {}
|
|
|
|
stat = full_path.stat()
|
|
|
|
# Count lines with proper file handle management
|
|
with open(full_path, 'r', encoding='utf-8') as f:
|
|
line_count = sum(1 for _ in f)
|
|
|
|
return {
|
|
"created": datetime.fromtimestamp(stat.st_ctime, tz=timezone.utc).isoformat(),
|
|
"modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
|
"size": stat.st_size,
|
|
"lines": line_count
|
|
}
|
|
|
|
|
|
def sanitize_filename(filename: str) -> str:
|
|
"""
|
|
Sanitize a filename by removing/replacing dangerous filesystem characters.
|
|
Supports Unicode characters (international text) while blocking:
|
|
- Windows forbidden: \\ / : * ? " < > |
|
|
- Control characters (0x00-0x1f)
|
|
|
|
Note: This is a safety net - the frontend validates before sending.
|
|
"""
|
|
if not filename:
|
|
return filename
|
|
|
|
# Get the extension first
|
|
parts = filename.rsplit('.', 1)
|
|
name = parts[0]
|
|
ext = parts[1] if len(parts) > 1 else ''
|
|
|
|
# Remove dangerous characters (replace with underscore)
|
|
# Blocklist approach: only remove what's truly dangerous
|
|
# Pattern: backslash, forward slash, colon, asterisk, question mark, quotes, angle brackets, pipe, control chars
|
|
name = re.sub(r'[\\/:*?"<>|\x00-\x1f]', '_', name)
|
|
|
|
# Collapse multiple underscores
|
|
name = re.sub(r'_+', '_', name)
|
|
|
|
# Strip leading/trailing underscores and spaces
|
|
name = name.strip('_ ')
|
|
|
|
# Ensure we have something left
|
|
if not name:
|
|
name = 'unnamed'
|
|
|
|
# Rejoin with extension
|
|
return f"{name}.{ext}" if ext else name
|
|
|
|
|
|
def get_attachment_dir(notes_dir: str, note_path: str) -> Path:
|
|
"""
|
|
Get the attachments directory for a given note.
|
|
If note is in root, returns /data/_attachments/
|
|
If note is in folder, returns /data/folder/_attachments/
|
|
"""
|
|
if not note_path:
|
|
# Root level
|
|
return Path(notes_dir) / "_attachments"
|
|
|
|
note_path_obj = Path(note_path)
|
|
folder = note_path_obj.parent
|
|
|
|
if str(folder) == '.':
|
|
# Note is in root
|
|
return Path(notes_dir) / "_attachments"
|
|
else:
|
|
# Note is in a folder
|
|
return Path(notes_dir) / folder / "_attachments"
|
|
|
|
|
|
def save_uploaded_image(notes_dir: str, note_path: str, filename: str, file_data: bytes) -> Optional[str]:
|
|
"""
|
|
Save an uploaded image to the appropriate attachments directory.
|
|
Returns the relative path to the image if successful, None otherwise.
|
|
|
|
Args:
|
|
notes_dir: Base notes directory
|
|
note_path: Path of the note the image is being uploaded to
|
|
filename: Original filename
|
|
file_data: Binary file data
|
|
|
|
Returns:
|
|
Relative path to the saved image, or None if failed
|
|
"""
|
|
# Sanitize filename
|
|
sanitized_name = sanitize_filename(filename)
|
|
|
|
# Get extension
|
|
ext = Path(sanitized_name).suffix
|
|
name_without_ext = Path(sanitized_name).stem
|
|
|
|
# Add timestamp to prevent collisions
|
|
timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
|
|
final_filename = f"{name_without_ext}-{timestamp}{ext}"
|
|
|
|
# Get attachments directory
|
|
attachments_dir = get_attachment_dir(notes_dir, note_path)
|
|
|
|
# Create directory if it doesn't exist
|
|
attachments_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Full path to save the image
|
|
full_path = attachments_dir / final_filename
|
|
|
|
# Security check
|
|
if not validate_path_security(notes_dir, full_path):
|
|
print(f"Security: Attempted to save image outside notes directory: {full_path}")
|
|
return None
|
|
|
|
try:
|
|
# Write the file
|
|
with open(full_path, 'wb') as f:
|
|
f.write(file_data)
|
|
|
|
# Return relative path from notes_dir
|
|
relative_path = full_path.relative_to(Path(notes_dir))
|
|
return str(relative_path.as_posix())
|
|
except Exception as e:
|
|
print(f"Error saving image: {e}")
|
|
return None
|
|
|
|
|
|
# Media file type definitions
|
|
MEDIA_EXTENSIONS = {
|
|
'image': {'.jpg', '.jpeg', '.png', '.gif', '.webp'},
|
|
'audio': {'.mp3', '.wav', '.ogg', '.m4a'},
|
|
'video': {'.mp4', '.webm', '.mov', '.avi'},
|
|
'document': {'.pdf'},
|
|
}
|
|
|
|
# All supported media extensions (flat set for quick lookup)
|
|
ALL_MEDIA_EXTENSIONS = set().union(*MEDIA_EXTENSIONS.values())
|
|
|
|
|
|
def get_media_type(filename: str) -> Optional[str]:
|
|
"""
|
|
Determine the media type based on file extension.
|
|
Returns: 'image', 'audio', 'video', 'document', or None if not a media file.
|
|
"""
|
|
ext = Path(filename).suffix.lower()
|
|
for media_type, extensions in MEDIA_EXTENSIONS.items():
|
|
if ext in extensions:
|
|
return media_type
|
|
return None
|
|
|
|
|
|
def parse_tags(content: str) -> List[str]:
|
|
"""
|
|
Extract tags from YAML frontmatter in markdown content.
|
|
|
|
Supported formats:
|
|
---
|
|
tags: [python, tutorial, backend]
|
|
---
|
|
|
|
or
|
|
|
|
---
|
|
tags:
|
|
- python
|
|
- tutorial
|
|
- backend
|
|
---
|
|
|
|
Args:
|
|
content: Markdown content with optional YAML frontmatter
|
|
|
|
Returns:
|
|
List of tag strings (lowercase, no duplicates)
|
|
"""
|
|
tags = []
|
|
|
|
# Check if content starts with frontmatter
|
|
if not content.strip().startswith('---'):
|
|
return tags
|
|
|
|
try:
|
|
# Extract frontmatter (between first two --- markers)
|
|
lines = content.split('\n')
|
|
if lines[0].strip() != '---':
|
|
return tags
|
|
|
|
# Find closing ---
|
|
end_idx = None
|
|
for i in range(1, len(lines)):
|
|
if lines[i].strip() == '---':
|
|
end_idx = i
|
|
break
|
|
|
|
if end_idx is None:
|
|
return tags
|
|
|
|
frontmatter_lines = lines[1:end_idx]
|
|
|
|
# Parse tags field
|
|
in_tags_list = False
|
|
for line in frontmatter_lines:
|
|
stripped = line.strip()
|
|
|
|
# Check for inline array format: tags: [tag1, tag2, tag3]
|
|
if stripped.startswith('tags:'):
|
|
rest = stripped[5:].strip()
|
|
if rest.startswith('[') and rest.endswith(']'):
|
|
# Parse inline array
|
|
tags_str = rest[1:-1] # Remove [ and ]
|
|
raw_tags = [t.strip() for t in tags_str.split(',')]
|
|
tags.extend([t.lower() for t in raw_tags if t])
|
|
break
|
|
elif rest:
|
|
# Single tag without brackets
|
|
tags.append(rest.lower())
|
|
break
|
|
else:
|
|
# Multi-line list format
|
|
in_tags_list = True
|
|
elif in_tags_list:
|
|
if stripped.startswith('-'):
|
|
# List item
|
|
tag = stripped[1:].strip()
|
|
if tag:
|
|
tags.append(tag.lower())
|
|
elif stripped and not stripped.startswith('#'):
|
|
# End of tags list
|
|
break
|
|
|
|
# Remove duplicates and return
|
|
return sorted(list(set(tags)))
|
|
|
|
except Exception as e:
|
|
# If parsing fails, return empty list
|
|
print(f"Error parsing tags: {e}")
|
|
return []
|
|
|
|
|
|
def get_tags_cached(file_path: Path) -> List[str]:
|
|
"""
|
|
Get tags for a file with caching based on modification time.
|
|
|
|
Args:
|
|
file_path: Path to the markdown file
|
|
|
|
Returns:
|
|
List of tags from the file (cached if mtime unchanged)
|
|
"""
|
|
global _tag_cache
|
|
|
|
try:
|
|
# Get current modification time
|
|
mtime = file_path.stat().st_mtime
|
|
file_key = str(file_path)
|
|
|
|
# Check cache
|
|
if file_key in _tag_cache:
|
|
cached_mtime, cached_tags = _tag_cache[file_key]
|
|
if cached_mtime == mtime:
|
|
# Cache hit! Return cached tags
|
|
return cached_tags
|
|
|
|
# Cache miss or stale - parse tags
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
tags = parse_tags(content)
|
|
|
|
# Update cache
|
|
_tag_cache[file_key] = (mtime, tags)
|
|
return tags
|
|
|
|
except Exception:
|
|
# If anything fails, return empty list
|
|
return []
|
|
|
|
|
|
def clear_tag_cache():
|
|
"""Clear the tag cache (useful for testing or manual cache invalidation)"""
|
|
global _tag_cache
|
|
_tag_cache.clear()
|
|
|
|
|
|
def get_all_tags(notes_dir: str) -> Dict[str, int]:
|
|
"""
|
|
Get all tags used across all notes with their count (cached).
|
|
|
|
Args:
|
|
notes_dir: Directory containing notes
|
|
|
|
Returns:
|
|
Dictionary mapping tag names to note counts
|
|
"""
|
|
tag_counts = {}
|
|
notes, _folders = scan_notes_fast_walk(notes_dir, include_media=False)
|
|
|
|
for note in notes:
|
|
md_file = Path(notes_dir) / note["path"]
|
|
# Get tags using cache
|
|
tags = get_tags_cached(md_file)
|
|
|
|
for tag in tags:
|
|
tag_counts[tag] = tag_counts.get(tag, 0) + 1
|
|
|
|
return dict(sorted(tag_counts.items()))
|
|
|
|
|
|
def get_notes_by_tag(notes_dir: str, tag: str) -> List[Dict]:
|
|
"""
|
|
Get all notes that have a specific tag (cached).
|
|
|
|
Args:
|
|
notes_dir: Directory containing notes
|
|
tag: Tag to filter by (case-insensitive)
|
|
|
|
Returns:
|
|
List of note dictionaries matching the tag
|
|
"""
|
|
matching_notes = []
|
|
tag_lower = tag.lower()
|
|
notes, _folders = scan_notes_fast_walk(notes_dir, include_media=False)
|
|
|
|
for note in notes:
|
|
md_file = Path(notes_dir) / note["path"]
|
|
# Get tags using cache
|
|
tags = get_tags_cached(md_file)
|
|
|
|
if tag_lower in tags:
|
|
matching_notes.append({
|
|
"name": note["name"],
|
|
"path": note["path"],
|
|
"folder": note["folder"],
|
|
"modified": note["modified"],
|
|
"size": note["size"],
|
|
"tags": tags
|
|
})
|
|
|
|
return matching_notes
|
|
|
|
|
|
# ============================================================================
|
|
# Template Functions
|
|
# ============================================================================
|
|
|
|
def get_templates(notes_dir: str) -> List[Dict]:
|
|
"""
|
|
Get all templates from the _templates folder.
|
|
|
|
Args:
|
|
notes_dir: Base notes directory
|
|
|
|
Returns:
|
|
List of template metadata (name, path, modified)
|
|
"""
|
|
templates = []
|
|
templates_path = Path(notes_dir) / "_templates"
|
|
|
|
if not templates_path.exists():
|
|
return templates
|
|
|
|
# Security check: ensure _templates folder is within notes directory
|
|
if not validate_path_security(notes_dir, templates_path):
|
|
print(f"Security: Templates directory is outside notes directory: {templates_path}")
|
|
return templates
|
|
|
|
try:
|
|
for template_file in templates_path.glob("*.md"):
|
|
try:
|
|
# Security check: ensure each template is within notes directory
|
|
if not validate_path_security(notes_dir, template_file):
|
|
print(f"Security: Skipping template outside notes directory: {template_file}")
|
|
continue
|
|
|
|
stat = template_file.stat()
|
|
templates.append({
|
|
"name": template_file.stem,
|
|
"path": str(template_file.relative_to(notes_dir).as_posix()),
|
|
"modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
|
|
})
|
|
except Exception as e:
|
|
print(f"Error reading template {template_file}: {e}")
|
|
continue
|
|
except Exception as e:
|
|
print(f"Error accessing templates directory: {e}")
|
|
|
|
return sorted(templates, key=lambda x: x['name'])
|
|
|
|
|
|
def get_template_content(notes_dir: str, template_name: str) -> Optional[str]:
|
|
"""
|
|
Get the content of a specific template.
|
|
|
|
Args:
|
|
notes_dir: Base notes directory
|
|
template_name: Name of the template (without .md extension)
|
|
|
|
Returns:
|
|
Template content or None if not found
|
|
"""
|
|
template_path = Path(notes_dir) / "_templates" / f"{template_name}.md"
|
|
|
|
if not template_path.exists():
|
|
return None
|
|
|
|
# Security check: ensure template is within notes directory
|
|
if not validate_path_security(notes_dir, template_path):
|
|
print(f"Security: Template path is outside notes directory: {template_path}")
|
|
return None
|
|
|
|
try:
|
|
with open(template_path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
except Exception as e:
|
|
print(f"Error reading template {template_name}: {e}")
|
|
return None
|
|
|
|
|
|
def apply_template_placeholders(content: str, note_path: str) -> str:
|
|
"""
|
|
Replace template placeholders with actual values.
|
|
|
|
Supported placeholders:
|
|
{{date}} - Current date (YYYY-MM-DD)
|
|
{{time}} - Current time (HH:MM:SS)
|
|
{{datetime}} - Current datetime (YYYY-MM-DD HH:MM:SS)
|
|
{{timestamp}} - Unix timestamp
|
|
{{year}} - Current year (YYYY)
|
|
{{month}} - Current month (MM)
|
|
{{day}} - Current day (DD)
|
|
{{title}} - Note name without extension
|
|
{{folder}} - Parent folder name
|
|
|
|
Args:
|
|
content: Template content with placeholders
|
|
note_path: Path of the note being created
|
|
|
|
Returns:
|
|
Content with placeholders replaced
|
|
"""
|
|
now = datetime.now()
|
|
note = Path(note_path)
|
|
|
|
replacements = {
|
|
'{{date}}': now.strftime('%Y-%m-%d'),
|
|
'{{time}}': now.strftime('%H:%M:%S'),
|
|
'{{datetime}}': now.strftime('%Y-%m-%d %H:%M:%S'),
|
|
'{{timestamp}}': str(int(now.timestamp())),
|
|
'{{year}}': now.strftime('%Y'),
|
|
'{{month}}': now.strftime('%m'),
|
|
'{{day}}': now.strftime('%d'),
|
|
'{{title}}': note.stem,
|
|
'{{folder}}': note.parent.name if str(note.parent) != '.' else 'Root',
|
|
}
|
|
|
|
result = content
|
|
for placeholder, value in replacements.items():
|
|
result = result.replace(placeholder, value)
|
|
|
|
return result
|
|
|
|
|
|
def get_backlinks(notes_dir: str, target_note_path: str) -> List[Dict]:
|
|
"""
|
|
Find all notes that link TO the specified note (reverse links / backlinks).
|
|
|
|
Args:
|
|
notes_dir: Base directory containing notes
|
|
target_note_path: Path of the note to find backlinks for
|
|
|
|
Returns:
|
|
List of backlink objects with path, context, and line_number
|
|
"""
|
|
backlinks = []
|
|
notes, _folders = scan_notes_fast_walk(notes_dir, include_media=False)
|
|
|
|
# Normalize target path for matching
|
|
target_path = target_note_path
|
|
target_path_lower = target_path.lower()
|
|
target_path_no_ext = target_path.replace('.md', '')
|
|
target_path_no_ext_lower = target_path_no_ext.lower()
|
|
target_name = Path(target_path).stem.lower()
|
|
|
|
# For wikilinks: global name matching (find note anywhere by name)
|
|
wikilink_refs = {
|
|
target_path_lower,
|
|
target_path_no_ext_lower,
|
|
target_name,
|
|
}
|
|
|
|
for note in notes:
|
|
if note.get('type') != 'note':
|
|
continue
|
|
|
|
source_path = note['path']
|
|
|
|
# Skip self-references
|
|
if source_path == target_path:
|
|
continue
|
|
|
|
# Get source folder for resolving relative markdown links
|
|
source_folder = str(Path(source_path).parent).replace('\\', '/')
|
|
if source_folder == '.':
|
|
source_folder = ''
|
|
|
|
# Read note content
|
|
full_path = Path(notes_dir) / source_path
|
|
try:
|
|
with open(full_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
except Exception:
|
|
continue
|
|
|
|
lines = content.split('\n')
|
|
found_links = []
|
|
|
|
for line_num, line in enumerate(lines, 1):
|
|
# Find wikilinks: [[target]] or [[target|display]]
|
|
# Wikilinks use GLOBAL matching (find note anywhere by name)
|
|
wikilink_matches = re.finditer(r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]', line)
|
|
for match in wikilink_matches:
|
|
link_target = match.group(1).strip().lower()
|
|
link_target_no_ext = link_target.replace('.md', '')
|
|
|
|
# Check if this wikilink points to our target (global match)
|
|
if link_target in wikilink_refs or link_target_no_ext in wikilink_refs:
|
|
start = max(0, match.start() - 30)
|
|
end = min(len(line), match.end() + 30)
|
|
context = line[start:end]
|
|
if start > 0:
|
|
context = '...' + context
|
|
if end < len(line):
|
|
context = context + '...'
|
|
|
|
found_links.append({
|
|
"line_number": line_num,
|
|
"context": context,
|
|
"type": "wikilink"
|
|
})
|
|
|
|
# Find markdown links: [text](path)
|
|
# Markdown links must RESOLVE as paths (relative to source or absolute)
|
|
markdown_matches = re.finditer(r'\[([^\]]+)\]\((?!https?://|mailto:|#|data:)([^\)]+)\)', line)
|
|
for match in markdown_matches:
|
|
link_path = match.group(2).split('#')[0] # Remove anchor
|
|
if not link_path:
|
|
continue
|
|
|
|
link_path = urllib.parse.unquote(link_path)
|
|
if link_path.startswith('./'):
|
|
link_path = link_path[2:]
|
|
|
|
# Add .md if not present
|
|
link_path_with_md = link_path if link_path.endswith('.md') else link_path + '.md'
|
|
|
|
# Resolve the link path to get the actual target
|
|
resolved_path = None
|
|
|
|
# 1. Try resolving relative to source folder
|
|
if source_folder and not link_path.startswith('/'):
|
|
relative_path = f"{source_folder}/{link_path_with_md}"
|
|
if relative_path.lower() == target_path_lower:
|
|
resolved_path = target_path
|
|
elif f"{source_folder}/{link_path}".lower() == target_path_no_ext_lower:
|
|
resolved_path = target_path
|
|
|
|
# 2. Try as absolute path from root
|
|
if not resolved_path:
|
|
if link_path_with_md.lower() == target_path_lower:
|
|
resolved_path = target_path
|
|
elif link_path.lower() == target_path_no_ext_lower:
|
|
resolved_path = target_path
|
|
|
|
# Only add if the resolved path matches the target
|
|
if resolved_path:
|
|
start = max(0, match.start() - 30)
|
|
end = min(len(line), match.end() + 30)
|
|
context = line[start:end]
|
|
if start > 0:
|
|
context = '...' + context
|
|
if end < len(line):
|
|
context = context + '...'
|
|
|
|
found_links.append({
|
|
"line_number": line_num,
|
|
"context": context,
|
|
"type": "markdown"
|
|
})
|
|
|
|
# If we found links in this note, add it to backlinks
|
|
if found_links:
|
|
backlinks.append({
|
|
"path": source_path,
|
|
"name": note['name'].replace('.md', ''),
|
|
"references": found_links[:3] # Limit to 3 references per note
|
|
})
|
|
|
|
return backlinks
|
|
|