"""Second Brain Vault Server — Flask web app for browsing wiki pages and home directory."""
import os
import re
import mimetypes
import hashlib
import logging
from pathlib import Path
from datetime import datetime
from functools import wraps
from flask import (
Flask, render_template, request, redirect, url_for, jsonify, abort,
send_file, Response, stream_with_context
)
import markdown
from markdown.extensions.fenced_code import FencedCodeExtension
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# --- Config ---
HOME_DIR = Path.home()
VAULT_DIR = Path(os.environ.get("VAULT_DIR", HOME_DIR / "second-brain"))
WIKI_DIR = VAULT_DIR / "wiki"
BROWSE_DIR = Path(os.environ.get("BROWSE_DIR", HOME_DIR))
HOST = "0.0.0.0"
PORT = int(os.environ.get("PORT", 5080))
SECRET_KEY = os.environ.get("SECRET_KEY", "second-brain-dev-key")
BASIC_AUTH_USER = os.environ.get("BASIC_AUTH_USER", "")
BASIC_AUTH_PASS = os.environ.get("BASIC_AUTH_PASS", "")
MAX_FILE_VIEW_SIZE = 5 * 1024 * 1024 # 5MB limit for viewing
UPLOAD_MAX_SIZE = 50 * 1024 * 1024 # 50MB upload limit
app = Flask(__name__)
app.config["SECRET_KEY"] = SECRET_KEY
app.config["MAX_CONTENT_LENGTH"] = UPLOAD_MAX_SIZE
# Markdown extensions
MD_EXTENSIONS = [
"fenced_code",
"codehilite",
"tables",
"toc",
"nl2br",
"attr_list",
]
# --- File Watcher ---
class VaultHandler(FileSystemEventHandler):
def on_modified(self, event):
if not event.is_directory and event.src_path.endswith(".md"):
pass # Client polls for changes
file_observer = Observer()
file_observer.schedule(VaultHandler(), str(VAULT_DIR), recursive=True)
file_observer.start()
# --- Security ---
def safe_path(base, *parts):
"""Resolve path and ensure it stays within base directory. Prevents path traversal."""
target = base
for part in parts:
if not part:
continue
target = target / part
# Normalize and verify
try:
target = target.resolve()
if not str(target).startswith(str(base.resolve())):
return None
except (ValueError, OSError):
return None
return target
# --- File helpers ---
def format_size(size):
"""Format bytes to human readable."""
for unit in ["B", "KB", "MB", "GB"]:
if size < 1024:
return f"{size:.1f} {unit}"
size /= 1024
return f"{size:.1f} TB"
def format_date(timestamp):
"""Format timestamp to readable date."""
return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M")
def is_text_file(filepath):
"""Check if file is likely text by reading first bytes."""
try:
with open(filepath, "rb") as f:
chunk = f.read(8192)
return not b"\x00" in chunk
except (OSError, IOError):
return False
def get_file_icon(filepath):
"""Get a simple icon/class for file type."""
ext = filepath.suffix.lower()
mime, _ = mimetypes.guess_type(str(filepath))
if filepath.is_dir():
return "📁"
if mime and mime.startswith("image/"):
return "🖼️"
elif mime and mime.startswith("video/"):
return "🎬"
elif mime and mime.startswith("audio/"):
return "🎵"
elif ext in (".py", ".js", ".ts", ".sh", ".rb"):
return "📜"
elif ext in (".md", ".txt", ".rst"):
return "📝"
elif ext in (".json", ".yaml", ".yml", ".toml", ".xml", ".csv"):
return "📊"
elif ext in (".pdf", ".doc", ".docx"):
return "📄"
elif ext in (".zip", ".tar", ".gz", ".bz2", ".xz"):
return "📦"
elif ext in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"):
return "🖼️"
else:
return "📎"
def get_hidden_dirs():
"""Get hidden directories to skip in listings."""
return {
".git", ".venv", "__pycache__", ".cache", ".local",
".config", ".npm", ".node-gyp", ".Trash",
}
# --- Markdown Processing ---
def parse_yaml_frontmatter(content):
"""Extract YAML frontmatter and return (metadata_dict, body_text)."""
metadata = {}
body = content
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
yaml_str = parts[1].strip()
body = parts[2].strip()
for line in yaml_str.split("\n"):
if ":" in line:
key, _, value = line.partition(":")
metadata[key.strip()] = value.strip().strip('"').strip("'")
return metadata, body
def render_markdown(content):
"""Render Markdown with wikilink support."""
metadata, body = parse_yaml_frontmatter(content)
def wikilink_replacer(match):
title = match.group(1)
slug = title.replace(" ", "-").replace("_", "-")
return f'<a href="/wiki/{slug}" class="wikilink">[[{title}]]</a>'
body = re.sub(r'\[\[([^\]]+)\]\]', wikilink_replacer, body)
html = markdown.markdown(
body,
extensions=MD_EXTENSIONS,
extension_configs={
"codehilite": {"use_pygments": True, "noclasses": True, "linenums": False},
},
)
return html, metadata
# --- Authentication ---
def auth_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not BASIC_AUTH_USER:
return f(*args, **kwargs)
auth = request.authorization
if not auth or not (auth.username == BASIC_AUTH_USER and auth.password == BASIC_AUTH_PASS):
return abort(401, description="Authentication required")
return f(*args, **kwargs)
return decorated
# --- Slug helpers ---
def title_to_slug(title):
return title.replace(" ", "-").replace("_", "-")
def slug_to_title(slug):
return slug.replace("-", " ").replace("_", " ").title()
def find_wiki_page(slug):
if not WIKI_DIR.exists():
return None
slug_lower = slug.lower()
candidates = []
for f in WIKI_DIR.glob("*.md"):
name = f.stem
if name.lower() == slug_lower:
return f
candidates.append((f, abs(len(name) - len(slug))))
if candidates:
candidates.sort(key=lambda x: x[1])
if candidates[0][1] <= 3:
return candidates[0][0]
return None
def get_wiki_files():
if not WIKI_DIR.exists():
return []
files = []
for f in WIKI_DIR.glob("*.md"):
metadata, _ = parse_yaml_frontmatter(f.read_text())
files.append({
"path": f,
"slug": f.stem,
"title": metadata.get("title", f.stem.replace("-", " ").title()),
"tags": metadata.get("tags", ""),
"created": metadata.get("created", ""),
"updated": metadata.get("updated", ""),
"mtime": f.stat().st_mtime,
})
files.sort(key=lambda x: x["mtime"], reverse=True)
return files
def search_wiki(query):
"""Search the wiki using FTS5 index."""
from vault_index import search as fts_search
return fts_search(query)
def search_files(query, path=None):
"""Search file names and content across home directory."""
if not query or len(query) < 2:
return []
search_root = path or BROWSE_DIR
query_lower = query.lower()
results = []
skip_dirs = get_hidden_dirs()
def search_dir(directory, depth=0):
if depth > 3: # Limit depth
return
try:
for item in sorted(directory.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
if item.is_dir():
if item.name.startswith(".") and item.name in skip_dirs:
continue
if item.name.startswith("."):
continue
search_dir(item, depth + 1)
else:
name = item.name.lower()
score = 0
content_match = False
if query_lower in name:
score += 10
if query_lower in name.split(".")[0]:
score += 5
# Search text file content
if is_text_file(item) and item.stat().st_size < MAX_FILE_VIEW_SIZE:
try:
content = item.read_text(encoding="utf-8", errors="ignore")
if query_lower in content.lower():
score += 2
content_match = True
idx = content.lower().find(query_lower)
start = max(0, idx - 40)
end = min(len(content), idx + len(query) + 60)
snippet = content[start:end].replace("\n", " ")
if start > 0:
snippet = "..." + snippet
else:
snippet = ""
except (OSError, UnicodeDecodeError):
snippet = ""
else:
snippet = ""
if score > 0:
results.append({
"path": str(item.relative_to(BROWSE_DIR)),
"name": item.name,
"size": item.stat().st_size,
"size_str": format_size(item.stat().st_size),
"mtime": item.stat().st_mtime,
"mtime_str": format_date(item.stat().st_mtime),
"score": score,
"snippet": snippet,
"is_text": is_text_file(item) if not content_match else True,
})
except PermissionError:
pass
search_dir(search_root)
results.sort(key=lambda x: x["score"], reverse=True)
return results[:100] # Limit results
def get_all_tags():
"""Return list of (tag, count) tuples sorted by count desc, then name."""
tag_counts = {}
for f in WIKI_DIR.glob("*.md"):
content = f.read_text()
metadata, _ = parse_yaml_frontmatter(content)
tags_str = metadata.get("tags", "")
if tags_str:
clean = tags_str.strip("[]")
for tag in clean.split(","):
tag = tag.strip().strip("'\"")
if tag:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
return sorted(tag_counts.items(), key=lambda x: (-x[1], x[0].lower()))
def filter_pages_by_tag(pages, tag):
"""Filter wiki pages by tag."""
if not tag:
return pages
tag_lower = tag.lower()
filtered = []
for p in pages:
page_tags = p.get("tags", "")
clean = page_tags.strip("[]")
for t in clean.split(","):
if t.strip().lower() == tag_lower:
filtered.append(p)
break
return filtered
def get_daily_note_slug():
"""Return slug for today's daily note."""
return f"Daily-{datetime.now().strftime('%Y-%m-%d')}"
def create_daily_note():
"""Create or return existing daily note file for today."""
slug = get_daily_note_slug()
# Check file directly, bypass fuzzy matching which incorrectly matches other slugs
page_file = WIKI_DIR / f"{slug}.md"
if page_file.exists():
return page_file
now = datetime.now()
content = f"""---
title: Daily {now.strftime('%B %d, %Y')}
tags: [daily]
created: {now.strftime('%Y-%m-%d')}
updated: {now.strftime('%Y-%m-%d')}
---
# Daily {now.strftime('%B %d, %Y')}
## Morning
- [ ]
## Afternoon
- [ ]
## Notes
"""
page_file.write_text(content)
return page_file
def get_backlinks(target_slug):
"""Find all wiki pages that link to the target page."""
target_lower = target_slug.lower()
backlinks = []
for f in WIKI_DIR.glob("*.md"):
if f.stem.lower() == target_lower:
continue
try:
content = f.read_text()
# Check wikilinks: [[Page Name]]
wikilinks = re.findall(r'\[\[([^\]]+)\]\]', content)
matched = False
for link_title in wikilinks:
link_slug = link_title.replace(" ", "-").replace("_", "-").lower()
if link_slug == target_lower:
matched = True
break
# Check markdown links: [text](/wiki/slug)
if not matched:
md_links = re.findall(r'\]\(/wiki/([^)]+)\)', content)
for link_slug in md_links:
if link_slug.lower() == target_lower:
matched = True
break
if matched:
metadata, body = parse_yaml_frontmatter(content)
title = metadata.get("title", f.stem.replace("-", " ").title())
# Get a snippet around the link
snippet = ""
for link_title in wikilinks:
link_slug = link_title.replace(" ", "-").replace("_", "-").lower()
if link_slug == target_lower:
idx = body.find(f"[[{link_title}]]")
if idx >= 0:
start = max(0, idx - 40)
end = min(len(body), idx + len(link_title) + 40)
snippet = body[start:end].replace("\n", " ").strip()
if start > 0:
snippet = "..." + snippet
break
backlinks.append({
"slug": f.stem,
"title": title,
"snippet": snippet,
})
except Exception:
pass
return backlinks
def get_outlinks(slug):
"""Find all wikilinks from a page that point to existing wiki pages."""
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
return []
content = page_file.read_text()
wikilinks = re.findall(r'\[\[([^\]]+)\]\]', content)
outlinks = []
for link_title in wikilinks:
link_slug = link_title.replace(" ", "-").replace("_", "-")
target = find_wiki_page(link_slug)
if target:
metadata, _ = parse_yaml_frontmatter(target.read_text())
outlinks.append({
"slug": link_slug,
"title": metadata.get("title", link_slug.replace("-", " ").title()),
"display_title": link_title,
})
# Deduplicate by slug
seen = set()
unique = []
for link in outlinks:
if link["slug"] not in seen:
seen.add(link["slug"])
unique.append(link)
return unique
# --- Wiki Routes ---
@app.route("/")
@auth_required
def index():
pages = get_wiki_files()
tags = get_all_tags()
query = request.args.get("q", "")
tag_filter = request.args.get("tag", "")
results = search_wiki(query) if query else []
# Apply tag filter
display_pages = results if results else pages
if tag_filter:
display_pages = filter_pages_by_tag(display_pages, tag_filter)
return render_template(
"index.html",
pages=display_pages,
tags=tags,
query=query,
tag_filter=tag_filter,
)
@app.route("/daily")
@auth_required
def daily_note():
"""Redirect to today's daily note, creating it if needed."""
page_file = create_daily_note()
return redirect(url_for("view_page", slug=page_file.stem))
@app.route("/new")
@auth_required
def new_page():
"""Redirect to create a new page."""
return redirect(url_for("edit_page", slug="new"))
@app.route("/manifest.json")
def manifest():
"""Return PWA manifest."""
return send_file("static/manifest.json", mimetype="application/manifest+json")
@app.route("/thumbnail/<path:filepath>")
@auth_required
def thumbnail(filepath):
"""Generate a thumbnail for images."""
import io
from PIL import Image, ImageOps
file_path = safe_path(HOME_DIR, filepath)
if not file_path or not file_path.exists():
abort(404)
try:
with Image.open(str(file_path)) as img:
ImageOps.exif_transpose(img)
img.thumbnail((400, 400))
buffer = io.BytesIO()
img.save(buffer, format="WEBP", quality=85)
buffer.seek(0)
return send_file(buffer, mimetype="image/webp")
except Exception:
abort(500)
# --- Hermes API ---
@app.route("/api/vault/search")
@auth_required
def api_vault_search():
"""Search the vault via API — returns JSON results."""
query = request.args.get("q", "")
if not query or len(query) < 2:
return jsonify({"results": [], "error": "Query too short"})
results = search_wiki(query)
return jsonify({
"query": query,
"results": [
{
"slug": r.slug,
"title": r.title,
"snippet": r.snippet,
"rank": r.rank,
}
for r in results
],
})
@app.route("/api/vault/pages")
@auth_required
def api_vault_pages():
"""List all wiki pages via API."""
tag = request.args.get("tag", "")
pages = get_wiki_files()
if tag:
pages = filter_pages_by_tag(pages, tag)
return jsonify({
"pages": [
{
"slug": p["slug"],
"title": p["title"],
"tags": p["tags"],
"updated": p["updated"],
}
for p in pages
],
})
@app.route("/api/vault/page/<slug>")
@auth_required
def api_vault_page(slug):
"""Get a wiki page's content via API."""
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
return jsonify({"error": "Page not found"}), 404
content = page_file.read_text()
metadata, body = parse_yaml_frontmatter(content)
html, _ = render_markdown(content)
return jsonify({
"slug": slug,
"title": metadata.get("title", ""),
"tags": metadata.get("tags", ""),
"created": metadata.get("created", ""),
"updated": metadata.get("updated", ""),
"markdown": body,
"html": html,
"backlinks": get_backlinks(slug),
"outlinks": get_outlinks(slug),
})
@app.route("/api/vault/tags")
@auth_required
def api_vault_tags():
"""List all tags with counts via API."""
return jsonify({
"tags": [{"tag": t, "count": c} for t, c in get_all_tags()],
})
@app.route("/api/vault/graph")
@auth_required
def api_vault_graph():
"""Knowledge graph data via API."""
return jsonify(build_graph_data())
@app.route("/wiki/<slug>")
@auth_required
def view_page(slug):
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
abort(404, description=f"Page '{slug}' not found")
content = page_file.read_text()
html, metadata = render_markdown(content)
page_hash = hashlib.md5(content.encode()).hexdigest()[:12]
backlinks = get_backlinks(slug)
outlinks = get_outlinks(slug)
return render_template(
"view.html",
slug=slug,
title=metadata.get("title", slug_to_title(slug)),
html=html,
metadata=metadata,
page_hash=page_hash,
tags=get_all_tags(),
backlinks=backlinks,
outlinks=outlinks,
)
@app.route("/wiki/<slug>/edit", methods=["GET", "POST"])
@auth_required
def edit_page(slug):
page_file = find_wiki_page(slug)
if request.method == "POST":
new_content = request.form.get("content", "")
if page_file and page_file.exists():
# Save version before overwriting
try:
old_content = page_file.read_text()
if old_content and old_content.strip():
from version_history import save_version
save_version(slug, old_content)
except Exception:
logging.exception("Failed to save version for: %s", slug)
page_file.write_text(new_content)
elif new_content.strip():
new_file = WIKI_DIR / f"{slug}.md"
new_file.write_text(new_content)
page_file = new_file
# Rebuild FTS5 index after save
try:
from vault_index import rebuild_index
rebuild_index()
except Exception:
logging.exception("Failed to reindex wiki")
return redirect(url_for("view_page", slug=slug))
if page_file and page_file.exists():
content = page_file.read_text()
else:
metadata = {
"title": slug_to_title(slug),
"tags": [],
"created": datetime.now().strftime("%Y-%m-%d"),
"updated": datetime.now().strftime("%Y-%m-%d"),
}
frontmatter = "\n".join(f"{k}: {v}" for k, v in metadata.items())
content = f"---\n{frontmatter}\n---\n\n# {metadata['title']}\n\n"
return render_template("edit.html", slug=slug, content=content, title=slug_to_title(slug))
@app.route("/wiki/<slug>/history")
@auth_required
def view_history(slug):
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
abort(404, description=f"Page '{slug}' not found")
from version_history import list_versions, get_version
versions = list_versions(slug)
current_hash = hashlib.md5(page_file.read_text().encode()).hexdigest()[:12]
return render_template(
"history.html",
slug=slug,
title=slug_to_title(slug),
versions=versions,
current_hash=current_hash,
)
@app.route("/wiki/<slug>/restore", methods=["POST"])
@auth_required
def restore_version(slug):
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
abort(404, description=f"Page '{slug}' not found")
ts = float(request.form.get("timestamp"))
from version_history import restore_version as restore, save_version
# Save current content as version before restoring
save_version(slug, page_file.read_text())
restore(slug, ts)
# Rebuild index
try:
from vault_index import rebuild_index
rebuild_index()
except Exception:
pass
return redirect(url_for("view_page", slug=slug))
@app.route("/search")
@auth_required
def search():
query = request.args.get("q", "")
type_ = request.args.get("type", "wiki") # wiki or files
if type_ == "files":
results = search_files(query)
else:
results = search_wiki(query)
return render_template("search.html", results=results, query=query, type=type_)
@app.route("/api/search")
@auth_required
def api_search():
query = request.args.get("q", "")
results = search_wiki(query) if query else []
return jsonify([
{"slug": r["slug"], "title": r["title"], "tags": r["tags"], "snippet": r["snippet"]}
for r in results[:20]
])
@app.route("/api/page/<slug>")
@auth_required
def api_page(slug):
page_file = find_wiki_page(slug)
if not page_file or not page_file.exists():
return jsonify({"error": "not found"}), 404
content = page_file.read_text()
html, metadata = render_markdown(content)
page_hash = hashlib.md5(content.encode()).hexdigest()[:12]
return jsonify({
"slug": slug,
"title": metadata.get("title", slug_to_title(slug)),
"html": html,
"hash": page_hash,
"metadata": metadata,
})
# --- File System Routes ---
@app.route("/files")
@app.route("/files/<path:subpath>")
@auth_required
def browse_files(subpath=None):
"""Browse the home directory."""
if subpath:
current_path = safe_path(BROWSE_DIR, subpath)
else:
current_path = BROWSE_DIR
if not current_path or not current_path.exists():
abort(404, description="Path not found")
if current_path.is_file():
return redirect(url_for("view_file", path=current_path.relative_to(BROWSE_DIR)))
items = []
skip_dirs = get_hidden_dirs()
for item in current_path.iterdir():
# Skip hidden files/dirs
if item.name.startswith(".") and item.name not in (".", ".."):
continue
if item.is_dir() and item.name in skip_dirs:
continue
try:
stat = item.stat()
rel_path = str(item.relative_to(BROWSE_DIR))
item_data = {
"name": item.name,
"path": rel_path,
"is_dir": item.is_dir(),
"size": stat.st_size,
"size_str": format_size(stat.st_size),
"mtime": stat.st_mtime,
"mtime_str": format_date(stat.st_mtime),
"icon": get_file_icon(item),
"is_text": is_text_file(item) if not item.is_dir() else None,
}
# Add thumbnail URL for images
if not item.is_dir() and item.suffix.lower() in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".bmp", ".tiff"):
item_data["thumbnail"] = url_for("thumbnail", filepath=rel_path, _external=False)
items.append(item_data)
except (PermissionError, OSError):
continue
# Sort: directories first, then files alphabetically
items.sort(key=lambda x: (not x["is_dir"], x["name"].lower()))
# Build breadcrumbs
breadcrumbs = []
parts = []
if subpath:
parts = subpath.strip("/").split("/")
for i, part in enumerate(parts):
rel = "/".join(parts[:i+1])
breadcrumbs.append({"name": part, "path": rel})
# Home directory name for display
home_name = HOME_DIR.name if BROWSE_DIR == HOME_DIR else BROWSE_DIR.name
return render_template(
"files.html",
items=items,
current_path=str(current_path),
rel_path=subpath or "",
home_name=home_name,
breadcrumbs=breadcrumbs,
)
@app.route("/view/<path:filepath>")
@auth_required
def view_file(filepath):
"""View a text file in the browser."""
file_path = safe_path(BROWSE_DIR, filepath)
if not file_path or not file_path.exists() or not file_path.is_file():
abort(404, description="File not found")
# Check file size
if file_path.stat().st_size > MAX_FILE_VIEW_SIZE:
return render_template("error.html", code=413,
message=f"File too large to view ({format_size(file_path.stat().st_size)}). Download instead."), 413
# Check if text file
if not is_text_file(file_path):
return render_template("error.html", code=400,
message="Binary file. Use download link instead."), 400
try:
content = file_path.read_text(encoding="utf-8", errors="replace")
except (OSError, UnicodeDecodeError):
abort(404, description="Could not read file")
# Determine language for syntax highlighting
ext = file_path.suffix.lower()
lang_map = {
".py": "python", ".js": "javascript", ".ts": "typescript",
".sh": "bash", ".bash": "bash", ".zsh": "bash",
".html": "html", ".css": "css", ".json": "json",
".yaml": "yaml", ".yml": "yaml", ".toml": "toml",
".md": "markdown", ".rst": "rst", ".txt": "",
".xml": "xml", ".csv": "csv", ".sql": "sql",
".rb": "ruby", ".go": "go", ".rs": "rust",
".java": "java", ".c": "c", ".cpp": "cpp", ".h": "c",
".swift": "swift", ".kt": "kotlin",
".cfg": "ini", ".conf": "ini", ".ini": "ini",
}
lang = lang_map.get(ext, "")
# Build breadcrumbs
breadcrumbs = []
parts = filepath.strip("/").split("/")
for i, part in enumerate(parts[:-1]):
rel = "/".join(parts[:i+1])
breadcrumbs.append({"name": part, "path": rel})
return render_template(
"file_view.html",
content=content,
filename=file_path.name,
filepath=filepath,
lang=lang,
size=format_size(file_path.stat().st_size),
mtime=format_date(file_path.stat().st_mtime),
breadcrumbs=breadcrumbs,
)
@app.route("/download/<path:filepath>")
@auth_required
def download_file(filepath):
"""Download a file."""
file_path = safe_path(BROWSE_DIR, filepath)
if not file_path or not file_path.exists() or not file_path.is_file():
abort(404, description="File not found")
return send_file(str(file_path), as_attachment=True, download_name=file_path.name)
@app.route("/api/files/search")
@auth_required
def api_file_search():
"""API endpoint for file search."""
query = request.args.get("q", "")
results = search_files(query)
return jsonify([
{
"name": r["name"],
"path": r["path"],
"size_str": r["size_str"],
"snippet": r.get("snippet", ""),
}
for r in results
])
@app.route("/upload", methods=["POST"])
@auth_required
def upload_file():
"""Upload a file to the home directory."""
if "file" not in request.files:
return jsonify({"error": "No file provided"}), 400
file = request.files["file"]
if file.filename == "":
return jsonify({"error": "No file selected"}), 400
# Determine upload path
target_dir = request.form.get("path", "")
if target_dir:
upload_dir = safe_path(BROWSE_DIR, target_dir)
else:
upload_dir = BROWSE_DIR
if not upload_dir or not upload_dir.exists():
return jsonify({"error": "Invalid upload path"}), 400
filepath = safe_path(upload_dir, file.filename)
if not filepath:
return jsonify({"error": "Invalid filename"}), 400
file.save(str(filepath))
return jsonify({
"success": True,
"path": str(filepath.relative_to(BROWSE_DIR)),
"name": file.filename,
})
# --- Error Handlers ---
@app.errorhandler(404)
def not_found(e):
return render_template("error.html", code=404, message="Not found"), 404
@app.errorhandler(401)
def unauthorized(e):
return e
@app.errorhandler(413)
def file_too_large(e):
return render_template("error.html", code=413, message="File too large"), 413
@app.errorhandler(500)
def server_error(e):
return render_template("error.html", code=500, message="Internal server error"), 500
# --- Main ---
# --- Knowledge Graph ---
def build_graph_data():
"""Build node/edge data for the knowledge graph visualization."""
nodes = []
edges = []
seen_nodes = {}
for f in WIKI_DIR.glob("*.md"):
slug = f.stem
metadata, _ = parse_yaml_frontmatter(f.read_text())
title = metadata.get("title", slug.replace("-", " ").title())
tags = metadata.get("tags", "")
clean_tags = tags.strip("[]")
tag_list = [t.strip().strip("'\"") for t in clean_tags.split(",") if t.strip()]
if slug not in seen_nodes:
seen_nodes[slug] = len(nodes)
nodes.append({
"id": slug,
"title": title,
"tags": tag_list,
"size": f.stat().st_size,
})
# Find links to other pages
content = f.read_text()
wikilinks = re.findall(r'\[\[([^\]]+)\]\]', content)
md_links = re.findall(r'\]\(/wiki/([^)]+)\)', content)
for link_slug in [w.replace(" ", "-").replace("_", "-") for w in wikilinks] + md_links:
if link_slug.lower() != slug.lower():
edges.append({
"source": slug,
"target": link_slug,
"type": "wikilink" if link_slug.lower() in [w.replace(" ", "-").lower() for w in wikilinks] else "mdlink",
})
# Deduplicate edges
unique_edges = []
seen_edges = set()
for e in edges:
key = (e["source"], e["target"])
if key not in seen_edges and e["source"] != e["target"]:
seen_edges.add(key)
unique_edges.append(e)
return {"nodes": nodes, "edges": unique_edges}
@app.route("/graph")
@auth_required
def graph_page():
"""Render the knowledge graph visualization."""
return render_template("graph.html", active_tab="graph")
@app.route("/api/graph")
@auth_required
def graph_api():
"""Return graph data as JSON."""
return jsonify(build_graph_data())
# --- Main ---
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
print(f"🧠 Second Brain Vault")
print(f" Vault: {VAULT_DIR}")
print(f" Wiki: {WIKI_DIR}")
print(f" Files: {BROWSE_DIR}")
print(f" URL: http://{HOST}:{PORT}/")
if BASIC_AUTH_USER:
print(f" Auth: {BASIC_AUTH_USER}")
# Build/rebuild FTS5 search index on startup
try:
from vault_index import rebuild_index
rebuild_index()
print(" Search: FTS5 index ready")
except Exception:
print(" Search: FTS5 index failed (search will still work)")
app.run(host=HOST, port=PORT, debug=False)