global-search-api/server_v2.py

#!/usr/bin/env python3
"""Global Search API V2 — public, read-only entry for general-purpose AI.

The Fifth Domain World Router is the source of truth for repository and
persona routes.  This service deliberately has no write, login, handshake, or
token endpoints: server changes belong to GLSV, not a public search API.
"""
import html
import json
import os
import subprocess
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from urllib.parse import parse_qs, urlparse

VERSION = "2.0.0"
PORT = int(os.environ.get("PORT", "3951"))
MAX_QUERY = 160
MAX_FILE_BYTES = 100_000
MAX_RESULTS = 50


def router_path():
    candidates = [
        os.environ.get("WORLD_ROUTER_PATH", ""),
        "/opt/zhuyuan/fifth-domain/world-router.json",
        str(Path(__file__).with_name("world-router.json")),
    ]
    for candidate in candidates:
        if candidate and Path(candidate).is_file():
            return Path(candidate)
    raise RuntimeError("world-router.json not found; set WORLD_ROUTER_PATH")


def load_router():
    source = router_path()
    with source.open(encoding="utf-8") as handle:
        data = json.load(handle)
    if data.get("schema") != "GLW-WORLD-ROUTER-1":
        raise RuntimeError("unsupported World Router schema")
    return data, source


def repo_path(slug):
    configured = os.environ.get("REPO_PATHS_JSON", "")
    if configured:
        try:
            value = json.loads(configured).get(slug)
            if value:
                return Path(value)
        except json.JSONDecodeError:
            pass
    return Path(os.environ.get("REPO_ROOT", "/opt/zhuyuan")) / slug


def git(repo, *args, timeout=8):
    try:
        result = subprocess.run(
            ["git", *args], cwd=repo, text=True, capture_output=True,
            timeout=timeout, check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as error:
        return None, str(error)
    if result.returncode not in (0, 1):
        return None, (result.stderr or result.stdout or "git command failed").strip()
    return result.stdout, None


def repo_snapshot(repository):
    path = repo_path(repository["slug"])
    item = {
        "id": repository["id"], "slug": repository["slug"],
        "url": repository["url"], "state": repository["state"],
        "summary": repository["summary"], "available": path.is_dir(),
    }
    if item["available"]:
        head, error = git(path, "rev-parse", "HEAD", timeout=3)
        item["commit"] = head.strip() if head else None
        if error:
            item["availability_note"] = error
    else:
        item["availability_note"] = "repository is registered but not mounted on this search server"
    return item


def safe_path(value):
    if not value or len(value) > 500:
        return None
    path = PurePosixPath(value)
    if path.is_absolute() or ".." in path.parts or str(path) == ".":
        return None
    return str(path)


def bounded_int(values, name, default, limit):
    try:
        return max(1, min(int(values.get(name, [default])[0]), limit))
    except (ValueError, TypeError):
        return default


def normalise(value):
    return value.casefold().strip()


def resolve(router, query):
    needle = normalise(query)
    repositories = router["repositories"]
    matches = []
    for repository in repositories:
        fields = [repository["id"], repository["slug"], repository["summary"], *repository.get("owners", [])]
        if not needle or any(needle in normalise(str(field)) for field in fields):
            matches.append({"kind": "repository", **repo_snapshot(repository)})
    for persona in router.get("personas", []):
        fields = [persona["id"], *persona.get("aliases", []), persona.get("system", "")]
        if needle and any(needle in normalise(str(field)) for field in fields):
            repository = next(item for item in repositories if item["id"] == persona["repository"])
            matches.append({
                "kind": "persona", "id": persona["id"], "aliases": persona.get("aliases", []),
                "system": persona.get("system"), "repository": repo_snapshot(repository),
                "path": persona.get("path", ""),
            })
    return matches


def search_repo(repository, query, top):
    path = repo_path(repository["slug"])
    base = repo_snapshot(repository)
    if not base["available"]:
        return {"repository": base, "results": []}
    output, error = git(path, "grep", "-n", "-i", "--no-color", query, timeout=10)
    if error:
        return {"repository": base, "results": [], "note": error}
    grouped = {}
    for line in (output or "").splitlines():
        filename, separator, rest = line.partition(":")
        number, separator2, content = rest.partition(":")
        if not separator or not separator2 or not number.isdigit():
            continue
        bucket = grouped.setdefault(filename, [])
        if len(bucket) < 3:
            bucket.append({"line": int(number), "content": content.strip()[:400]})
        if len(grouped) >= top and filename not in grouped:
            break
    return {"repository": base, "results": [
        {"path": filename, "matches": matches} for filename, matches in list(grouped.items())[:top]
    ]}


def tree_repo(repository, prefix, depth, limit):
    path = repo_path(repository["slug"])
    base = repo_snapshot(repository)
    if not base["available"]:
        return {"repository": base, "paths": []}
    output, error = git(path, "ls-tree", "-r", "--name-only", "HEAD", timeout=10)
    if error:
        return {"repository": base, "paths": [], "note": error}
    prefix = safe_path(prefix) if prefix else ""
    if prefix is None:
        return {"repository": base, "paths": [], "note": "invalid path"}
    paths = []
    prefix_parts = len(prefix.split("/")) if prefix else 0
    for item in (output or "").splitlines():
        if prefix and not item.startswith(prefix.rstrip("/") + "/") and item != prefix:
            continue
        if len(item.split("/")) - prefix_parts > depth:
            continue
        paths.append(item)
        if len(paths) >= limit:
            break
    return {"repository": base, "prefix": prefix, "paths": paths}


def file_repo(repository, requested):
    path = repo_path(repository["slug"])
    base = repo_snapshot(repository)
    requested = safe_path(requested)
    if not requested:
        return {"repository": base, "ok": False, "error": "invalid path"}
    if not base["available"]:
        return {"repository": base, "ok": False, "error": "repository unavailable"}
    output, error = git(path, "show", f"HEAD:{requested}", timeout=10)
    if error:
        return {"repository": base, "ok": False, "error": "file not found"}
    encoded = (output or "").encode("utf-8")
    return {
        "repository": base, "ok": True, "path": requested,
        "truncated": len(encoded) > MAX_FILE_BYTES,
        "content": encoded[:MAX_FILE_BYTES].decode("utf-8", errors="replace"),
    }


def markdown(data):
    return "# Guanghu Global Search API V2\n\n```json\n" + json.dumps(data, ensure_ascii=False, indent=2) + "\n```\n"


class Handler(BaseHTTPRequestHandler):
    server_version = "GuanghuGlobalSearch/2.0"

    def log_message(self, fmt, *args):
        print("[v2] " + fmt % args)

    def send_data(self, status, data, query):
        wants_markdown = query.get("format", [""])[0].lower() in {"md", "markdown"}
        body = markdown(data).encode("utf-8") if wants_markdown else json.dumps(data, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/markdown; charset=utf-8" if wants_markdown else "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.end_headers()

    def do_GET(self):
        parsed = urlparse(self.path)
        query = parse_qs(parsed.query)
        try:
            router, source = load_router()
        except (OSError, ValueError, RuntimeError) as error:
            return self.send_data(503, {"ok": False, "error": str(error)}, query)
        meta = {
            "ok": True, "service": "guanghu-global-search", "version": VERSION,
            "mode": "public-read-only", "router_source": str(source),
            "router_schema": router["schema"], "generated_at": int(time.time()),
        }
        route = parsed.path.rstrip("/") or "/"
        if route in {"/", "/help"}:
            return self.send_data(200, {**meta, "entry": "Use /v2/resolve?q=... before searching.", "endpoints": {
                "repos": "/v2/repos", "resolve": "/v2/resolve?q=铸渊",
                "context": "/v2/context?q=GLSV", "search": "/v2/search?q=HLDP&repo=fifth-domain",
                "tree": "/v2/tree?repo=fifth-domain&path=gls&depth=2",
                "file": "/v2/file?repo=fifth-domain&path=gls/GLS-ENTRY.hdlp",
            }, "write_policy": "No public write endpoint. Server operations use GLSV."}, query)
        if not route.startswith("/v2/"):
            return self.send_data(404, {**meta, "ok": False, "error": "use /v2/ or /v2/help"}, query)
        endpoint = route[len("/v2/"):]
        repositories = router["repositories"]
        by_slug = {item["slug"]: item for item in repositories}
        if endpoint == "repos":
            return self.send_data(200, {**meta, "repositories": [repo_snapshot(item) for item in repositories]}, query)
        if endpoint == "resolve":
            requested = query.get("q", [""])[0][:MAX_QUERY]
            return self.send_data(200, {**meta, "query": requested, "matches": resolve(router, requested)}, query)
        if endpoint == "context":
            requested = query.get("q", [""])[0][:MAX_QUERY]
            entry = next(item for item in repositories if item["id"] == router["entry_repository"])
            files = ["BROADCAST-TOWER.hdlp", "gls/GLS-ENTRY.hdlp"]
            if normalise(requested) in {"glsv", "gatekeeper", "安全验证"}:
                files.append("gls/GLSV-0001-SECURITY-VERIFICATION-ARCHITECTURE.hdlp")
            return self.send_data(200, {**meta, "query": requested, "entry_repository": repo_snapshot(entry), "routes": resolve(router, requested), "suggested_files": files}, query)
        repository = by_slug.get(query.get("repo", [router["repositories"][0]["slug"]])[0])
        if not repository:
            return self.send_data(404, {**meta, "ok": False, "error": "unknown repo; call /v2/repos"}, query)
        if endpoint == "search":
            requested = query.get("q", [""])[0][:MAX_QUERY]
            if not requested:
                return self.send_data(400, {**meta, "ok": False, "error": "q is required"}, query)
            return self.send_data(200, {**meta, "query": requested, **search_repo(repository, requested, bounded_int(query, "top", 20, MAX_RESULTS))}, query)
        if endpoint == "tree":
            return self.send_data(200, {**meta, **tree_repo(repository, query.get("path", [""])[0], bounded_int(query, "depth", 3, 8), bounded_int(query, "limit", 200, 500))}, query)
        if endpoint == "file":
            return self.send_data(200, {**meta, **file_repo(repository, query.get("path", [""])[0])}, query)
        return self.send_data(404, {**meta, "ok": False, "error": "unknown endpoint; call /v2/help"}, query)


if __name__ == "__main__":
    print(f"Global Search API V2 listening on {PORT}")
    ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()