claude-configs/skills/notediscovery/client.py

#!/usr/bin/env python3
"""
NoteDiscovery Client - Lightweight HTTP client for NoteDiscovery knowledge base.

Usage:
    from client import NoteDiscoveryClient

    client = NoteDiscoveryClient()
    results = client.search("tdarr")
    note = client.read_note("homelab/tdarr-notes")
    client.save_note("homelab/new-note", "# Content")
"""

import json
import os
from pathlib import Path
from typing import Any

import httpx

# Auto-load .env from skill directory if not already in environment
_env_file = Path(__file__).parent / ".env"
if _env_file.exists():
    for line in _env_file.read_text().strip().split("\n"):
        if "=" in line and not line.startswith("#"):
            key, value = line.split("=", 1)
            if key not in os.environ:
                os.environ[key] = value


class NoteDiscoveryClient:
    """HTTP client for NoteDiscovery API with session-based auth."""

    def __init__(
        self,
        base_url: str | None = None,
        password: str | None = None,
        timeout: float = 30.0,
    ):
        """
        Initialize client.

        Args:
            base_url: NoteDiscovery URL (default: from NOTEDISCOVERY_URL env)
            password: Auth password (default: from NOTEDISCOVERY_PASSWORD env)
            timeout: Request timeout in seconds
        """
        self.base_url = (base_url or os.environ.get("NOTEDISCOVERY_URL", "https://notes.manticorum.com")).rstrip("/")
        self.password = password or os.environ.get("NOTEDISCOVERY_PASSWORD", "")
        self.client = httpx.Client(timeout=timeout, follow_redirects=True)
        self._authenticated = False

    def _ensure_auth(self) -> None:
        """Authenticate if not already done."""
        if self._authenticated:
            return

        if not self.password:
            raise ValueError("No password configured - set NOTEDISCOVERY_PASSWORD environment variable")

        response = self.client.post(
            f"{self.base_url}/login",
            data={"password": self.password},
            follow_redirects=False,
        )

        if response.status_code in (200, 302, 303):
            self._authenticated = True
        else:
            raise ConnectionError(f"Authentication failed: {response.status_code}")

    def _request(self, method: str, endpoint: str, **kwargs) -> dict | list | str:
        """Make authenticated request to API."""
        self._ensure_auth()

        url = f"{self.base_url}{endpoint}"
        response = self.client.request(method, url, **kwargs)

        # Re-auth on 401
        if response.status_code == 401:
            self._authenticated = False
            self._ensure_auth()
            response = self.client.request(method, url, **kwargs)

        if response.status_code >= 400:
            raise httpx.HTTPStatusError(
                f"HTTP {response.status_code}: {response.text}",
                request=response.request,
                response=response,
            )

        # Try JSON first, fall back to text
        try:
            return response.json()
        except json.JSONDecodeError:
            return response.text

    # -------------------------------------------------------------------------
    # Core Operations
    # -------------------------------------------------------------------------

    def search(self, query: str) -> dict:
        """
        Search notes by keyword.

        Args:
            query: Search query (searches titles and content)

        Returns:
            Dict with search results
        """
        return self._request("GET", "/api/search", params={"q": query})

    def list_notes(self) -> dict:
        """
        Get all notes with metadata.

        Returns:
            Dict with hierarchical note structure
        """
        return self._request("GET", "/api/notes")

    def read_note(self, path: str) -> dict:
        """
        Get content of a specific note.

        Args:
            path: Note path (e.g., 'folder/note-name', without .md extension)

        Returns:
            Dict with note content and metadata
        """
        return self._request("GET", f"/api/notes/{path}")

    def save_note(self, path: str, content: str) -> dict:
        """
        Create or update a note.

        Args:
            path: Note path (creates folders if needed)
            content: Markdown content (include # heading as title)

        Returns:
            Dict with operation result
        """
        return self._request("POST", f"/api/notes/{path}", json={"content": content})

    def delete_note(self, path: str) -> dict:
        """
        Delete a note.

        Args:
            path: Note path to delete

        Returns:
            Dict with operation result
        """
        return self._request("DELETE", f"/api/notes/{path}")

    # -------------------------------------------------------------------------
    # Tag Operations
    # -------------------------------------------------------------------------

    def list_tags(self) -> dict:
        """
        Get all tags with note counts.

        Returns:
            Dict with tag names and counts
        """
        return self._request("GET", "/api/tags")

    def get_notes_by_tag(self, tag: str) -> dict:
        """
        Get all notes with a specific tag.

        Args:
            tag: Tag name (without #)

        Returns:
            Dict with matching notes
        """
        return self._request("GET", f"/api/tags/{tag}")

    # -------------------------------------------------------------------------
    # Graph Operations
    # -------------------------------------------------------------------------

    def get_graph(self) -> dict:
        """
        Get note relationship graph.

        Returns:
            Dict with nodes and edges representing note links
        """
        return self._request("GET", "/api/graph")

    # -------------------------------------------------------------------------
    # Folder Operations
    # -------------------------------------------------------------------------

    def create_folder(self, path: str) -> dict:
        """
        Create a new folder.

        Args:
            path: Folder path to create

        Returns:
            Dict with operation result
        """
        return self._request("POST", "/api/folders", json={"path": path})

    # -------------------------------------------------------------------------
    # Convenience Methods
    # -------------------------------------------------------------------------

    def note_exists(self, path: str) -> bool:
        """Check if a note exists at the given path."""
        try:
            self.read_note(path)
            return True
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                return False
            raise

    def append_to_note(self, path: str, content: str, separator: str = "\n\n---\n\n") -> dict:
        """
        Append content to an existing note.

        Args:
            path: Note path
            content: Content to append
            separator: Separator between existing and new content

        Returns:
            Dict with operation result
        """
        existing = self.read_note(path)
        existing_content = existing.get("content", "")
        new_content = f"{existing_content}{separator}{content}"
        return self.save_note(path, new_content)

    def search_or_create(
        self,
        search_query: str,
        create_path: str,
        create_content: str,
        similarity_threshold: float = 0.8,
    ) -> tuple[str, dict]:
        """
        Search for existing note, create if not found.

        Args:
            search_query: Query to search for existing notes
            create_path: Path for new note if not found
            create_content: Content for new note
            similarity_threshold: Not implemented yet (for future fuzzy matching)

        Returns:
            Tuple of (action, result) where action is 'found', 'created', or 'updated'
        """
        results = self.search(search_query)

        # Check if we have matches
        if results and isinstance(results, dict):
            notes = results.get("results", results.get("notes", []))
            if notes:
                # Return first match
                return ("found", notes[0])

        # Create new note
        result = self.save_note(create_path, create_content)
        return ("created", result)

    def close(self) -> None:
        """Close the HTTP client."""
        self.client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()


# CLI for testing
if __name__ == "__main__":
    import sys

    client = NoteDiscoveryClient()

    if len(sys.argv) < 2:
        print("Usage: python client.py <command> [args...]")
        print("Commands: search, read, list, tags, save")
        sys.exit(1)

    command = sys.argv[1]

    if command == "search" and len(sys.argv) > 2:
        result = client.search(" ".join(sys.argv[2:]))
        print(json.dumps(result, indent=2))

    elif command == "read" and len(sys.argv) > 2:
        result = client.read_note(sys.argv[2])
        print(json.dumps(result, indent=2))

    elif command == "list":
        result = client.list_notes()
        print(json.dumps(result, indent=2))

    elif command == "tags":
        result = client.list_tags()
        print(json.dumps(result, indent=2))

    elif command == "save" and len(sys.argv) > 3:
        path = sys.argv[2]
        content = " ".join(sys.argv[3:])
        result = client.save_note(path, content)
        print(json.dumps(result, indent=2))

    else:
        print(f"Unknown command or missing arguments: {command}")
        sys.exit(1)