276 lines
12 KiB
Python
276 lines
12 KiB
Python
#!/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()
|