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

feat: add hidden files configuration with per-vault settings for includeHidden and hiddenWhitelist, supporting environment variables and UI controls for selective indexing of dot-prefixed files and folders

Browse Source
This commit is contained in:
bruno
2026-03-25 09:54:34 -04:00
parent 8c30b0d238
commit 9e42fb072b
8 changed files with 790 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/attachment_indexer.py
+15 -7
View File
@@ -1,8 +1,10 @@
import asyncio
import logging
import threading
from pathlib import Path
from typing import Dict, List, Optional, Set
import threading
from backend.indexer import _should_include_path
logger = logging.getLogger("obsigate.attachment_indexer")
@@ -34,7 +36,7 @@ def clear_resolution_cache(vault_name: Optional[str] = None) -> None:
del _resolution_cache[key]
def _scan_vault_attachments(vault_name: str, vault_path: str) -> Dict[str, List[Path]]:
def _scan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict = None) -> Dict[str, List[Path]]:
"""Synchronously scan a vault directory for image attachments.
Walks the vault tree and builds a filename -> absolute path mapping
@@ -43,6 +45,7 @@ def _scan_vault_attachments(vault_name: str, vault_path: str) -> Dict[str, List[
Args:
vault_name: Display name of the vault.
vault_path: Absolute filesystem path to the vault root.
vault_cfg: Optional vault configuration dict with hidden files settings.
Returns:
Dict mapping lowercase filenames to lists of absolute paths.
@@ -50,6 +53,10 @@ def _scan_vault_attachments(vault_name: str, vault_path: str) -> Dict[str, List[
vault_root = Path(vault_path)
index: Dict[str, List[Path]] = {}
# Default config if not provided
if vault_cfg is None:
vault_cfg = {"includeHidden": False, "hiddenWhitelist": []}
if not vault_root.exists():
logger.warning(f"Vault path does not exist for attachment scan: {vault_path}")
return index
@@ -58,9 +65,9 @@ def _scan_vault_attachments(vault_name: str, vault_path: str) -> Dict[str, List[
try:
for fpath in vault_root.rglob("*"):
# Skip hidden files and directories
# Check if path should be included based on hidden files configuration
rel_parts = fpath.relative_to(vault_root).parts
if any(part.startswith(".") for part in rel_parts):
if not _should_include_path(rel_parts, vault_cfg):
continue
# Only process files
@@ -121,7 +128,7 @@ async def build_attachment_index(vault_config: Dict[str, Dict[str, any]]) -> Non
new_index[name] = {}
continue
tasks.append((name, loop.run_in_executor(None, _scan_vault_attachments, name, vault_path)))
tasks.append((name, loop.run_in_executor(None, _scan_vault_attachments, name, vault_path, config)))
for name, task in tasks:
new_index[name] = await task
@@ -136,18 +143,19 @@ async def build_attachment_index(vault_config: Dict[str, Dict[str, any]]) -> Non
logger.info(f"Attachment index built: {len(attachment_index)} vaults, {total_attachments} total attachments")
async def rescan_vault_attachments(vault_name: str, vault_path: str) -> int:
async def rescan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict = None) -> int:
"""Rescan attachments for a single vault.
Args:
vault_name: Name of the vault to rescan.
vault_path: Absolute path to the vault root.
vault_cfg: Optional vault configuration dict with hidden files settings.
Returns:
Number of attachments indexed.
"""
loop = asyncio.get_event_loop()
new_vault_index = await loop.run_in_executor(None, _scan_vault_attachments, vault_name, vault_path)
new_vault_index = await loop.run_in_executor(None, _scan_vault_attachments, vault_name, vault_path, vault_cfg)
with _attachment_lock:
attachment_index[vault_name] = new_vault_index
backend/indexer.py
+62 -6
View File
@@ -61,12 +61,16 @@ def load_vault_config() -> Dict[str, Dict[str, Any]]:
Also reads optional configuration:
- VAULT_N_ATTACHMENTS_PATH: relative path to attachments folder
- VAULT_N_SCAN_ATTACHMENTS: "true"/"false" to enable/disable scanning
- VAULT_N_INCLUDE_HIDDEN: "true"/"false" to include all hidden files/folders
- VAULT_N_HIDDEN_WHITELIST: comma-separated list of hidden paths to include (e.g., ".obsidian,.github")
Returns:
Dict mapping vault names to configuration dicts with keys:
- path: filesystem path (required)
- attachmentsPath: relative attachments folder (optional)
- scanAttachmentsOnStartup: boolean (default True)
- includeHidden: boolean (default False) - include all hidden files/folders
- hiddenWhitelist: list of hidden paths to include even if includeHidden is False
- type: "VAULT" or "DIR"
"""
vaults: Dict[str, Dict[str, Any]] = {}
@@ -80,11 +84,16 @@ def load_vault_config() -> Dict[str, Dict[str, Any]]:
# Optional configuration
attachments_path = os.environ.get(f"VAULT_{n}_ATTACHMENTS_PATH")
scan_attachments = os.environ.get(f"VAULT_{n}_SCAN_ATTACHMENTS", "true").lower() == "true"
include_hidden = os.environ.get(f"VAULT_{n}_INCLUDE_HIDDEN", "false").lower() == "true"
hidden_whitelist_str = os.environ.get(f"VAULT_{n}_HIDDEN_WHITELIST", "")
hidden_whitelist = [item.strip() for item in hidden_whitelist_str.split(",") if item.strip()]
vaults[name] = {
"path": path,
"attachmentsPath": attachments_path,
"scanAttachmentsOnStartup": scan_attachments,
"includeHidden": include_hidden,
"hiddenWhitelist": hidden_whitelist,
"type": "VAULT"
}
n += 1
@@ -96,10 +105,16 @@ def load_vault_config() -> Dict[str, Dict[str, Any]]:
if not name or not path:
break
include_hidden = os.environ.get(f"DIR_{n}_INCLUDE_HIDDEN", "false").lower() == "true"
hidden_whitelist_str = os.environ.get(f"DIR_{n}_HIDDEN_WHITELIST", "")
hidden_whitelist = [item.strip() for item in hidden_whitelist_str.split(",") if item.strip()]
vaults[name] = {
"path": path,
"attachmentsPath": None,
"scanAttachmentsOnStartup": False,
"includeHidden": include_hidden,
"hiddenWhitelist": hidden_whitelist,
"type": "DIR"
}
n += 1
@@ -107,15 +122,49 @@ def load_vault_config() -> Dict[str, Dict[str, Any]]:
return vaults
def _should_include_path(rel_parts: tuple, vault_config: Dict[str, Any]) -> bool:
"""Check if a path should be included based on hidden files configuration.
Args:
rel_parts: Tuple of path parts relative to vault root
vault_config: Vault configuration dict with includeHidden and hiddenWhitelist
Returns:
True if the path should be included, False otherwise
"""
include_hidden = vault_config.get("includeHidden", False)
hidden_whitelist = vault_config.get("hiddenWhitelist", [])
# Check if any part of the path starts with a dot (hidden)
hidden_parts = [part for part in rel_parts if part.startswith(".")]
if not hidden_parts:
# No hidden parts, always include
return True
if include_hidden:
# Include all hidden files/folders
return True
# Check if any hidden part is in the whitelist
for hidden_part in hidden_parts:
if hidden_part in hidden_whitelist:
return True
# Not in whitelist and includeHidden is False
return False
# 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)
_CODE_BLOCK_RE = re.compile(r'```[\s\S]*?```', re.MULTILINE)
_INLINE_CODE_RE = re.compile(r'`[^`]+`')
def _extract_tags(post: frontmatter.Post) -> List[str]:
"""Extract tags from frontmatter metadata.
Handles tags as comma-separated string, list, or other types.
Strips leading ``#`` from each tag.
@@ -196,7 +245,7 @@ def parse_markdown_file(raw: str) -> frontmatter.Post:
return frontmatter.Post(content, **{})
def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
def _scan_vault(vault_name: str, vault_path: str, vault_cfg: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Synchronously scan a single vault directory and build file index.
Walks the vault tree, reads supported files, extracts metadata
@@ -206,6 +255,7 @@ def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
Args:
vault_name: Display name of the vault.
vault_path: Absolute filesystem path to the vault root.
vault_cfg: Optional vault configuration dict with hidden files settings.
Returns:
Dict with keys ``files`` (list), ``tags`` (counter dict), ``path`` (str), ``paths`` (list).
@@ -214,15 +264,19 @@ def _scan_vault(vault_name: str, vault_path: str) -> Dict[str, Any]:
files: List[Dict[str, Any]] = []
tag_counts: Dict[str, int] = {}
paths: List[Dict[str, str]] = []
# Default config if not provided
if vault_cfg is None:
vault_cfg = {"includeHidden": False, "hiddenWhitelist": []}
if not vault_root.exists():
logger.warning(f"Vault path does not exist: {vault_path}")
return {"files": [], "tags": {}, "path": vault_path, "paths": []}
for fpath in vault_root.rglob("*"):
# Skip hidden files and directories
# Check if path should be included based on hidden files configuration
rel_parts = fpath.relative_to(vault_root).parts
if any(part.startswith(".") for part in rel_parts):
if not _should_include_path(rel_parts, vault_cfg):
continue
rel_path_str = str(fpath.relative_to(vault_root)).replace("\\", "/")
@@ -327,7 +381,7 @@ async def build_index(progress_callback=None) -> None:
async def _process_vault(name: str, config: Dict[str, Any]):
vault_path = config["path"]
vault_data = await loop.run_in_executor(None, _scan_vault, name, vault_path)
vault_data = await loop.run_in_executor(None, _scan_vault, name, vault_path, config)
vault_data["config"] = config
# Build lookup entries for the new vault
@@ -698,10 +752,12 @@ async def add_vault_to_index(vault_name: str, vault_path: str) -> Dict[str, Any]
"path": vault_path,
"attachmentsPath": None,
"scanAttachmentsOnStartup": True,
"includeHidden": False,
"hiddenWhitelist": [],
}
loop = asyncio.get_event_loop()
vault_data = await loop.run_in_executor(None, _scan_vault, vault_name, vault_path)
vault_data = await loop.run_in_executor(None, _scan_vault, vault_name, vault_path, vault_config[vault_name])
vault_data["config"] = vault_config[vault_name]
# Build lookup entries for the new vault
backend/main.py
+104
View File
@@ -44,6 +44,12 @@ from backend.indexer import (
from backend.search import search, get_all_tags, advanced_search, suggest_titles, suggest_tags
from backend.image_processor import preprocess_images
from backend.attachment_indexer import rescan_vault_attachments, get_attachment_stats
from backend.vault_settings import (
get_vault_setting,
update_vault_setting,
get_all_vault_settings,
delete_vault_setting,
)
logging.basicConfig(
level=logging.INFO,
@@ -1385,6 +1391,104 @@ async def api_attachment_stats(vault: Optional[str] = Query(None, description="V
return {"vaults": stats}
# ---------------------------------------------------------------------------
# Vault Settings API — Hidden files configuration
# ---------------------------------------------------------------------------
@app.get("/api/vaults/{vault_name}/settings")
async def api_get_vault_settings(vault_name: str, current_user=Depends(require_auth)):
"""Get settings for a specific vault including hidden files configuration.
Args:
vault_name: Name of the vault.
Returns:
Dict with vault settings including includeHidden and hiddenWhitelist.
"""
if vault_name not in index:
raise HTTPException(status_code=404, detail=f"Vault '{vault_name}' not found")
# Get current vault config from environment
vault_cfg = vault_config.get(vault_name, {})
# Get persisted settings
persisted = get_vault_setting(vault_name) or {}
# Merge with defaults
settings = {
"includeHidden": vault_cfg.get("includeHidden", False),
"hiddenWhitelist": vault_cfg.get("hiddenWhitelist", []),
}
settings.update(persisted)
return settings
@app.post("/api/vaults/{vault_name}/settings")
async def api_update_vault_settings(vault_name: str, body: dict = Body(...), current_user=Depends(require_admin)):
"""Update settings for a specific vault.
Args:
vault_name: Name of the vault.
body: Dict with settings to update (includeHidden, hiddenWhitelist).
Returns:
Updated settings dict.
"""
if vault_name not in index:
raise HTTPException(status_code=404, detail=f"Vault '{vault_name}' not found")
# Validate settings
settings_to_update = {}
if "includeHidden" in body:
if not isinstance(body["includeHidden"], bool):
raise HTTPException(status_code=400, detail="includeHidden must be a boolean")
settings_to_update["includeHidden"] = body["includeHidden"]
if "hiddenWhitelist" in body:
if not isinstance(body["hiddenWhitelist"], list):
raise HTTPException(status_code=400, detail="hiddenWhitelist must be a list")
# Validate each item is a string
if not all(isinstance(item, str) for item in body["hiddenWhitelist"]):
raise HTTPException(status_code=400, detail="All hiddenWhitelist items must be strings")
settings_to_update["hiddenWhitelist"] = body["hiddenWhitelist"]
# Update persisted settings
updated = update_vault_setting(vault_name, settings_to_update)
# Update in-memory vault config
if vault_name in vault_config:
vault_config[vault_name].update(settings_to_update)
logger.info(f"Updated settings for vault '{vault_name}': {settings_to_update}")
return updated
@app.get("/api/vaults/settings/all")
async def api_get_all_vault_settings(current_user=Depends(require_auth)):
"""Get settings for all vaults.
Returns:
Dict mapping vault names to their settings.
"""
all_settings = {}
for vault_name in index.keys():
vault_cfg = vault_config.get(vault_name, {})
persisted = get_vault_setting(vault_name) or {}
settings = {
"includeHidden": vault_cfg.get("includeHidden", False),
"hiddenWhitelist": vault_cfg.get("hiddenWhitelist", []),
}
settings.update(persisted)
all_settings[vault_name] = settings
return all_settings
# ---------------------------------------------------------------------------
# Configuration API
# ---------------------------------------------------------------------------
backend/vault_settings.py
+121
View File
@@ -0,0 +1,121 @@
"""Vault-specific settings management.
Provides persistent storage for per-vault configuration like hidden files settings.
Settings are stored in /app/data/vault_settings.json
"""
import json
import logging
from pathlib import Path
from typing import Dict, Any, Optional
import threading
logger = logging.getLogger("obsigate.vault_settings")
_SETTINGS_PATH = Path("/app/data/vault_settings.json")
_settings_lock = threading.Lock()
# In-memory cache of vault settings
_vault_settings: Dict[str, Dict[str, Any]] = {}
def load_vault_settings() -> Dict[str, Dict[str, Any]]:
"""Load vault settings from disk.
Returns:
Dict mapping vault names to their settings.
"""
global _vault_settings
with _settings_lock:
if _SETTINGS_PATH.exists():
try:
content = _SETTINGS_PATH.read_text(encoding="utf-8")
_vault_settings = json.loads(content)
logger.info(f"Loaded settings for {len(_vault_settings)} vaults")
except Exception as e:
logger.warning(f"Failed to load vault settings: {e}")
_vault_settings = {}
else:
_vault_settings = {}
return dict(_vault_settings)
def save_vault_settings() -> None:
"""Persist vault settings to disk."""
with _settings_lock:
try:
_SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
_SETTINGS_PATH.write_text(
json.dumps(_vault_settings, indent=2, ensure_ascii=False),
encoding="utf-8"
)
logger.info(f"Saved settings for {len(_vault_settings)} vaults")
except Exception as e:
logger.error(f"Failed to save vault settings: {e}")
raise
def get_vault_setting(vault_name: str) -> Optional[Dict[str, Any]]:
"""Get settings for a specific vault.
Args:
vault_name: Name of the vault.
Returns:
Dict with vault settings or None if not found.
"""
with _settings_lock:
return _vault_settings.get(vault_name)
def update_vault_setting(vault_name: str, settings: Dict[str, Any]) -> Dict[str, Any]:
"""Update settings for a specific vault.
Args:
vault_name: Name of the vault.
settings: Dict with settings to update (partial update supported).
Returns:
Updated settings dict for the vault.
"""
with _settings_lock:
if vault_name not in _vault_settings:
_vault_settings[vault_name] = {}
_vault_settings[vault_name].update(settings)
save_vault_settings()
return dict(_vault_settings[vault_name])
def delete_vault_setting(vault_name: str) -> bool:
"""Delete settings for a specific vault.
Args:
vault_name: Name of the vault.
Returns:
True if settings were deleted, False if vault not found.
"""
with _settings_lock:
if vault_name in _vault_settings:
del _vault_settings[vault_name]
save_vault_settings()
return True
return False
def get_all_vault_settings() -> Dict[str, Dict[str, Any]]:
"""Get all vault settings.
Returns:
Dict mapping vault names to their settings.
"""
with _settings_lock:
return dict(_vault_settings)
# Initialize settings on module load
load_vault_settings()