""" Utility functions for file operations, search, and markdown processing. Heavy work (backlinks, graph, full-text search, tag aggregation) is delegated to the in-memory index in note_index.py. """ import logging import os import re import shutil import threading import time import traceback import urllib.parse from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path from typing import List, Dict, Optional, Tuple, Any, TypeVar, Callable from datetime import datetime, timezone from . import note_index from .note_index import NoteRecord, extract_links_from_content logger = logging.getLogger("uvicorn.error") # ============================================================================ # 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 caches # # Two layers: # _tag_cache / _links_cache : per-file, mtime-keyed. Avoid re-parsing the # same file on warm scans. Used by get_tags_and_links_cached. # _SCAN_WALK_CACHE : per-scan TTL cache. Avoids repeated full- # directory walks when several endpoints fire in quick succession. # # All three are invalidated on every mutation (save/delete/move). The # note_index in note_index.py is the system of record for derived data # (links, backlinks, tags-by-name, search) — these caches just exist to # avoid double-reading files we already parsed. # ============================================================================ _tag_cache: Dict[str, Tuple[float, List[str]]] = {} _links_cache: Dict[str, Tuple[float, Dict[str, List[str]]]] = {} _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 _scan_cache_invalidate() -> None: """Drop the TTL scan cache. Called from every mutation handler.""" with _SCAN_WALK_CACHE_LOCK: _SCAN_WALK_CACHE.clear() def ensure_index_built(notes_dir: str) -> None: """Trigger a fresh scan when the NoteIndex hasn't been populated yet. Scans with include_media=True so every index consumer (stats, notes, backlinks, search) shares one fingerprint and never thrashes.""" if not note_index.get_index().is_built(): scan_notes_fast_walk(notes_dir, use_cache=False, include_media=True) def get_tags_and_links_cached( file_path: Path, rel_path: Optional[str] = None, ) -> Tuple[List[str], Dict[str, List[str]]]: """Fused tags + raw-links extraction, mtime-cached. Tries in-process per-file caches first, then the NoteIndex (when rel_path is provided), then reads the file.""" try: mtime = file_path.stat().st_mtime file_key = str(file_path) tag_cached = _tag_cache.get(file_key) link_cached = _links_cache.get(file_key) tags_ok = tag_cached is not None and tag_cached[0] == mtime links_ok = link_cached is not None and link_cached[0] == mtime if tags_ok and links_ok: return tag_cached[1], link_cached[1] if rel_path is not None: indexed = note_index.try_get_extraction(rel_path, mtime) if indexed is not None: idx_tags, idx_links = indexed _tag_cache[file_key] = (mtime, idx_tags) _links_cache[file_key] = (mtime, idx_links) return idx_tags, idx_links with open(file_path, 'r', encoding='utf-8') as f: content = f.read() tags = tag_cached[1] if tags_ok else parse_tags(content) links = link_cached[1] if links_ok else extract_links_from_content(content) if not tags_ok: _tag_cache[file_key] = (mtime, tags) if not links_ok: _links_cache[file_key] = (mtime, links) return tags, links except Exception: return [], {"wikilinks": [], "mdlinks": []} 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 if not validate_path_security(notes_dir, full_path): return False full_path.mkdir(parents=True, exist_ok=True) _scan_cache_invalidate() 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 # Walk the vault once. Tag/link extraction is parallelized below across # the markdown files we collected here. notes: List[Dict] = [] folders_set = set() md_to_extract: List[Tuple[int, Path, str]] = [] # (index, full_path, rel_path) for root, dirnames, filenames in os.walk(notes_path): 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() rel_str = relative_path.as_posix() notes.append({ "name": full_path.stem, "path": rel_str, "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": [], "_mtime": st.st_mtime, # internal — popped before returning }) if is_markdown: md_to_extract.append((len(notes) - 1, full_path, rel_str)) # Tag + raw-link extraction in parallel. Search-index terms are built # later, lazily, on the first /api/search call. sources_raw: Dict[str, Dict[str, List[str]]] = {} if md_to_extract: path_pairs = [(full, rel) for (_, full, rel) in md_to_extract] if len(md_to_extract) >= 50: workers = min(8, (os.cpu_count() or 4)) with ThreadPoolExecutor(max_workers=workers) as ex: extraction = list(ex.map( lambda pair: get_tags_and_links_cached(pair[0], pair[1]), path_pairs, )) else: extraction = [get_tags_and_links_cached(full, rel) for (full, rel) in path_pairs] for (idx, _full, rel_str), (tags, links) in zip(md_to_extract, extraction): notes[idx]["tags"] = tags sources_raw[rel_str] = links # Push the result into the NoteIndex. Short-circuits on warm scans. notes_meta = [ NoteRecord( path=n["path"], name=n["name"], folder=n["folder"], modified=n["modified"], size=n["size"], type=n["type"], mtime=n["_mtime"], tags=tuple(n["tags"]), ) for n in notes ] note_index.populate_from_scan(notes_meta, folders_set, sources_raw) for n in notes: n.pop("_mtime", None) 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. Returns (success, error_message).""" old_full_path = Path(notes_dir) / old_path new_full_path = Path(notes_dir) / new_path 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}" if new_full_path.exists(): return False, f"A note already exists at: {new_path}" _drop_path_caches(old_full_path) try: new_full_path.parent.mkdir(parents=True, exist_ok=True) old_full_path.rename(new_full_path) except Exception as e: return False, f"Failed to move file: {str(e)}" note_index.on_note_renamed(notes_dir, old_full_path, new_full_path) _scan_cache_invalidate() return True, "" def move_folder(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]: """Move a folder. Returns (success, error_message).""" old_full_path = Path(notes_dir) / old_path new_full_path = Path(notes_dir) / new_path 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}" if new_full_path.exists(): return False, f"A folder already exists at: {new_path}" _drop_prefix_caches(old_full_path) try: new_full_path.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(old_full_path), str(new_full_path)) except Exception as e: return False, f"Failed to move folder: {str(e)}" note_index.on_folder_renamed(notes_dir, old_full_path, new_full_path) _scan_cache_invalidate() return True, "" def rename_folder(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]: """Rename a folder (same as move, named 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 if not validate_path_security(notes_dir, full_path): logger.warning("Security: Path is outside notes directory: %s", full_path) return False if not full_path.exists(): logger.warning("Folder does not exist: %s", full_path) return False if not full_path.is_dir(): logger.warning("Path is not a directory: %s", full_path) return False _drop_prefix_caches(full_path) shutil.rmtree(full_path) note_index.on_folder_deleted(notes_dir, full_path) _scan_cache_invalidate() return True except Exception as e: logger.error("Error deleting folder '%s': %s", folder_path, e) logger.error(traceback.format_exc()) return False def _drop_path_caches(full_path: Path) -> None: """Evict the per-file mtime caches for a single note.""" key = str(full_path) _tag_cache.pop(key, None) _links_cache.pop(key, None) def _drop_prefix_caches(folder_full_path: Path) -> None: """Evict per-file mtime caches for every entry under a folder.""" prefix = str(folder_full_path) for k in [k for k in _tag_cache if k.startswith(prefix)]: _tag_cache.pop(k, None) for k in [k for k in _links_cache if k.startswith(prefix)]: _links_cache.pop(k, None) 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 if not note_path.endswith('.md'): full_path = full_path.with_suffix('.md') if not validate_path_security(notes_dir, full_path): return False full_path.parent.mkdir(parents=True, exist_ok=True) with open(full_path, 'w', encoding='utf-8') as f: f.write(content) # Refresh the per-file mtime caches with what we just wrote, so the next # scan doesn't re-parse this file. extract_links + parse_tags + save are # tiny relative to the file write. try: mtime = full_path.stat().st_mtime file_key = str(full_path) _tag_cache[file_key] = (mtime, parse_tags(content)) _links_cache[file_key] = (mtime, extract_links_from_content(content)) except Exception: pass # caches are best-effort note_index.on_note_saved(notes_dir, full_path, content) _scan_cache_invalidate() 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 if not validate_path_security(notes_dir, full_path): return False _drop_path_caches(full_path) full_path.unlink() note_index.on_note_deleted(notes_dir, full_path) _scan_cache_invalidate() return True def search_notes(notes_dir: str, query: str) -> List[Dict]: """Full-text search through note contents. Narrow the candidate set via the inverted index, then run the snippet extractor on each candidate.""" from html import escape results: List[Dict] = [] ensure_index_built(notes_dir) note_index.ensure_search_index(notes_dir) candidates = note_index.get_search_candidates(query) idx = note_index.get_index() if candidates is None: # Query too short to tokenize — iterate every indexed note instead. candidate_records = idx.all_note_records() else: candidate_records = [(p, idx.get_note_record(p)) for p in candidates] candidate_records = [(p, r) for (p, r) in candidate_records if r is not None and r.type == "note"] candidate_records.sort(key=lambda pr: pr[1].mtime, reverse=True) iterable = [{"path": p, "name": r.name} for (p, r) in candidate_records] for note in iterable: path = note["path"] md_file = Path(notes_dir) / path try: with open(md_file, 'r', encoding='utf-8') as f: content = f.read() matches = list(re.finditer(re.escape(query), content, re.IGNORECASE)) if not matches: continue matched_lines = [] for match in matches[:3]: start_index = match.start() end_index = match.end() matched_text = match.group() context_start = max(0, start_index - 15) context_end = min(len(content), end_index + 15) 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', ' ')) snippet = f'{before}{matched_clean}{after}' if context_start > 0: snippet = '...' + snippet if context_end < len(content): snippet = snippet + '...' line_number = content.count('\n', 0, start_index) + 1 matched_lines.append({ "line_number": line_number, "context": snippet, }) relative_path = Path(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, *, sibling_folder: Optional[str] = None, ) -> Optional[str]: """ Save uploaded media under the vault. Default (sibling_folder is None): store in ``_attachments`` next to the note implied by ``note_path`` (drag/drop, paste, etc.). If ``sibling_folder`` is set (including ``""`` for vault root): store ``drawing-{timestamp}.png`` in that folder next to ``.md`` files — used for new drawings from the + menu. Returns a relative path from ``notes_dir``, or None on failure. """ base = Path(notes_dir) if sibling_folder is not None: cf = (sibling_folder or "").strip().replace("\\", "/") segments = [p for p in cf.split("/") if p and p != "."] if any(p == ".." for p in segments): return None dest_dir = base.joinpath(*segments) if segments else base if not validate_path_security(notes_dir, dest_dir / ".nd_probe"): return None try: dest_dir.mkdir(parents=True, exist_ok=True) except OSError: return None timestamp = datetime.now().strftime("%Y%m%d%H%M%S") full_path = dest_dir / f"drawing-{timestamp}.png" if not validate_path_security(notes_dir, full_path): return None try: with open(full_path, "wb") as f: f.write(file_data) _scan_cache_invalidate() return str(full_path.relative_to(base).as_posix()) except OSError as e: logger.error("Error saving image: %s", e) return None sanitized_name = sanitize_filename(filename) ext = Path(sanitized_name).suffix name_without_ext = Path(sanitized_name).stem timestamp = datetime.now().strftime("%Y%m%d%H%M%S") final_filename = f"{name_without_ext}-{timestamp}{ext}" attachments_dir = get_attachment_dir(notes_dir, note_path) attachments_dir.mkdir(parents=True, exist_ok=True) full_path = attachments_dir / final_filename if not validate_path_security(notes_dir, full_path): logger.warning("Security: Attempted to save image outside notes directory: %s", full_path) return None try: with open(full_path, "wb") as f: f.write(file_data) _scan_cache_invalidate() return str(full_path.relative_to(base).as_posix()) except OSError as e: logger.error("Error saving image: %s", 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', 'drawing', or None if not a media file. Drawings are PNG files stored like images but named drawing-*.png (editable canvas in the app). """ name_lower = Path(filename).name.lower() if name_lower.startswith('drawing-') and name_lower.endswith('.png'): return 'drawing' 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: logger.error("Error parsing tags: %s", e) return [] def get_all_tags(notes_dir: str) -> Dict[str, int]: """All tags in the vault with note counts.""" ensure_index_built(notes_dir) return note_index.get_all_tags() def get_notes_by_tag(notes_dir: str, tag: str) -> List[Dict]: """All notes carrying `tag` (case-insensitive).""" ensure_index_built(notes_dir) idx = note_index.get_index() records = [idx.get_note_record(p) for p in note_index.get_paths_for_tag(tag.lower())] matching = [ { "name": r.name, "path": r.path, "folder": r.folder, "modified": r.modified, "size": r.size, "tags": list(r.tags), } for r in records if r is not None and r.type == "note" ] matching.sort(key=lambda n: n["modified"], reverse=True) return matching # ============================================================================ # 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): logger.warning("Security: Templates directory is outside notes directory: %s", templates_path) return templates try: for template_file in templates_path.glob("*.md"): try: if not validate_path_security(notes_dir, template_file): logger.warning("Security: Skipping template outside notes directory: %s", 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: logger.error("Error reading template %s: %s", template_file, e) continue except Exception as e: logger.error("Error accessing templates directory: %s", 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): logger.warning("Security: Template path is outside notes directory: %s", template_path) return None try: with open(template_path, 'r', encoding='utf-8') as f: return f.read() except Exception as e: logger.error("Error reading template %s: %s", template_name, e) return None # Matches custom strftime placeholders like {{date:%Y%m%d}} or {{time:%H%M%S}}. # The three prefixes mirror the bare {{date}}, {{time}}, {{datetime}} placeholders; # disallowing '{' and '}' inside the format keeps parsing unambiguous and # guarantees we never swallow an adjacent placeholder. _STRFTIME_PLACEHOLDER_RE = re.compile(r'\{\{(date|time|datetime):([^{}]+)\}\}') def apply_template_placeholders(content: str, note_path: str) -> str: """ Replace template placeholders with actual values. Built-in named placeholders (fixed default formats): {{date}} - Current date (YYYY-MM-DD) {{time}} - Current time (HH:MM:SS) {{datetime}} - Current datetime (YYYY-MM-DD HH:MM:SS) {{timestamp}} - Unix timestamp (seconds) {{year}} - Current year (YYYY) {{month}} - Current month (MM) {{day}} - Current day (DD) {{title}} - Note name without extension {{folder}} - Parent folder name Custom date/time formats (strftime escape hatch): {{date:FMT}}, {{time:FMT}}, {{datetime:FMT}} Any of the three date/time placeholders above also accepts an optional ":FMT" suffix where FMT is a Python strftime() format string. Pick the prefix whose default format covers the same components you want to format. Examples: {{datetime:%Y%m%d%H%M%S}} -> 20260506154200 (filename-safe stamp) {{date:%d/%m/%Y}} -> 06/05/2026 (European date) {{date:%A}} -> Wednesday (full weekday name) {{time:%H%M%S}} -> 154200 (compact time) {{date:%V}} -> 19 (ISO week number) Invalid format strings are left in the output unchanged so a typo is visible rather than silently swallowed. 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) def _strftime_sub(match: 're.Match[str]') -> str: fmt = match.group(2) try: return now.strftime(fmt) except (ValueError, TypeError): return match.group(0) content = _STRFTIME_PLACEHOLDER_RE.sub(_strftime_sub, content) 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 _extract_backlink_references( notes_dir: str, source_path: str, target_path: str, target_path_lower: str, target_path_no_ext_lower: str, wikilink_refs: set, ) -> List[Dict]: """Read one source file and pull line-level references that point to target. Same regex + resolution rules as the legacy get_backlinks loop body; just factored out so both the index-driven and the legacy code paths use the exact same context-extraction logic. """ source_folder = str(Path(source_path).parent).replace('\\', '/') if source_folder == '.': source_folder = '' full_path = Path(notes_dir) / source_path try: with open(full_path, 'r', encoding='utf-8') as f: content = f.read() except Exception: return [] lines = content.split('\n') found_links: List[Dict] = [] for line_num, line in enumerate(lines, 1): for match in re.finditer(r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]', line): link_target = match.group(1).strip().lower() link_target_no_ext = link_target.replace('.md', '') 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", }) for match in re.finditer(r'\[([^\]]+)\]\((?!https?://|mailto:|#|data:)([^\)]+)\)', line): link_path = match.group(2).split('#')[0] if not link_path: continue link_path = urllib.parse.unquote(link_path) if link_path.startswith('./'): link_path = link_path[2:] link_path_with_md = link_path if link_path.endswith('.md') else link_path + '.md' resolved_path = None 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 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 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", }) return found_links def get_backlinks(notes_dir: str, target_note_path: str) -> List[Dict]: """All notes that link TO `target_note_path`. The index narrows the candidate set; we only read the candidate files for line context.""" target_path = target_note_path target_path_lower = target_path.lower() target_path_no_ext_lower = target_path_lower.replace('.md', '') wikilink_refs = { target_path_lower, target_path_no_ext_lower, Path(target_path).stem.lower(), } ensure_index_built(notes_dir) idx = note_index.get_index() candidates = note_index.get_backlink_candidates(target_path) records = [(p, idx.get_note_record(p)) for p in candidates] records = [(p, r) for (p, r) in records if r is not None and r.type == "note" and p != target_path] records.sort(key=lambda pr: pr[1].mtime, reverse=True) backlinks: List[Dict] = [] for source_path, record in records: refs = _extract_backlink_references( notes_dir, source_path, target_path, target_path_lower, target_path_no_ext_lower, wikilink_refs, ) if refs: backlinks.append({ "path": source_path, "name": record.name.replace('.md', ''), "references": refs[:3], }) return backlinks