#!/usr/bin/env python3 """ outline-reader.py — Harvey's Outline KB reader Pulls documents from Outline API for use in Leah's context. Usage: python3 scripts/outline-reader.py --test python3 scripts/outline-reader.py --search "cancellation policy" python3 scripts/outline-reader.py --collection "Leah KB" python3 scripts/outline-reader.py --doc python3 scripts/outline-reader.py --export-leah # Full Leah KB export for context injection """ import sys import os import json import argparse import urllib.request import urllib.parse import urllib.error # ── Config ──────────────────────────────────────────────────────────────────── def get_config(): token_path = os.path.expanduser("~/.openclaw/secrets/outline-api-token.txt") if os.path.exists(token_path): token = open(token_path).read().strip() else: token = os.environ.get("OUTLINE_API_TOKEN", "") url = os.environ.get("OUTLINE_API_URL", "http://localhost:3300/api") return token, url def api_post(endpoint, payload, token, base_url): url = f"{base_url}/{endpoint.lstrip('/')}" data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url, data=data, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {token}", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") print(f"HTTP {e.code}: {body[:200]}", file=sys.stderr) return None except urllib.error.URLError as e: print(f"Connection error: {e.reason}", file=sys.stderr) print("Is Outline running? Try: docker compose up -d (in workspace/outline/)", file=sys.stderr) return None # ── Commands ────────────────────────────────────────────────────────────────── def cmd_test(token, base_url): """Verify connection and list collections.""" print(f"Testing connection to {base_url}...") result = api_post("collections.list", {"limit": 25}, token, base_url) if not result: print("❌ Connection failed.") return False if not result.get("ok"): print(f"❌ API error: {result}") return False collections = result.get("data", {}).get("data", []) print(f"✅ Connected. Found {len(collections)} collection(s):") for c in collections: print(f" - {c['name']} (id: {c['id']}, docs: {c.get('documentsCount', '?')})") return True def cmd_search(query, token, base_url, limit=10): """Full-text search across all documents.""" result = api_post("documents.search", { "query": query, "limit": limit, "includeArchived": False, }, token, base_url) if not result or not result.get("ok"): print(f"Search failed: {result}") return items = result.get("data", []) print(f"Found {len(items)} result(s) for '{query}':\n") for item in items: doc = item.get("document", {}) ctx = item.get("context", "") print(f"📄 {doc.get('title', 'Untitled')}") print(f" Collection: {doc.get('collectionId', '?')}") print(f" Context: ...{ctx[:200]}...") print() def get_collection_id_by_name(name, token, base_url): result = api_post("collections.list", {"limit": 25}, token, base_url) if not result or not result.get("ok"): return None for c in result.get("data", {}).get("data", []): if c["name"].lower() == name.lower(): return c["id"] return None def cmd_collection(name, token, base_url): """List all documents in a named collection.""" cid = get_collection_id_by_name(name, token, base_url) if not cid: print(f"Collection '{name}' not found.") return result = api_post("documents.list", { "collectionId": cid, "limit": 50, }, token, base_url) if not result or not result.get("ok"): print(f"Failed: {result}") return docs = result.get("data", []) print(f"Collection '{name}' — {len(docs)} document(s):\n") for doc in docs: print(f" 📄 {doc.get('title', 'Untitled')} (id: {doc['id']})") print(f" Updated: {doc.get('updatedAt', '?')[:10]}") def cmd_doc(doc_id, token, base_url): """Fetch a document's full text content.""" result = api_post("documents.info", {"id": doc_id}, token, base_url) if not result or not result.get("ok"): print(f"Failed: {result}") return doc = result.get("data", {}) print(f"# {doc.get('title', 'Untitled')}\n") print(doc.get("text", "(no content)")) def cmd_export_leah(token, base_url, output_file=None): """ Export the full Leah KB collection as plain text for context injection. Output suitable for pasting into Leah's system prompt or runtime context. """ cid = get_collection_id_by_name("Leah KB", token, base_url) if not cid: print("'Leah KB' collection not found. Create it in Outline first.") return result = api_post("documents.list", {"collectionId": cid, "limit": 100}, token, base_url) if not result or not result.get("ok"): print(f"Failed to list documents: {result}") return docs = result.get("data", []) if not docs: print("Leah KB is empty. Import content first.") return sections = [] for doc in docs: doc_result = api_post("documents.info", {"id": doc["id"]}, token, base_url) if not doc_result or not doc_result.get("ok"): continue d = doc_result.get("data", {}) title = d.get("title", "Untitled") text = d.get("text", "").strip() sections.append(f"## {title}\n\n{text}") combined = "# NMC Knowledge Base — Leah Context\n\n" + "\n\n---\n\n".join(sections) if output_file: with open(output_file, "w") as f: f.write(combined) print(f"✅ Exported {len(docs)} documents to {output_file}") else: print(combined) # ── Import helper ───────────────────────────────────────────────────────────── def cmd_import_text(title, text, collection_name, token, base_url, parent_id=None): """Create a document in Outline from plain text/markdown.""" cid = get_collection_id_by_name(collection_name, token, base_url) if not cid: print(f"Collection '{collection_name}' not found.") return None payload = { "title": title, "text": text, "collectionId": cid, "publish": True, } if parent_id: payload["parentDocumentId"] = parent_id result = api_post("documents.create", payload, token, base_url) if not result or not result.get("ok"): print(f"Failed to create '{title}': {result}") return None doc = result.get("data", {}) print(f"✅ Created: {doc.get('title')} (id: {doc.get('id')})") return doc.get("id") # ── Main ────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Harvey's Outline KB reader") parser.add_argument("--test", action="store_true", help="Test connection") parser.add_argument("--search", metavar="QUERY", help="Search documents") parser.add_argument("--collection", metavar="NAME", help="List docs in collection") parser.add_argument("--doc", metavar="ID", help="Fetch a document by ID") parser.add_argument("--export-leah", action="store_true", help="Export Leah KB as context") parser.add_argument("--output", metavar="FILE", help="Output file for --export-leah") parser.add_argument("--limit", type=int, default=10, help="Result limit for search") args = parser.parse_args() token, base_url = get_config() if not token: print("No API token found.") print("1. Start Outline: docker compose up -d (in workspace/outline/)") print("2. Log in at http://localhost:3300") print("3. Settings > API > Create Token named 'Harvey'") print("4. echo 'YOUR_TOKEN' > ~/.openclaw/secrets/outline-api-token.txt") sys.exit(1) if args.test: cmd_test(token, base_url) elif args.search: cmd_search(args.search, token, base_url, limit=args.limit) elif args.collection: cmd_collection(args.collection, token, base_url) elif args.doc: cmd_doc(args.doc, token, base_url) elif args.export_leah: cmd_export_leah(token, base_url, output_file=args.output) else: parser.print_help() if __name__ == "__main__": main()