Projets/ObsiGate
1
1
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases 19 Wiki Activity

Implement multi-stage Docker build with security hardening, add health check endpoint, optimize in-memory search with O(1) wikilink lookup, extract inline tags from markdown content, and enhance documentation with architecture diagrams and performance metrics

Browse Source
This commit is contained in:
bruno
2026-03-22 19:03:34 -04:00
parent d9add8dcba
commit d311a09527
11 changed files with 1097 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/indexer.py
+145 -31
View File
@@ -2,6 +2,7 @@ import os
import asyncio
import logging
import re
import threading
from pathlib import Path
from datetime import datetime, timezone
from typing import Dict, List, Optional, Any
@@ -16,6 +17,15 @@ index: Dict[str, Dict[str, Any]] = {}
# Vault config: {name: path}
vault_config: Dict[str, str] = {}
# Thread-safe lock for index updates
_index_lock = threading.Lock()
# O(1) lookup table for wikilink resolution: {filename_lower: [{vault, path}, ...]}
_file_lookup: Dict[str, List[Dict[str, str]]] = {}
# Maximum content size stored per file for in-memory search (bytes)
SEARCH_CONTENT_LIMIT = 100_000
# Supported text-based file extensions
SUPPORTED_EXTENSIONS = {
".md", ".txt", ".log", ".py", ".js", ".ts", ".jsx", ".tsx",
@@ -30,7 +40,15 @@ SUPPORTED_EXTENSIONS = {
def load_vault_config() -> Dict[str, str]:
"""Read VAULT_N_NAME / VAULT_N_PATH env vars and return {name: path}."""
"""Read VAULT_N_NAME / VAULT_N_PATH env vars and return {name: path}.
Scans environment variables ``VAULT_1_NAME``/``VAULT_1_PATH``,
``VAULT_2_NAME``/``VAULT_2_PATH``, etc. in sequential order.
Stops at the first missing pair.
Returns:
Dict mapping vault display names to filesystem paths.
"""
vaults: Dict[str, str] = {}
n = 1
while True:
@@ -43,8 +61,25 @@ def load_vault_config() -> Dict[str, str]:
return vaults
# Regex for extracting inline #tags from markdown body (excludes code blocks)
_INLINE_TAG_RE = re.compile(r'(?:^|\s)#([a-zA-Z][a-zA-Z0-9_/-]{1,50})', re.MULTILINE)
# Regex patterns for stripping code blocks before inline tag extraction
_CODE_BLOCK_RE = re.compile(r'```.*?```', re.DOTALL)
_INLINE_CODE_RE = re.compile(r'`[^`]+`')
def _extract_tags(post: frontmatter.Post) -> List[str]:
"""Extract tags from frontmatter metadata."""
"""Extract tags from frontmatter metadata.
Handles tags as comma-separated string, list, or other types.
Strips leading ``#`` from each tag.
Args:
post: Parsed frontmatter Post object.
Returns:
List of cleaned tag strings.
"""
tags = post.metadata.get("tags", [])
if isinstance(tags, str):
tags = [t.strip().lstrip("#") for t in tags.split(",") if t.strip()]
@@ -55,8 +90,36 @@ def _extract_tags(post: frontmatter.Post) -> List[str]:
return tags
def _extract_inline_tags(content: str) -> List[str]:
"""Extract inline #tag patterns from markdown content.
Strips fenced and inline code blocks before scanning to avoid
false positives from code comments or shell commands.
Args:
content: Raw markdown content (without frontmatter).
Returns:
Deduplicated list of inline tag strings.
"""
stripped = _CODE_BLOCK_RE.sub('', content)
stripped = _INLINE_CODE_RE.sub('', stripped)
return list(set(_INLINE_TAG_RE.findall(stripped)))
def _extract_title(post: frontmatter.Post, filepath: Path) -> str:
"""Extract title from frontmatter or derive from filename."""
"""Extract title from frontmatter or derive from filename.
Falls back to the file stem with hyphens/underscores replaced by spaces
when no ``title`` key is present in frontmatter.
Args:
post: Parsed frontmatter Post object.
filepath: Path to the source file.
Returns:
Human-readable title string.
"""
title = post.metadata.get("title", "")
if not title:
title = filepath.stem.replace("-", " ").replace("_", " ")
@@ -64,7 +127,17 @@ def _extract_title(post: frontmatter.Post, filepath: Path) -> str:
def parse_markdown_file(raw: str) -> frontmatter.Post:
"""Parse markdown frontmatter, falling back to plain content if YAML is invalid."""
"""Parse markdown frontmatter, falling back to plain content if YAML is invalid.
When the YAML block is malformed, strips it and returns a Post with
empty metadata so that rendering can still proceed.
Args:
raw: Full raw markdown string including optional frontmatter.
Returns:
``frontmatter.Post`` with ``.content`` and ``.metadata`` attributes.
"""
try:
return frontmatter.loads(raw)
except Exception as exc:
@@ -78,7 +151,19 @@ def parse_markdown_file(raw: str) -> frontmatter.Post:
def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
"""Synchronously scan a single vault directory."""
"""Synchronously scan a single vault directory and build file index.
Walks the vault tree, reads supported files, extracts metadata
(tags, title, content preview) and stores a capped content snapshot
for in-memory full-text search.
Args:
vault_name: Display name of the vault.
vault_path: Absolute filesystem path to the vault root.
Returns:
Dict with keys ``files`` (list), ``tags`` (counter dict), ``path`` (str).
"""
vault_root = Path(vault_path)
files: List[Dict[str, Any]] = []
tag_counts: Dict[str, int] = {}
@@ -113,6 +198,9 @@ def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
if ext == ".md":
post = parse_markdown_file(raw)
tags = _extract_tags(post)
# Merge inline #tags found in content body
inline_tags = _extract_inline_tags(post.content)
tags = list(set(tags) | set(inline_tags))
title = _extract_title(post, fpath)
content_preview = post.content[:200].strip()
@@ -121,6 +209,7 @@ def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
"title": title,
"tags": tags,
"content_preview": content_preview,
"content": raw[:SEARCH_CONTENT_LIMIT],
"size": stat.st_size,
"modified": modified,
"extension": ext,
@@ -138,7 +227,12 @@ def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
async def build_index() -> None:
"""Build the full in-memory index for all configured vaults."""
"""Build the full in-memory index for all configured vaults.
Runs vault scans concurrently in a thread pool, then performs
an atomic swap of the global index and lookup table under a lock
to ensure thread-safe reads during reload.
"""
global index, vault_config
vault_config = load_vault_config()
@@ -156,14 +250,35 @@ async def build_index() -> None:
for name, task in tasks:
new_index[name] = await task
index.clear()
index.update(new_index)
# Build O(1) lookup table for wikilink resolution
new_lookup: Dict[str, List[Dict[str, str]]] = {}
for vname, vdata in new_index.items():
for f in vdata["files"]:
entry = {"vault": vname, "path": f["path"]}
fname = f["path"].rsplit("/", 1)[-1].lower()
fpath_lower = f["path"].lower()
for key in (fname, fpath_lower):
if key not in new_lookup:
new_lookup[key] = []
new_lookup[key].append(entry)
# Atomic swap under lock for thread safety during concurrent reads
with _index_lock:
index.clear()
index.update(new_index)
_file_lookup.clear()
_file_lookup.update(new_lookup)
total_files = sum(len(v["files"]) for v in index.values())
logger.info(f"Index built: {len(index)} vaults, {total_files} total files")
async def reload_index() -> Dict[str, Any]:
"""Force a full re-index and return stats."""
"""Force a full re-index of all vaults and return per-vault statistics.
Returns:
Dict mapping vault names to their file/tag counts.
"""
await build_index()
stats = {}
for name, data in index.items():
@@ -172,39 +287,38 @@ async def reload_index() -> Dict[str, Any]:
def get_vault_names() -> List[str]:
"""Return the list of all indexed vault names."""
return list(index.keys())
def get_vault_data(vault_name: str) -> Optional[Dict[str, Any]]:
"""Return the full index data for a vault, or ``None`` if not found."""
return index.get(vault_name)
def find_file_in_index(link_target: str, current_vault: str) -> Optional[Dict[str, str]]:
"""Find a file matching a wikilink target. Search current vault first, then all."""
"""Find a file matching a wikilink target using O(1) lookup table.
Searches by filename first, then by full relative path.
Prefers results from *current_vault* when multiple matches exist.
Args:
link_target: The wikilink target (e.g. ``"My Note"`` or ``"folder/My Note"``).
current_vault: Name of the vault the link originates from.
Returns:
Dict with ``vault`` and ``path`` keys, or ``None`` if not found.
"""
target_lower = link_target.lower().strip()
if not target_lower.endswith(".md"):
target_lower += ".md"
def _search_vault(vname: str, vdata: Dict[str, Any]):
for f in vdata["files"]:
fpath = f["path"].lower()
fname = fpath.rsplit("/", 1)[-1]
if fname == target_lower or fpath == target_lower:
return {"vault": vname, "path": f["path"]}
candidates = _file_lookup.get(target_lower, [])
if not candidates:
return None
# Search current vault first
if current_vault in index:
result = _search_vault(current_vault, index[current_vault])
if result:
return result
# Search all other vaults
for vname, vdata in index.items():
if vname == current_vault:
continue
result = _search_vault(vname, vdata)
if result:
return result
return None
# Prefer current vault when multiple vaults contain a match
for c in candidates:
if c["vault"] == current_vault:
return c
return candidates[0]