skills/gitnexus-search/scripts/gitnexus-search.py

#!/usr/bin/env python3
"""Universal GitNexus search CLI for Cursor (fallback) and Hermes."""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request

DEFAULT_SERVER = "https://gitnexus.mobile.agency-ai.dev"

REPO_ALIASES: dict[str, str] = {
    "gateway": "one-gateway",
    "one-gateway": "one-gateway",
    "ai-lawyer": "ai-lawyer-srvr",
    "ai-lawyer-srvr": "ai-lawyer-srvr",
}

MCP_PATHS: dict[str, str] = {
    "one-gateway": "/data/gitnexus/repos/one-gateway",
    "ai-lawyer-srvr": "/data/gitnexus/repos/ai-lawyer-srvr",
}


def base_url() -> str:
    return os.environ.get("GITNEXUS_SERVER_URL", DEFAULT_SERVER).rstrip("/")


def normalize_repo(name: str) -> str:
    key = name.strip().lower()
    if key in REPO_ALIASES:
        return REPO_ALIASES[key]
    return name.strip()


def request(method: str, path: str, body: dict | None = None) -> object:
    url = f"{base_url()}{path}"
    headers = {"Accept": "application/json"}
    data = None
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            raw = resp.read().decode("utf-8")
            if not raw.strip():
                return {}
            return json.loads(raw)
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        print(f"HTTP {exc.code} {exc.reason}: {detail}", file=sys.stderr)
        raise SystemExit(1) from exc
    except urllib.error.URLError as exc:
        print(f"Request failed: {exc.reason}", file=sys.stderr)
        raise SystemExit(1) from exc


def list_repos() -> list[dict]:
    repos = request("GET", "/api/repos")
    if not isinstance(repos, list):
        raise SystemExit(f"Unexpected /api/repos response: {repos!r}")
    return repos


def resolve_repo(name: str) -> dict:
    canonical = normalize_repo(name)
    matches = [r for r in list_repos() if r.get("name") == canonical]
    if not matches:
        available = sorted({r.get("name", "") for r in list_repos() if r.get("name")})
        print(f"Unknown repo: {name!r} (normalized: {canonical!r})", file=sys.stderr)
        print(f"Available: {', '.join(available) or '(none)'}", file=sys.stderr)
        raise SystemExit(1)

    with_files = [r for r in matches if (r.get("stats") or {}).get("files", 0) > 0]
    if with_files:
        return with_files[0]
    if len(matches) == 1:
        print(
            f"WARN: repo {canonical!r} is indexed but has 0 files — re-index may be needed",
            file=sys.stderr,
        )
        return matches[0]
    print(f"Ambiguous repo {canonical!r}: multiple entries, none with files", file=sys.stderr)
    for r in matches:
        print(f"  - path={r.get('path')}", file=sys.stderr)
    raise SystemExit(1)


def search(repo_name: str, query: str, limit: int) -> dict:
    repo_entry = resolve_repo(repo_name)
    canonical = repo_entry.get("name", normalize_repo(repo_name))
    payload = {"query": query, "repo": canonical, "limit": limit}
    result = request("POST", "/api/search", payload)
    if not isinstance(result, dict):
        raise SystemExit(f"Unexpected /api/search response: {result!r}")
    result["_repo"] = canonical
    result["_repo_path"] = repo_entry.get("path", "")
    result["_mcp_path"] = MCP_PATHS.get(canonical, repo_entry.get("path", ""))
    return result


def format_text(result: dict) -> str:
    lines = [
        f"GitNexus search",
        f"Server: {base_url()}",
        f"Repo: {result.get('_repo')} ({result.get('_repo_path')})",
        f"MCP path: {result.get('_mcp_path')}",
        "",
    ]
    results = result.get("results") or []
    if not results:
        lines.append("No results.")
        return "\n".join(lines)

    lines.append(f"Top {len(results)} hits:")
    lines.append("")
    for item in results:
        rank = item.get("rank", "?")
        path = item.get("filePath", "?")
        score = item.get("score", 0)
        nodes = item.get("nodeIds") or []
        sym = ", ".join(n.split(":")[-1] for n in nodes[:3]) if nodes else "-"
        lines.append(f"  {rank}. {path}  (score={score:.2f})")
        if sym != "-":
            lines.append(f"      symbols: {sym}")
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Search GitNexus remote index by repo name and query",
    )
    parser.add_argument("repo", help="Repository name or alias (e.g. one-gateway, ai-lawyer)")
    parser.add_argument("query", nargs="+", help="Search prompt (words after repo)")
    parser.add_argument("--limit", type=int, default=10, help="Max results (default: 10)")
    parser.add_argument("--json", action="store_true", help="Output raw JSON")
    parser.add_argument(
        "--server",
        default=None,
        help="Override GITNEXUS_SERVER_URL",
    )
    args = parser.parse_args()

    if args.server:
        os.environ["GITNEXUS_SERVER_URL"] = args.server.rstrip("/")

    query = " ".join(args.query).strip()
    if not query:
        print("Query cannot be empty", file=sys.stderr)
        raise SystemExit(1)

    result = search(args.repo, query, args.limit)
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print(format_text(result))


if __name__ == "__main__":
    main()