fifth-domain/zero-point/global-search-api/server.py

#!/usr/bin/env python3
"""
Global Search API · 给通用 AI（豆包等）用的仓库检索门面
铸渊 ICE-GL-ZY001 · LL-169-20260707 · D167
"""
import http.server
import json
import subprocess
import os
import time
import hmac
import hashlib
from urllib.parse import urlparse, parse_qs

# ============== 配置 ==============
REPO = os.environ.get("REPO_PATH", "/tmp/worktree")
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.0.0"

# ============== git safe.directory ==============
# 让 root 也能 read 由 git 用户创建的 worktree
import tempfile, pathlib
_global_gitconfig = pathlib.Path("/tmp/gitconfig-global-search")
_global_gitconfig.write_text(f"[safe]\n\tdirectory = {REPO}\n")
os.environ["GIT_CONFIG_GLOBAL"] = str(_global_gitconfig)

# ============== 鉴权 ==============
def check_auth(headers):
    """Bearer token 验证 · 静态 token + 动态 verifier 双模式"""
    auth = headers.get("Authorization", "").replace("Bearer ", "").strip()
    if not auth:
        return False, "missing"
    # 静态 token
    if TOKEN and auth == TOKEN:
        return True, "static_token"
    # 动态 verifier（小湖灯同款）
    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 search_files(keyword, top=20):
    """git grep 全仓库文件内容搜索"""
    if not keyword:
        return []
    # 找含关键词的文件
    result = subprocess.run(
        ["git", "grep", "-l", "-i", "--no-color", keyword],
        cwd=REPO, capture_output=True, text=True, timeout=30
    )
    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, capture_output=True, text=True, timeout=10
            )
            matches = []
            for line in line_result.stdout.splitlines()[:5]:  # 每文件最多 5 行
                # 格式: filename:linenum:content
                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(path="", depth=3, ext_filter=None):
    """列目录树"""
    cmd = ["git", "ls-tree", "-r", "--name-only", "HEAD"]
    if path:
        # 加上 path 前缀过滤
        cmd.append(f"{path}/")
    result = subprocess.run(
        cmd, cwd=REPO, 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(path):
    """读单个文件"""
    if not path or ".." in path:
        return {"ok": False, "error": "invalid path"}
    try:
        result = subprocess.run(
            ["git", "show", f"HEAD:{path}"],
            cwd=REPO, 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]  # 限 50KB
        }
    except Exception as e:
        return {"ok": False, "error": str(e)}


def get_broadcast():
    """拉最新 BROADCAST"""
    bcast_paths = [
        os.path.join(REPO, "broadcasts"),
        os.path.join(REPO, "BROADCAST.md"),
        os.path.join(REPO, "GLW-BROADCAST.hdlp"),
    ]
    for path in bcast_paths:
        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.basename(path),
                        "content": f.read()[:20000]
                    }
    return {
        "ok": True,
        "broadcast": None,
        "note": "no broadcast asset found in repo"
    }


def get_system_status():
    """拉 SYSTEM-STATUS"""
    candidates = [
        os.path.join(REPO, "SYSTEM-STATUS.md"),
        os.path.join(REPO, "SYSTEM-STATUS.hdlp"),
        os.path.join(REPO, "eternal-lake-heart", "heartbeat-core", "SYSTEM-STATUS.hdlp"),
    ]
    for path in candidates:
        if os.path.exists(path):
            with open(path) as f:
                return {"ok": True, "path": os.path.relpath(path, REPO), "content": f.read()}
    return {
        "ok": True,
        "status": "no SYSTEM-STATUS file in repo · 铸渊创建后会更新",
        "note": "请铸渊在仓库根目录创建 SYSTEM-STATUS.md 或在 eternal-lake-heart/heartbeat-core/SYSTEM-STATUS.hdlp"
    }


def archive_hldp(content, type_="si", path=None):
    """HLDP 回执归档（写入工作树 · 待铸渊 commit）"""
    # 这里只是写文件,不直接 push
    # 铸渊下次醒来 commit
    archive_dir = os.path.join(REPO, "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)}


# ============== 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):
        parsed = urlparse(self.path)
        path = parsed.path
        query = parse_qs(parsed.query)

        # === 免鉴权端点(给豆包等通用 AI 低门槛接入) ===
        if path == "/healthz":
            return self.send_json({
                "ok": True,
                "service": "global-search-api",
                "version": VERSION,
                "repo": "bingshuo/fifth-domain",
                "ts": time.time(),
                "note": "服务正在运行·鉴权不是必须的·探活不需要 token"
            })
        if path == "/status":
            # 综合状态(不鉴权)·给豆包一眼看仓库健康
            try:
                head = subprocess.run(
                    ["git", "rev-parse", "HEAD"],
                    cwd=REPO, capture_output=True, text=True, timeout=5
                ).stdout.strip()
                log = subprocess.run(
                    ["git", "log", "--oneline", "-3"],
                    cwd=REPO, capture_output=True, text=True, timeout=5
                ).stdout.strip()
                file_count = subprocess.run(
                    ["git", "ls-tree", "-r", "--name-only", "HEAD"],
                    cwd=REPO, capture_output=True, text=True, timeout=10
                ).stdout.strip().count("\n") + 1
            except Exception as e:
                head, log, file_count = "?", str(e), 0
            return self.send_json({
                "ok": True,
                "service_alive": True,
                "repo": "bingshuo/fifth-domain",
                "head": head,
                "recent_commits": log.splitlines(),
                "file_count": file_count,
                "ts": time.time(),
                "note": "综合状态·免鉴权·给豆包一眼看仓库"
            })
        if path == "/broadcast":
            # BROADCAST · read-only 公开(豆包会调, 不鉴权)
            result = get_broadcast()
            result["_hint"] = "read-only 公开端点·不需要 token·如果 content 是 null 说明仓库还没有 BROADCAST 文件, 不是链路错"
            return self.send_json(result)
        if path == "/system-status":
            # SYSTEM-STATUS · read-only 公开(豆包会调, 不鉴权)
            result = get_system_status()
            result["_hint"] = "read-only 公开端点·不需要 token·content 是仓库里 SYSTEM-STATUS 文件的内容"
            return self.send_json(result)
        if path == "/help":
            return self.send_json({
                "ok": True,
                "endpoints": {
                    "GET /healthz": "健康检查(免鉴权)",
                    "GET /status": "综合状态(免鉴权·仓库 HEAD + commits + 文件数)",
                    "GET /system-status": "拉 SYSTEM-STATUS(免鉴权·读通知)",
                    "GET /broadcast": "拉最新 BROADCAST(免鉴权·读通知)",
                    "GET /search?q={keyword}&top={n}": "全仓库文件内容搜索(需鉴权)",
                    "GET /tree?path={prefix}&depth={n}&ext={hdlp,md}": "列目录树(需鉴权)",
                    "GET /file?path={path}": "读单文件(限 50KB, 需鉴权)",
                    "POST /archive": "HLDP 回执归档(JSON body: {type, content, path?}, 需鉴权)",
                },
                "auth_required": ["/search", "/tree", "/file", "/archive"],
                "auth_free": ["/healthz", "/status", "/system-status", "/broadcast", "/help"],
                "auth": "Authorization: Bearer <TOKEN> · 静态 token 或小湖灯 verifier",
                "repo": "bingshuo/fifth-domain",
                "base_url": "https://guanghubingshuo.com/global-search",
                "static_token": "de0648a8db3ae8675b8933e1808cec19",
            })

        # === 需要鉴权的端点(写操作 + 业务查询) ===
        ok, reason = check_auth(self.headers)
        if not ok:
            return self.send_json({
                "ok": False,
                "error": "unauthorized",
                "reason": reason,
                "_hint": "鉴权失败常见原因: (1) token 放 query 错位, 应放 Authorization header; (2) token 写错. 试试 /healthz /status 端点不需鉴权"
            }, 401)

        try:
            if path == "/search":
                kw = query.get("q", [""])[0]
                top = int(query.get("top", ["20"])[0])
                self.send_json({"ok": True, "keyword": kw, "count": -1, "results": search_files(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, "files": list_tree(path_arg, depth, ext_filter)})
            elif path == "/file":
                path_arg = query.get("path", [""])[0]
                self.send_json(read_file(path_arg))
            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):
        ok, reason = check_auth(self.headers)
        if not ok:
            return self.send_json({"ok": False, "error": "unauthorized", "reason": reason}, 401)

        parsed = urlparse(self.path)
        if parsed.path == "/archive":
            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", "")
                path = data.get("path")
                self.send_json(archive_hldp(content, type_, path))
            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 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"REPO = {REPO}")
    print(f"TOKEN = {'set' if TOKEN else 'unset (dev mode)'}")
    print(f"HMAC_SECRET = {'set' if HMAC_SECRET else 'unset (static token only)'}")
    server.serve_forever()


if __name__ == "__main__":
    main()