refactor(api): single scan_notes_fast_walk, remove get_all_*
This commit is contained in:
parent
7073904143
commit
4d4c68b46c
|
|
@ -13,8 +13,8 @@ from pathlib import Path
|
||||||
from typing import Optional, Tuple
|
from typing import Optional, Tuple
|
||||||
import mimetypes
|
import mimetypes
|
||||||
|
|
||||||
# Import shared media type definitions from utils to avoid duplication
|
# Import shared media type definitions and scanner from utils to avoid duplication
|
||||||
from backend.utils import MEDIA_EXTENSIONS, get_media_type
|
from backend.utils import MEDIA_EXTENSIONS, get_media_type, scan_notes_fast_walk
|
||||||
|
|
||||||
|
|
||||||
def get_media_as_base64(media_path: Path) -> Optional[Tuple[str, str]]:
|
def get_media_as_base64(media_path: Path) -> Optional[Tuple[str, str]]:
|
||||||
|
|
@ -111,8 +111,10 @@ def find_media_in_attachments(media_name: str, note_folder: Path, notes_dir: Pat
|
||||||
# Fallback: search all _attachments folders recursively (slower but thorough)
|
# Fallback: search all _attachments folders recursively (slower but thorough)
|
||||||
# This handles cross-folder media references like in Obsidian
|
# This handles cross-folder media references like in Obsidian
|
||||||
try:
|
try:
|
||||||
for attachment_folder in notes_dir.rglob('_attachments'):
|
_files, folders = scan_notes_fast_walk(str(notes_dir), "*")
|
||||||
if attachment_folder.is_dir():
|
for folder in folders:
|
||||||
|
if folder == '_attachments' or folder.endswith('/_attachments'):
|
||||||
|
attachment_folder = notes_dir / folder
|
||||||
candidate = attachment_folder / media_name
|
candidate = attachment_folder / media_name
|
||||||
if candidate.exists() and candidate.is_file():
|
if candidate.exists() and candidate.is_file():
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -21,7 +21,7 @@ from slowapi.util import get_remote_address
|
||||||
from slowapi.errors import RateLimitExceeded
|
from slowapi.errors import RateLimitExceeded
|
||||||
|
|
||||||
from .utils import (
|
from .utils import (
|
||||||
get_all_notes,
|
scan_notes_fast_walk,
|
||||||
get_note_content,
|
get_note_content,
|
||||||
save_note,
|
save_note,
|
||||||
delete_note,
|
delete_note,
|
||||||
|
|
@ -29,7 +29,6 @@ from .utils import (
|
||||||
create_note_metadata,
|
create_note_metadata,
|
||||||
ensure_directories,
|
ensure_directories,
|
||||||
create_folder,
|
create_folder,
|
||||||
get_all_folders,
|
|
||||||
move_note,
|
move_note,
|
||||||
move_folder,
|
move_folder,
|
||||||
rename_folder,
|
rename_folder,
|
||||||
|
|
@ -900,8 +899,7 @@ async def create_note_from_template(request: Request, data: dict):
|
||||||
async def list_notes():
|
async def list_notes():
|
||||||
"""List all notes with metadata"""
|
"""List all notes with metadata"""
|
||||||
try:
|
try:
|
||||||
notes = get_all_notes(config['storage']['notes_dir'])
|
notes, folders = scan_notes_fast_walk(config['storage']['notes_dir'], "*.md")
|
||||||
folders = get_all_folders(config['storage']['notes_dir'])
|
|
||||||
return {"notes": notes, "folders": folders}
|
return {"notes": notes, "folders": folders}
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to list notes"))
|
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to list notes"))
|
||||||
|
|
@ -1023,7 +1021,7 @@ async def get_graph():
|
||||||
try:
|
try:
|
||||||
import re
|
import re
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
notes = get_all_notes(config['storage']['notes_dir'])
|
notes, _folders = scan_notes_fast_walk(config['storage']['notes_dir'], "*.md")
|
||||||
nodes = []
|
nodes = []
|
||||||
edges = []
|
edges = []
|
||||||
|
|
||||||
|
|
|
||||||
139
backend/utils.py
139
backend/utils.py
|
|
@ -15,6 +15,7 @@ from datetime import datetime, timezone
|
||||||
_tag_cache: Dict[str, Tuple[float, List[str]]] = {}
|
_tag_cache: Dict[str, Tuple[float, List[str]]] = {}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def validate_path_security(notes_dir: str, path: Path) -> bool:
|
def validate_path_security(notes_dir: str, path: Path) -> bool:
|
||||||
"""
|
"""
|
||||||
Validate that a path is within the notes directory (security check).
|
Validate that a path is within the notes directory (security check).
|
||||||
|
|
@ -58,20 +59,54 @@ def create_folder(notes_dir: str, folder_path: str) -> bool:
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
def get_all_folders(notes_dir: str) -> List[str]:
|
def scan_notes_fast_walk(notes_dir: str, file_filter: Optional[str] = None, use_cache: bool = True) -> Tuple[List[Dict], List[str]]:
|
||||||
"""Get all folders in the notes directory, including empty ones"""
|
"""Fast scanner using os.walk (pure Python + stdlib).
|
||||||
folders = []
|
|
||||||
|
Args:
|
||||||
|
notes_dir: Base notes directory
|
||||||
|
file_filter: Optional glob-style filter (e.g., "*", "*.md") similar to Path.rglob
|
||||||
|
"""
|
||||||
notes_path = Path(notes_dir)
|
notes_path = Path(notes_dir)
|
||||||
|
|
||||||
for item in notes_path.rglob("*"):
|
notes: List[Dict] = []
|
||||||
if item.is_dir():
|
folders_set = set()
|
||||||
relative_path = item.relative_to(notes_path)
|
|
||||||
folder_path = str(relative_path.as_posix())
|
|
||||||
if folder_path and not folder_path.startswith('.'):
|
|
||||||
folders.append(folder_path)
|
|
||||||
|
|
||||||
return sorted(folders)
|
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()
|
||||||
|
|
||||||
|
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": [],
|
||||||
|
})
|
||||||
|
|
||||||
|
return sorted(notes, key=lambda x: x.get('modified', ''), reverse=True), sorted(folders_set)
|
||||||
|
|
||||||
def move_note(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]:
|
def move_note(notes_dir: str, old_path: str, new_path: str) -> tuple[bool, str]:
|
||||||
"""Move a note to a different location
|
"""Move a note to a different location
|
||||||
|
|
@ -200,34 +235,6 @@ def delete_folder(notes_dir: str, folder_path: str) -> bool:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def get_all_notes(notes_dir: str) -> List[Dict]:
|
|
||||||
"""Recursively get all markdown notes and images"""
|
|
||||||
items = []
|
|
||||||
notes_path = Path(notes_dir)
|
|
||||||
|
|
||||||
# Get all markdown notes
|
|
||||||
for md_file in notes_path.rglob("*.md"):
|
|
||||||
relative_path = md_file.relative_to(notes_path)
|
|
||||||
stat = md_file.stat()
|
|
||||||
|
|
||||||
# Get tags for this note (cached)
|
|
||||||
tags = get_tags_cached(md_file)
|
|
||||||
|
|
||||||
items.append({
|
|
||||||
"name": md_file.stem,
|
|
||||||
"path": str(relative_path.as_posix()),
|
|
||||||
"folder": str(relative_path.parent.as_posix()) if str(relative_path.parent) != "." else "",
|
|
||||||
"modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
|
||||||
"size": stat.st_size,
|
|
||||||
"type": "note",
|
|
||||||
"tags": tags
|
|
||||||
})
|
|
||||||
|
|
||||||
# Get all images
|
|
||||||
images = get_all_images(notes_dir)
|
|
||||||
items.extend(images)
|
|
||||||
|
|
||||||
return sorted(items, key=lambda x: x['modified'], reverse=True)
|
|
||||||
|
|
||||||
|
|
||||||
def get_note_content(notes_dir: str, note_path: str) -> Optional[str]:
|
def get_note_content(notes_dir: str, note_path: str) -> Optional[str]:
|
||||||
|
|
@ -297,9 +304,10 @@ def search_notes(notes_dir: str, query: str) -> List[Dict]:
|
||||||
"""
|
"""
|
||||||
from html import escape
|
from html import escape
|
||||||
results = []
|
results = []
|
||||||
notes_path = Path(notes_dir)
|
notes, _folders = scan_notes_fast_walk(notes_dir, "*.md")
|
||||||
|
|
||||||
for md_file in notes_path.rglob("*.md"):
|
for note in notes:
|
||||||
|
md_file = Path(notes_dir) / note["path"]
|
||||||
try:
|
try:
|
||||||
with open(md_file, 'r', encoding='utf-8') as f:
|
with open(md_file, 'r', encoding='utf-8') as f:
|
||||||
content = f.read()
|
content = f.read()
|
||||||
|
|
@ -343,7 +351,7 @@ def search_notes(notes_dir: str, query: str) -> List[Dict]:
|
||||||
"context": snippet
|
"context": snippet
|
||||||
})
|
})
|
||||||
|
|
||||||
relative_path = md_file.relative_to(notes_path)
|
relative_path = Path(note["path"])
|
||||||
results.append({
|
results.append({
|
||||||
"name": md_file.stem,
|
"name": md_file.stem,
|
||||||
"path": str(relative_path.as_posix()),
|
"path": str(relative_path.as_posix()),
|
||||||
|
|
@ -517,28 +525,20 @@ def get_all_images(notes_dir: str) -> List[Dict]:
|
||||||
Note: Function name kept as 'get_all_images' for backward compatibility.
|
Note: Function name kept as 'get_all_images' for backward compatibility.
|
||||||
"""
|
"""
|
||||||
media_files = []
|
media_files = []
|
||||||
notes_path = Path(notes_dir)
|
media_candidates, _folders = scan_notes_fast_walk(notes_dir, "*")
|
||||||
|
|
||||||
# Find all media files recursively in the entire notes directory
|
# Find all media files recursively in the entire notes directory
|
||||||
for media_file in notes_path.rglob("*"):
|
for media_file in media_candidates:
|
||||||
# Skip directories and hidden files/folders
|
|
||||||
if media_file.is_dir():
|
|
||||||
continue
|
|
||||||
if any(part.startswith('.') for part in media_file.parts):
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Check if it's a media file
|
# Check if it's a media file
|
||||||
media_type = get_media_type(media_file.name)
|
filename = Path(media_file["path"]).name
|
||||||
|
media_type = get_media_type(filename)
|
||||||
if media_type:
|
if media_type:
|
||||||
relative_path = media_file.relative_to(notes_path)
|
|
||||||
stat = media_file.stat()
|
|
||||||
|
|
||||||
media_files.append({
|
media_files.append({
|
||||||
"name": media_file.name,
|
"name": filename,
|
||||||
"path": str(relative_path.as_posix()),
|
"path": media_file["path"],
|
||||||
"folder": str(relative_path.parent.as_posix()) if relative_path.parent != Path('.') else "",
|
"folder": media_file["folder"],
|
||||||
"modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
"modified": media_file["modified"],
|
||||||
"size": stat.st_size,
|
"size": media_file["size"],
|
||||||
"type": media_type # 'image', 'audio', 'video', or 'document'
|
"type": media_type # 'image', 'audio', 'video', or 'document'
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
@ -688,9 +688,10 @@ def get_all_tags(notes_dir: str) -> Dict[str, int]:
|
||||||
Dictionary mapping tag names to note counts
|
Dictionary mapping tag names to note counts
|
||||||
"""
|
"""
|
||||||
tag_counts = {}
|
tag_counts = {}
|
||||||
notes_path = Path(notes_dir)
|
notes, _folders = scan_notes_fast_walk(notes_dir, "*.md")
|
||||||
|
|
||||||
for md_file in notes_path.rglob("*.md"):
|
for note in notes:
|
||||||
|
md_file = Path(notes_dir) / note["path"]
|
||||||
# Get tags using cache
|
# Get tags using cache
|
||||||
tags = get_tags_cached(md_file)
|
tags = get_tags_cached(md_file)
|
||||||
|
|
||||||
|
|
@ -713,22 +714,20 @@ def get_notes_by_tag(notes_dir: str, tag: str) -> List[Dict]:
|
||||||
"""
|
"""
|
||||||
matching_notes = []
|
matching_notes = []
|
||||||
tag_lower = tag.lower()
|
tag_lower = tag.lower()
|
||||||
notes_path = Path(notes_dir)
|
notes, _folders = scan_notes_fast_walk(notes_dir, "*.md")
|
||||||
|
|
||||||
for md_file in notes_path.rglob("*.md"):
|
for note in notes:
|
||||||
|
md_file = Path(notes_dir) / note["path"]
|
||||||
# Get tags using cache
|
# Get tags using cache
|
||||||
tags = get_tags_cached(md_file)
|
tags = get_tags_cached(md_file)
|
||||||
|
|
||||||
if tag_lower in tags:
|
if tag_lower in tags:
|
||||||
relative_path = md_file.relative_to(notes_path)
|
|
||||||
stat = md_file.stat()
|
|
||||||
|
|
||||||
matching_notes.append({
|
matching_notes.append({
|
||||||
"name": md_file.stem,
|
"name": note["name"],
|
||||||
"path": str(relative_path.as_posix()),
|
"path": note["path"],
|
||||||
"folder": str(relative_path.parent.as_posix()) if str(relative_path.parent) != "." else "",
|
"folder": note["folder"],
|
||||||
"modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
"modified": note["modified"],
|
||||||
"size": stat.st_size,
|
"size": note["size"],
|
||||||
"tags": tags
|
"tags": tags
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue