""" Utility functions for file operations, search, and markdown processing """ import os import re import shutil import threading import time from pathlib import Path from typing import List, Dict, Optional, Tuple from datetime import datetime, timezone # 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, file_filter) -> (cached_at_monotonic_seconds, (notes, folders)) _SCAN_WALK_CACHE: Dict[Tuple[str, str], Tuple[float, Tuple[List[Dict], List[str]]]] = {} def _scan_cache_get(key: Tuple[str, str]) -> 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, str], 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, file_filter: Optional[str] = None, use_cache: bool = True) -> Tuple[List[Dict], List[str]]: """Fast scanner using os.walk (pure Python + stdlib). Args: notes_dir: Base notes directory file_filter: Optional glob-style filter (e.g., "*", "*.md") similar to Path.rglob """ notes_path = Path(notes_dir) cache_key = (str(notes_path.resolve()), file_filter or '') if use_cache: cached = _scan_cache_get(cache_key) if cached is not None: return cached 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) if file_filter and not relative_path.match(file_filter): continue folder = relative_path.parent.as_posix() # Get tags for this note (cached) tags = get_tags_cached(full_path) 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": "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, "*.md") 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 highlight (styled via CSS) snippet = f'{before}{matched_clean}{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(): 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 get_all_images(notes_dir: str) -> List[Dict]: """ Get all media files (images, audio, video, documents) from anywhere in the notes directory. Returns list of media dictionaries with metadata. Note: Function name kept as 'get_all_images' for backward compatibility. """ media_files = [] media_candidates, _folders = scan_notes_fast_walk(notes_dir, "*") # Find all media files recursively in the entire notes directory for media_file in media_candidates: # Check if it's a media file filename = Path(media_file["path"]).name media_type = get_media_type(filename) if media_type: media_files.append({ "name": filename, "path": media_file["path"], "folder": media_file["folder"], "modified": media_file["modified"], "size": media_file["size"], "type": media_type # 'image', 'audio', 'video', or 'document' }) return media_files 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, "*.md") 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, "*.md") 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