hermes-wiki-static/server.py

"""
Obsidian Web Viewer — Browse Your Vault in the Browser
========================================================
Web-based Obsidian vault browser with 3D graph, file tree, markdown
rendering, wikilink navigation, and full-text search.

Usage:
    python server.py --vault /path/to/your/vault
    python server.py --vault ~/Documents/MyVault --port 8080
"""
import sys
import os
import json
import re
import argparse
import time
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from datetime import datetime
from urllib.parse import unquote, quote

if sys.platform == 'win32':
    try:
        sys.stdout.reconfigure(encoding='utf-8')
    except Exception:
        pass

ROOT = Path(__file__).parent
VAULT_DIR = None  # Set via --vault CLI arg

# YAML parsing (optional)
_YAML_AVAILABLE = False
try:
    import yaml
    _YAML_AVAILABLE = True
except ImportError:
    pass

MIME = {
    ".html": "text/html; charset=utf-8",
    ".css": "text/css",
    ".js": "application/javascript",
    ".json": "application/json",
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".svg": "image/svg+xml",
}

# Cache for vault tree and graph
_cache = {"tree": None, "graph": None, "ts": 0}
_CACHE_TTL = 300


def _build_tree(vault_dir: Path) -> dict:
    """Build a JSON file tree of the vault directory."""
    now = time.time()
    if _cache["tree"] and (now - _cache["ts"]) < _CACHE_TTL:
        return _cache["tree"]

    def _walk(d: Path) -> dict:
        node = {"type": "folder", "name": d.name, "children": []}
        try:
            entries = sorted(d.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower()))
        except PermissionError:
            return node
        for entry in entries:
            if entry.name.startswith('.') or entry.name.startswith('_'):
                continue
            if entry.is_dir():
                child = _walk(entry)
                if child["children"]:
                    node["children"].append(child)
            elif entry.suffix.lower() == '.md':
                node["children"].append({"type": "file", "name": entry.name})
        return node

    if not vault_dir.exists():
        return {"type": "folder", "name": "", "children": []}
    tree = _walk(vault_dir)
    tree["name"] = ""
    _cache["tree"] = tree
    _cache["ts"] = time.time()
    return tree


def _parse_frontmatter(raw: str) -> tuple:
    """Parse YAML frontmatter from markdown. Returns (dict, body_str)."""
    if not raw.startswith('---'):
        return {}, raw
    end = raw.find('---', 3)
    if end == -1:
        return {}, raw
    fm_text = raw[3:end].strip()
    body = raw[end + 3:].strip()
    fm = {}
    if _YAML_AVAILABLE:
        try:
            fm = yaml.safe_load(fm_text) or {}
            if not isinstance(fm, dict):
                fm = {"value": fm}
        except Exception:
            fm = {}
    else:
        for line in fm_text.splitlines():
            if ':' in line:
                k, _, v = line.partition(':')
                v = v.strip().strip('"').strip("'")
                if v.startswith('[') and v.endswith(']'):
                    v = [x.strip().strip('"').strip("'") for x in v[1:-1].split(',') if x.strip()]
                fm[k.strip()] = v
    return fm, body


def _search(vault_dir: Path, query: str, max_results: int = 30) -> list:
    """Search vault files by name and content."""
    results = []
    query_lower = query.lower()
    for md_file in vault_dir.rglob("*.md"):
        if md_file.name.startswith('.') or md_file.name.startswith('_'):
            continue
        try:
            rel = md_file.relative_to(vault_dir).as_posix()
        except ValueError:
            continue
        name = md_file.stem
        name_match = query_lower in name.lower()
        snippet = ""
        content_match = False
        if not name_match:
            try:
                text = md_file.read_text(encoding="utf-8", errors="ignore")[:5000]
                idx = text.lower().find(query_lower)
                if idx >= 0:
                    content_match = True
                    start = max(0, idx - 40)
                    end = min(len(text), idx + len(query) + 60)
                    snippet = ("..." if start > 0 else "") + text[start:end].replace("\n", " ") + ("..." if end < len(text) else "")
            except Exception:
                continue
        if name_match or content_match:
            results.append({
                "path": rel, "name": name,
                "folder": str(md_file.parent.relative_to(vault_dir)) if md_file.parent != vault_dir else "",
                "snippet": snippet, "name_match": name_match,
            })
        if len(results) >= max_results:
            break
    results.sort(key=lambda r: (not r["name_match"], r["name"].lower()))
    return results


def _build_graph(vault_dir: Path) -> dict:
    """Build a graph of wikilink connections between vault notes."""
    if _cache["graph"] and (time.time() - _cache["ts"]) < _CACHE_TTL:
        return _cache["graph"]

    wikilink_re = re.compile(r'\[\[([^\]|]+?)(?:\|[^\]]+?)?\]\]')
    nodes = []
    edges = []
    name_to_path = {}
    all_paths = []

    for md_file in vault_dir.rglob("*.md"):
        if md_file.name.startswith('.') or md_file.name.startswith('_'):
            continue
        try:
            rel = md_file.relative_to(vault_dir).as_posix()
        except ValueError:
            continue
        all_paths.append((rel, md_file))
        name_to_path[md_file.stem.lower()] = rel

    for rel, md_file in all_paths:
        try:
            text = md_file.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            text = ""

        # Classify by folder for coloring
        rp = rel.lower()
        node_type = "other"
        for folder in ["wiki", "reference", "daily", "journal", "notes", "projects"]:
            if rp.startswith(folder + "/"):
                node_type = folder
                break

        nodes.append({"id": rel, "label": md_file.stem, "type": node_type})
        links = wikilink_re.findall(text)
        seen = set()
        for target in links:
            resolved = name_to_path.get(target.strip().lower())
            if resolved and resolved != rel and resolved not in seen:
                edges.append({"source": rel, "target": resolved})
                seen.add(resolved)

    result = {"nodes": nodes, "edges": edges}
    _cache["graph"] = result
    return result


class VaultHandler(BaseHTTPRequestHandler):
    def log_message(self, *a): pass

    def do_GET(self):
        path = self.path.split("?")[0].rstrip("/")
        query_string = self.path.split("?")[1] if "?" in self.path else ""

        if path == "" or path == "/" or path == "/vault":
            self._serve_file(ROOT / "vault.html")
        elif path == "/api/vault/tree":
            self._json(_build_tree(VAULT_DIR))
        elif path == "/api/vault/graph":
            self._json(_build_graph(VAULT_DIR))
        elif path == "/api/vault/search":
            import urllib.parse
            params = urllib.parse.parse_qs(query_string)
            q = params.get("q", [""])[0]
            self._json({"results": _search(VAULT_DIR, q) if q else []})
        elif path.startswith("/api/vault/file/"):
            rel = unquote(path[len("/api/vault/file/"):])
            # Lukas-add: also support ?file=<path> query for clean URL navigation
            if "?" in rel:
                rel, _, query = rel.partition("?")
                from urllib.parse import parse_qs
                file_param = parse_qs(query).get("file", [None])[0]
                if file_param:
                    rel = file_param
            file_path = VAULT_DIR / rel
            if not file_path.exists() or not str(file_path).startswith(str(VAULT_DIR)):
                self._json({"error": "Not found"}, 404)
                return
            try:
                raw = file_path.read_text(encoding="utf-8")
                fm, body = _parse_frontmatter(raw)
                self._json({"path": rel, "name": file_path.stem, "frontmatter": fm, "body": body})
            except Exception as e:
                self._json({"error": str(e)}, 500)
        elif path == "/config.json":
            self._serve_file(ROOT / "config.json")
        else:
            f = ROOT / path.lstrip("/")
            if f.exists() and f.is_file():
                self._serve_file(f)
            else:
                self.send_error(404)

    def _serve_file(self, path):
        if not path.exists():
            self.send_error(404); return
        data = path.read_bytes()
        mime = MIME.get(path.suffix.lower(), "application/octet-stream")
        self.send_response(200)
        self.send_header("Content-Type", mime)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        self.wfile.write(data)

    def _json(self, obj, code=200):
        data = json.dumps(obj, indent=2, default=str).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        self.wfile.write(data)


def main():
    parser = argparse.ArgumentParser(description="Obsidian Web Viewer")
    parser.add_argument("--vault", required=True, help="Path to your Obsidian vault folder")
    parser.add_argument("--port", type=int, default=8765)
    parser.add_argument("--host", default="0.0.0.0")
    args = parser.parse_args()

    global VAULT_DIR
    VAULT_DIR = Path(args.vault).resolve()

    if not VAULT_DIR.exists():
        print(f"ERROR: Vault path does not exist: {VAULT_DIR}")
        sys.exit(1)

    md_count = sum(1 for _ in VAULT_DIR.rglob("*.md"))
    print(f"Obsidian Web Viewer")
    print(f"  Vault: {VAULT_DIR}")
    print(f"  Notes: {md_count}")
    print(f"  URL:   http://{args.host}:{args.port}")

    server = ThreadingHTTPServer((args.host, args.port), VaultHandler)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")


if __name__ == "__main__":
    main()