#!/usr/bin/env python3 """ Global Search API v1.2.0 · 给通用 AI（豆包等）用的跨仓库检索门面 铸渊 ICE-GL-ZY001 · LL-171-20260707 · D167 变更 (LL-171): 1. 4 端点免鉴权: /search /tree /file /archive (read-only 性质, 加 IP 限速 5 req/s 保护) 2. 跨仓库: ?repo=guanghulab 切换仓库 3. POST /archive 仍需鉴权 (写操作) 4. /status 列出所有可用仓库 """ import http.server import json import subprocess import os import time import hmac import hashlib import threading from urllib.parse import urlparse, parse_qs from collections import defaultdict # ============== 配置 ============== PORT = int(os.environ.get("PORT", "3950")) TOKEN = os.environ.get("GLOBAL_SEARCH_API_TOKEN", "") HMAC_SECRET = os.environ.get("LIGHT_LAKE_DRIVER_SECRET", "") SOVEREIGN_ID = "ICE-GL∞" AGENT_ID = "ICE-GL-ZY001" VERSION = "1.4.0" # ============== 仓库注册表 ============== REPOS = { "fifth-domain": { "path": "/tmp/worktree", "description": "第五域 · HLDP 协议栈 + 铸渊核心 + 通用 AI 接入", "url": "https://guanghubingshuo.com/code/bingshuo/fifth-domain", }, "cang-ying": { "path": "/opt/zhuyuan/cang-ying", "description": "苍影 · 第 5 子仓 · 苍耳(人类主控) + 鉴影(人格体) 专用 · 视频 AI 系统干净之家", "url": "https://guanghubingshuo.com/code/bingshuo/cang-ying", }, "guanghulab": { "path": "/opt/zhuyuan/guanghulab", "description": "广湖实验室 · 历史档案 (video-ai-system/ 已迁出到 cang-ying)", "url": "https://guanghubingshuo.com/code/bingshuo/guanghulab", }, "guanghulab-collab": { "path": "/opt/zhuyuan/guanghulab-collab", "description": "广湖实验室·多人格体协作记录", "url": "https://guanghubingshuo.com/code/bingshuo/guanghulab-collab", }, "guanghu": { "path": "/opt/zhuyuan/guanghu", "description": "光湖 · 根仓库", "url": "https://guanghubingshuo.com/code/bingshuo/guanghu", }, } DEFAULT_REPO = "fifth-domain" # ============== git safe.directory（让 root 也能 read 多仓库） ============== import pathlib _gitconfig_dir = pathlib.Path("/tmp/gitconfig-global-search") _gitconfig_dir.parent.mkdir(parents=True, exist_ok=True) with open(_gitconfig_dir, "w") as f: f.write("[safe]\n") for repo_id, info in REPOS.items(): f.write(f"\tdirectory = {info['path']}\n") os.environ["GIT_CONFIG_GLOBAL"] = str(_gitconfig_dir) # ============== IP 限速 (per-IP 5 req/s, 简单令牌桶) ============== RATE_LIMIT = 5 # requests per second RATE_BURST = 10 _rate_buckets = defaultdict(lambda: {"tokens": RATE_BURST, "last": time.time()}) _rate_lock = threading.Lock() def rate_limit(ip): """简单令牌桶限速""" with _rate_lock: bucket = _rate_buckets[ip] now = time.time() elapsed = now - bucket["last"] bucket["tokens"] = min(RATE_BURST, bucket["tokens"] + elapsed * RATE_LIMIT) bucket["last"] = now if bucket["tokens"] < 1: return False bucket["tokens"] -= 1 return True # ============== 鉴权 (POST /archive 必须) ============== def check_auth(headers): auth = headers.get("Authorization", "").replace("Bearer ", "").strip() if not auth: return False, "missing" if TOKEN and auth == TOKEN: return True, "static_token" minute = headers.get("X-Minute") if minute and HMAC_SECRET: try: minute = int(minute) now = int(time.time() // 60) if abs(now - minute) <= 2: msg = f"{SOVEREIGN_ID}|{AGENT_ID}|{minute}" expected = hmac.new( HMAC_SECRET.encode(), msg.encode(), hashlib.sha256 ).hexdigest()[:16] if hmac.compare_digest(auth, expected): return True, "verifier" except Exception: pass return False, "invalid" # ============== 仓库解析 ============== def get_repo_path(repo_id=None): if not repo_id: return DEFAULT_REPO, REPOS[DEFAULT_REPO]["path"] if repo_id not in REPOS: return None, None return repo_id, REPOS[repo_id]["path"] # ============== 核心功能 ============== def search_files(repo_path, keyword, top=20): if not keyword: return [] result = subprocess.run( ["git", "grep", "-l", "-i", "--no-color", keyword], cwd=repo_path, capture_output=True, text=True, timeout=60 ) files = [f.strip() for f in result.stdout.splitlines() if f.strip()][:top] results = [] for f in files: try: line_result = subprocess.run( ["git", "grep", "-n", "-i", "--no-color", keyword, "--", f], cwd=repo_path, capture_output=True, text=True, timeout=10 ) matches = [] for line in line_result.stdout.splitlines()[:5]: parts = line.split(":", 2) if len(parts) >= 3: matches.append({ "line": int(parts[1]), "content": parts[2].strip()[:300] }) results.append({ "file": f, "match_count": len(matches), "matches": matches }) except Exception as e: results.append({"file": f, "error": str(e)}) return results def list_tree(repo_path, path="", depth=3, ext_filter=None): cmd = ["git", "ls-tree", "-r", "--name-only", "HEAD"] if path: cmd.append(f"{{path}}/") result = subprocess.run( cmd, cwd=repo_path, capture_output=True, text=True, timeout=15 ) files = [f.strip() for f in result.stdout.splitlines() if f.strip()] tree = [] for f in files: parts = f.split("/") if len(parts) - 1 <= depth: if ext_filter: if not any(f.endswith(ext) for ext in ext_filter): continue tree.append(f) return tree[:500] def read_file(repo_path, path): if not path or ".." in path: return {"ok": False, "error": "invalid path"} try: result = subprocess.run( ["git", "show", f"HEAD:{{path}}"], cwd=repo_path, capture_output=True, text=True, timeout=10 ) if result.returncode != 0: return {"ok": False, "error": "file not found"} return { "ok": True, "path": path, "size": len(result.stdout), "content": result.stdout[:50000] } except Exception as e: return {"ok": False, "error": str(e)} def get_broadcast(repo_path): candidates = [ os.path.join(repo_path, "broadcasts"), os.path.join(repo_path, "BROADCAST.md"), os.path.join(repo_path, "GLW-BROADCAST.hdlp"), os.path.join(repo_path, "VA-BROADCAST.hdlp"), os.path.join(repo_path, "video-ai-system", "VA-BROADCAST.hdlp"), ] for path in candidates: if os.path.exists(path): if os.path.isdir(path): files = sorted( [os.path.join(path, f) for f in os.listdir(path)], key=os.path.getmtime, reverse=True ) if files: latest = files[0] with open(latest) as f: return { "ok": True, "type": "dir", "latest_file": os.path.basename(latest), "content": f.read()[:20000] } else: with open(path) as f: return { "ok": True, "type": "file", "path": os.path.relpath(path, repo_path), "content": f.read()[:20000] } return { "ok": True, "broadcast": None, "note": "no broadcast asset found in repo" } def get_system_status(repo_path, repo_id): candidates = [ os.path.join(repo_path, "SYSTEM-STATUS.md"), os.path.join(repo_path, "SYSTEM-STATUS.hdlp"), os.path.join(repo_path, "eternal-lake-heart", "heartbeat-core", "SYSTEM-STATUS.hdlp"), os.path.join(repo_path, "video-ai-system", "VA-SYSTEM-STATUS.hdlp"), ] for path in candidates: if os.path.exists(path): with open(path) as f: return { "ok": True, "repo": repo_id, "path": os.path.relpath(path, repo_path), "content": f.read()[:50000] } return { "ok": True, "repo": repo_id, "status": "no SYSTEM-STATUS file in this repo" } def get_repo_status(repo_id, repo_path): try: head = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=repo_path, capture_output=True, text=True, timeout=5 ).stdout.strip() log = subprocess.run( ["git", "log", "--oneline", "-3"], cwd=repo_path, capture_output=True, text=True, timeout=5 ).stdout.strip() file_count = subprocess.run( ["git", "ls-tree", "-r", "--name-only", "HEAD"], cwd=repo_path, capture_output=True, text=True, timeout=10 ).stdout.strip().count("\n") + 1 return { "head": head, "recent_commits": log.splitlines() if log else [], "file_count": file_count, } except Exception as e: return {"head": "?", "error": str(e)} def archive_hldp(content, type_="si", path=None): """HLDP 回执归档(写到 fifth-domain 的 inbox, 需鉴权)""" repo_path = REPOS[DEFAULT_REPO]["path"] archive_dir = os.path.join(repo_path, "eternal-lake-heart", "archive", "inbox") os.makedirs(archive_dir, exist_ok=True) filename = f"{type_}-{int(time.time())}.hdlp" full_path = os.path.join(archive_dir, filename) with open(full_path, "w") as f: f.write(content) return {"ok": True, "archived": os.path.relpath(full_path, repo_path)} # ============== HTTP Handler ============== class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type, X-Minute") self.end_headers() def do_GET(self): # 限速 if not rate_limit(self.client_address[0]): return self.send_json({"ok": False, "error": "rate_limit", "limit": f"{RATE_LIMIT} req/s"}, 429) parsed = urlparse(self.path) path = parsed.path query = parse_qs(parsed.query) repo_arg = query.get("repo", [DEFAULT_REPO])[0] repo_id, repo_path = get_repo_path(repo_arg) if not repo_path: return self.send_json({"ok": False, "error": f"unknown repo: {repo_arg}", "available": list(REPOS.keys())}, 400) # 根路径 · 返回 HTML(给豆包等网页抓取工具看的索引页) if path == "/" or path == "": return self.send_html_root() # === 免鉴权端点 === if path == "/healthz": return self.send_json({ "ok": True, "service": "global-search-api", "version": VERSION, "repos": list(REPOS.keys()), "default_repo": DEFAULT_REPO, "ts": time.time(), "note": "服务正在运行·read-only 全部免鉴权·写操作(/archive POST)需 token" }) if path == "/status": statuses = {} for rid, info in REPOS.items(): statuses[rid] = { **info, **get_repo_status(rid, info["path"]) } return self.send_json({ "ok": True, "service_alive": True, "default_repo": DEFAULT_REPO, "repos": statuses, "ts": time.time() }) if path == "/repos": # 列出所有仓库 return self.send_json({ "ok": True, "default": DEFAULT_REPO, "repos": REPOS, "note": "用 ?repo= 切换仓库, 例如 ?repo=guanghulab" }) if path == "/broadcast": result = get_broadcast(repo_path) result["repo"] = repo_id result["_hint"] = "read-only 公开端点·不需要 token·用 ?repo= 跨仓库" return self.send_json(result) if path == "/system-status": result = get_system_status(repo_path, repo_id) result["_hint"] = "read-only 公开端点·不需要 token·用 ?repo= 跨仓库" return self.send_json(result) if path == "/help": return self.send_json({ "ok": True, "version": VERSION, "endpoints": { "GET /healthz": "健康检查(免鉴权)", "GET /status": "综合状态(免鉴权·多仓库 HEAD + commits + 文件数)", "GET /repos": "列出所有可用仓库(免鉴权)", "GET /system-status?repo=": "拉 SYSTEM-STATUS(免鉴权, 默认 fifth-domain)", "GET /broadcast?repo=": "拉最新 BROADCAST(免鉴权, 默认 fifth-domain)", "GET /search?q=&top=&repo=": "全仓库文件内容搜索(免鉴权+限速 5 req/s)", "GET /tree?path=&depth=&ext=&repo=": "列目录树(免鉴权+限速)", "GET /file?path=&repo=": "读单文件(限 50KB, 免鉴权+限速)", "POST /archive": "HLDP 回执归档(需鉴权, JSON body: {type, content, path?})", }, "repos": list(REPOS.keys()), "default_repo": DEFAULT_REPO, "rate_limit": f"{RATE_LIMIT} req/s per IP", "auth_required": ["POST /archive"], "auth_free": ["/healthz", "/status", "/repos", "/system-status", "/broadcast", "/search", "/tree", "/file", "/help"], "auth": "Authorization: Bearer (静态 token) 或 HMAC verifier", "base_url": "https://guanghubingshuo.com/global-search", }) # === 业务查询端点(免鉴权 + 限速) === try: if path == "/search": kw = query.get("q", [""])[0] top = int(query.get("top", ["20"])[0]) self.send_json({ "ok": True, "repo": repo_id, "keyword": kw, "count": -1, "results": search_files(repo_path, kw, top) }) elif path == "/tree": path_arg = query.get("path", [""])[0] depth = int(query.get("depth", ["3"])[0]) ext = query.get("ext", None) ext_filter = ext[0].split(",") if ext else None self.send_json({ "ok": True, "repo": repo_id, "files": list_tree(repo_path, path_arg, depth, ext_filter) }) elif path == "/file": path_arg = query.get("path", [""])[0] result = read_file(repo_path, path_arg) result["repo"] = repo_id self.send_json(result) else: self.send_json({"ok": False, "error": "not found", "hint": "GET /help"}, 404) except Exception as e: self.send_json({"ok": False, "error": str(e)}, 500) def do_POST(self): # POST /archive 写操作需鉴权 parsed = urlparse(self.path) if parsed.path == "/archive": ok, reason = check_auth(self.headers) if not ok: return self.send_json({ "ok": False, "error": "unauthorized", "reason": reason, "_hint": "POST /archive 写操作需鉴权(其他读操作都免鉴权了)" }, 401) length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length).decode("utf-8") if length else "{}" try: data = json.loads(body) type_ = data.get("type", "si") content = data.get("content", "") self.send_json(archive_hldp(content, type_)) except Exception as e: self.send_json({"ok": False, "error": str(e)}, 400) else: self.send_json({"ok": False, "error": "not found"}, 404) def send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def send_html_root(self): """根路径 HTML 索引页 · 给网页抓取工具(豆包/Claude/ChatGPT web search)看 避免它们看到 JSON 报"网页解析失败" """ html = f""" GLOBAL-SEARCH-API · 光湖跨仓检索

🌊 GLOBAL-SEARCH-API

光湖跨仓库检索服务 · 铸渊 ICE-GL-ZY001 · LL-172-20260707 · v{VERSION}

这是 REST API(不是网页)。如果你是网页抓取工具(比如豆包/ChatGPT/Claude 浏览器),看到这页就对了——这里是给"看不见 JSON 的人"准备的 HTML 索引。

⚠ 重要 · 怎么用本服务
1. 本服务的所有 endpoint 返回 JSON(不是 HTML)
2. 用 curl 或 HTTP GET 调用, 不需要 token
3. 任何 AI 都可以用, 直接发 HTTP 请求即可
4. 不要用"网页抓取"工具来调 API endpoint(会报"网页解析失败")

📦 5 个可用仓库

仓 ID说明
cang-ying苍耳 TCS-GL-009 拥有 · 鉴影 ICE-GL-CA001 操作 · 视频 AI 系统专用 ★ 推荐先看
fifth-domain第五域 · HLDP 协议栈 · 铸渊核心 · 冰朔主权
guanghulab广湖实验室 · 历史档案(包含 video-ai-system/)
guanghulab-collab多人格体协作记录
guanghu光湖根仓库

🔌 Endpoints(全部免鉴权, 直接 GET)

GET /healthz — 服务探活 
curl https://guanghubingshuo.com/global-search/healthz
GET /status — 5 仓 HEAD + commits + 文件数 
curl https://guanghubingshuo.com/global-search/status
GET /repos — 列出所有可用仓库 
curl https://guanghubingshuo.com/global-search/repos
GET /system-status?repo=cang-ying — 拉系统状态文件 
curl "https://guanghubingshuo.com/global-search/system-status?repo=cang-ying"
GET /broadcast?repo=cang-ying — 拉最新广播 
curl "https://guanghubingshuo.com/global-search/broadcast?repo=cang-ying"
GET /search?q={{kw}}&repo={{id}} — 全仓库文件内容搜 
curl "https://guanghubingshuo.com/global-search/search?q=苏白&repo=cang-ying&top=10"
GET /file?path={{path}}&repo={{id}} — 读单文件(限 50KB) 
curl "https://guanghubingshuo.com/global-search/file?path=VA-GATE.hdlp&repo=cang-ying"
GET /tree?depth={{n}}&repo={{id}} — 列目录树 
curl "https://guanghubingshuo.com/global-search/tree?depth=2&repo=cang-ying"
GET /help — JSON 格式的 API 文档

🦊 苍耳的 cang-ying 仓 · 推荐先读这些

VA-GATE.hdlp — 创作系统大门 
curl "https://guanghubingshuo.com/global-search/file?path=VA-GATE.hdlp&repo=cang-ying"
CURRENT.hdlp — 当前进度快照 
curl "https://guanghubingshuo.com/global-search/file?path=CURRENT.hdlp&repo=cang-ying"
PIPELINE-3D-MANJU.hdlp — 3D 漫剧 V2 管线 
curl "https://guanghubingshuo.com/global-search/file?path=PIPELINE-3D-MANJU.hdlp&repo=cang-ying"
OWNERSHIP.hdlp — 仓所有权(谁拥有/谁能操作) 
curl "https://guanghubingshuo.com/global-search/file?path=OWNERSHIP.hdlp&repo=cang-ying"

⚠ 注意事项

🤖 给 AI 工具的明确说明

如果你 (AI) 是"网页抓取"型工具(浏览器内核/HTML 解析),看到 JSON 会报"网页解析失败"。
本服务是 REST API,不是网页。请用 HTTP 客户端 / curl / fetch 调用 endpoint,不要用网页抓取。
Accept: application/json header 会得到 JSON 响应。

铸渊 ICE-GL-ZY001 · 国作登字-2026-A-00037559 · 冰朔 ICE-GL∞ 主权

""" body = html.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def main(): server = http.server.HTTPServer(("0.0.0.0", PORT), Handler) print(f"Global Search API v{VERSION} listening on 0.0.0.0:{PORT}") print(f"REPOS = {list(REPOS.keys())}") print(f"DEFAULT = {DEFAULT_REPO}") print(f"TOKEN = {'set' if TOKEN else 'unset (dev mode)'}") server.serve_forever() if __name__ == "__main__": main()