garc-gws-agent-runtime/scripts/garc-kg-query.py

#!/usr/bin/env python3
"""
GARC KG Query — Safe query/show interface for the knowledge graph cache file.
Arguments are passed via argv (no shell interpolation).
"""

import argparse
import json
import sys
from pathlib import Path


def load_kg(cache_path: str) -> dict:
    p = Path(cache_path)
    if not p.exists():
        print(f"❌ Knowledge graph cache not found: {cache_path}", file=sys.stderr)
        print("   Run: garc kg build", file=sys.stderr)
        sys.exit(1)
    try:
        with open(p) as f:
            return json.load(f)
    except Exception as e:
        print(f"❌ Failed to load KG cache: {e}", file=sys.stderr)
        sys.exit(1)


def kg_query(cache_path: str, query: str, max_results: int = 10):
    """Search nodes by keyword in title or content_preview."""
    kg = load_kg(cache_path)
    nodes = kg.get("nodes", [])
    query_lower = query.lower()

    matches = []
    for node in nodes:
        title = node.get("title", "").lower()
        content = node.get("content_preview", "").lower()
        if query_lower in title or query_lower in content:
            matches.append(node)

    built_at = kg.get("built_at", "?")[:10]
    print(f"Knowledge graph — built: {built_at}  nodes: {len(nodes)}")
    print()

    if not matches:
        print(f"No results for: \"{query}\"")
        return

    limit = min(max_results, len(matches))
    print(f"Results for \"{query}\" ({len(matches)} match{'es' if len(matches) != 1 else ''}, showing {limit}):")
    print()
    for m in matches[:limit]:
        doc_id = m.get("doc_id", "")
        title = m.get("title", "")
        link = m.get("web_link", "")
        links_count = len(m.get("links", []))
        preview = m.get("content_preview", "")[:120].replace("\n", " ")
        print(f"  [{doc_id[:16]}] {title}")
        if preview:
            print(f"    {preview}...")
        if links_count:
            print(f"    ↳ {links_count} linked doc(s)")
        if link:
            print(f"    🔗 {link}")
        print()


def kg_show(cache_path: str, doc_id: str):
    """Show full details for a specific doc."""
    kg = load_kg(cache_path)
    nodes = kg.get("nodes", [])

    for node in nodes:
        if node.get("doc_id") == doc_id:
            print(f"Title    : {node.get('title', '')}")
            print(f"Doc ID   : {doc_id}")
            print(f"MIME     : {node.get('mime_type', '')}")
            print(f"Modified : {node.get('modified_time', '')[:19]}")
            link = node.get("web_link", "")
            if link:
                print(f"URL      : {link}")
            print()
            preview = node.get("content_preview", "")
            if preview:
                print("Content preview:")
                print(preview[:800])
                print()
            links = node.get("links", [])
            if links:
                print(f"Linked documents ({len(links)}):")
                for lnk in links:
                    print(f"  → {lnk}")
            return

    print(f"Document '{doc_id}' not found in knowledge graph.")
    print(f"Run 'garc kg build' to refresh the index.")


def main():
    parser = argparse.ArgumentParser(description="GARC KG Query")
    sub = parser.add_subparsers(dest="command")

    qp = sub.add_parser("query", help="Search the knowledge graph")
    qp.add_argument("--cache", required=True, help="Path to knowledge-graph.json")
    qp.add_argument("--query", required=True, help="Search keyword")
    qp.add_argument("--max", type=int, default=10, help="Max results")

    sp = sub.add_parser("show", help="Show a specific doc")
    sp.add_argument("--cache", required=True, help="Path to knowledge-graph.json")
    sp.add_argument("--doc-id", required=True, help="Document ID")

    args = parser.parse_args()

    if args.command == "query":
        kg_query(args.cache, args.query, args.max)
    elif args.command == "show":
        kg_show(args.cache, args.doc_id)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()