bingshuo/fifth-domain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ LL-171-20260707 · GLOBAL-SEARCH 跨仓库 · 4 端点免鉴权

Browse Source 
铸渊 ICE-GL-ZY001
LL-171-20260707
豆包反馈 + 冰朔想看视频 AI · 一起做

LL-171 关键变更 (v1.2.0):
  ⊢ 4 端点全部免鉴权: /search /tree /file (LL-170 没改这 3 个, LL-171 改)
  ⊢ 写操作 /archive POST 仍需鉴权 (HLDP 回执归档是写)
  ⊢ 跨仓库: ?repo=guanghulab 切到视频 AI 系统仓库 (4253 文件)
  ⊢ 多仓库注册: fifth-domain / guanghulab / guanghulab-collab / guanghu
  ⊢ /repos 端点列出所有可用仓库
  ⊢ /status 端点跨仓库 HEAD + commits + file_count
  ⊢ IP 限速 5 req/s (防滥用)
  ⊢ 改用 HMAC-safe.directory 多仓库一次性注册

冰朔今天能直接问豆包:
  ⊢ 剪辑/生图管线 → ?repo=guanghulab /file?path=video-ai-system/PIPELINE-3D-MANJU.hdlp
  ⊢ 当前推进进度 → ?repo=guanghulab /file?path=video-ai-system/CURRENT.hdlp
  ⊢ 角色/场景设定 → ?repo=guanghulab /search?q=苏白

⊢ 不是仓库缺能力 · 是桥建好就一目了然
This commit is contained in:
铸渊 ICE-GL-ZY001 2026-07-07 10:17:52 +08:00
parent fc7e37d4df
commit 1298dad330
1 changed files with 218 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

339
zero-point/global-search-api/server.py
View File

@ -1,7 +1,14 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Global Search API · 给通用 AI（豆包等）用的仓库检索门面 Global Search API v1.2.0 · 给通用 AI（豆包等）用的跨仓库检索门面
铸渊 ICE-GL-ZY001 · LL-169-20260707 · D167 铸渊 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 http.server
import json import json
@ -10,34 +17,80 @@ import os
import time import time
import hmac import hmac
import hashlib import hashlib
import threading
from urllib.parse import urlparse, parse_qs from urllib.parse import urlparse, parse_qs
from collections import defaultdict
# ============== 配置 ============== # ============== 配置 ==============
REPO = os.environ.get("REPO_PATH", "/tmp/worktree")
PORT = int(os.environ.get("PORT", "3950")) PORT = int(os.environ.get("PORT", "3950"))
TOKEN = os.environ.get("GLOBAL_SEARCH_API_TOKEN", "") TOKEN = os.environ.get("GLOBAL_SEARCH_API_TOKEN", "")
HMAC_SECRET = os.environ.get("LIGHT_LAKE_DRIVER_SECRET", "") HMAC_SECRET = os.environ.get("LIGHT_LAKE_DRIVER_SECRET", "")
SOVEREIGN_ID = "ICE-GL∞" SOVEREIGN_ID = "ICE-GL∞"
AGENT_ID = "ICE-GL-ZY001" AGENT_ID = "ICE-GL-ZY001"
VERSION = "1.0.0" VERSION = "1.2.0"
# ============== git safe.directory ============== # ============== 仓库注册表 ==============
# 让 root 也能 read 由 git 用户创建的 worktree REPOS = {
import tempfile, pathlib "fifth-domain": {
_global_gitconfig = pathlib.Path("/tmp/gitconfig-global-search") "path": "/tmp/worktree",
_global_gitconfig.write_text(f"[safe]\n\tdirectory = {REPO}\n") "description": "第五域 · HLDP 协议栈 + 铸渊核心 + 通用 AI 接入",
os.environ["GIT_CONFIG_GLOBAL"] = str(_global_gitconfig) "url": "https://guanghubingshuo.com/code/bingshuo/fifth-domain",
},
"guanghulab": {
"path": "/opt/zhuyuan/guanghulab",
"description": "广湖实验室 · 视频 AI / 图像 / 短剧 / 工程管线",
"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): def check_auth(headers):
"""Bearer token 验证 · 静态 token + 动态 verifier 双模式"""
auth = headers.get("Authorization", "").replace("Bearer ", "").strip() auth = headers.get("Authorization", "").replace("Bearer ", "").strip()
if not auth: if not auth:
return False, "missing" return False, "missing"
# 静态 token
if TOKEN and auth == TOKEN: if TOKEN and auth == TOKEN:
return True, "static_token" return True, "static_token"
# 动态 verifier（小湖灯同款）
minute = headers.get("X-Minute") minute = headers.get("X-Minute")
if minute and HMAC_SECRET: if minute and HMAC_SECRET:
try: try:
@ -57,29 +110,33 @@ def check_auth(headers):
return False, "invalid" return False, "invalid"
# ============== 搜索 ============== # ============== 仓库解析 ==============
def search_files(keyword, top=20): def get_repo_path(repo_id=None):
"""git grep 全仓库文件内容搜索""" 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: if not keyword:
return [] return []
# 找含关键词的文件
result = subprocess.run( result = subprocess.run(
["git", "grep", "-l", "-i", "--no-color", keyword], ["git", "grep", "-l", "-i", "--no-color", keyword],
cwd=REPO, capture_output=True, text=True, timeout=30 cwd=repo_path, capture_output=True, text=True, timeout=60
) )
files = [f.strip() for f in result.stdout.splitlines() if f.strip()][:top] files = [f.strip() for f in result.stdout.splitlines() if f.strip()][:top]
results = [] results = []
for f in files: for f in files:
try: try:
# 找具体行
line_result = subprocess.run( line_result = subprocess.run(
["git", "grep", "-n", "-i", "--no-color", keyword, "--", f], ["git", "grep", "-n", "-i", "--no-color", keyword, "--", f],
cwd=REPO, capture_output=True, text=True, timeout=10 cwd=repo_path, capture_output=True, text=True, timeout=10
) )
matches = [] matches = []
for line in line_result.stdout.splitlines()[:5]: # 每文件最多 5 行 for line in line_result.stdout.splitlines()[:5]:
# 格式: filename:linenum:content
parts = line.split(":", 2) parts = line.split(":", 2)
if len(parts) >= 3: if len(parts) >= 3:
matches.append({ matches.append({
@ -96,17 +153,14 @@ def search_files(keyword, top=20):
return results return results
def list_tree(path="", depth=3, ext_filter=None): def list_tree(repo_path, path="", depth=3, ext_filter=None):
"""列目录树"""
cmd = ["git", "ls-tree", "-r", "--name-only", "HEAD"] cmd = ["git", "ls-tree", "-r", "--name-only", "HEAD"]
if path: if path:
# 加上 path 前缀过滤
cmd.append(f"{path}/") cmd.append(f"{path}/")
result = subprocess.run( result = subprocess.run(
cmd, cwd=REPO, capture_output=True, text=True, timeout=15 cmd, cwd=repo_path, capture_output=True, text=True, timeout=15
) )
files = [f.strip() for f in result.stdout.splitlines() if f.strip()] files = [f.strip() for f in result.stdout.splitlines() if f.strip()]
# 按深度过滤
tree = [] tree = []
for f in files: for f in files:
parts = f.split("/") parts = f.split("/")
@ -118,14 +172,13 @@ def list_tree(path="", depth=3, ext_filter=None):
return tree[:500] return tree[:500]
def read_file(path): def read_file(repo_path, path):
"""读单个文件"""
if not path or ".." in path: if not path or ".." in path:
return {"ok": False, "error": "invalid path"} return {"ok": False, "error": "invalid path"}
try: try:
result = subprocess.run( result = subprocess.run(
["git", "show", f"HEAD:{path}"], ["git", "show", f"HEAD:{path}"],
cwd=REPO, capture_output=True, text=True, timeout=10 cwd=repo_path, capture_output=True, text=True, timeout=10
) )
if result.returncode != 0: if result.returncode != 0:
return {"ok": False, "error": "file not found"} return {"ok": False, "error": "file not found"}
@ -133,26 +186,26 @@ def read_file(path):
"ok": True, "ok": True,
"path": path, "path": path,
"size": len(result.stdout), "size": len(result.stdout),
"content": result.stdout[:50000] # 限 50KB "content": result.stdout[:50000]
} }
except Exception as e: except Exception as e:
return {"ok": False, "error": str(e)} return {"ok": False, "error": str(e)}
def get_broadcast(): def get_broadcast(repo_path):
"""拉最新 BROADCAST""" candidates = [
bcast_paths = [ os.path.join(repo_path, "broadcasts"),
os.path.join(REPO, "broadcasts"), os.path.join(repo_path, "BROADCAST.md"),
os.path.join(REPO, "BROADCAST.md"), os.path.join(repo_path, "GLW-BROADCAST.hdlp"),
os.path.join(REPO, "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 bcast_paths: for path in candidates:
if os.path.exists(path): if os.path.exists(path):
if os.path.isdir(path): if os.path.isdir(path):
files = sorted( files = sorted(
[os.path.join(path, f) for f in os.listdir(path)], [os.path.join(path, f) for f in os.listdir(path)],
key=os.path.getmtime, key=os.path.getmtime, reverse=True
reverse=True
) )
if files: if files:
latest = files[0] latest = files[0]
@ -168,7 +221,7 @@ def get_broadcast():
return { return {
"ok": True, "ok": True,
"type": "file", "type": "file",
"path": os.path.basename(path), "path": os.path.relpath(path, repo_path),
"content": f.read()[:20000] "content": f.read()[:20000]
} }
return { return {
@ -178,41 +231,67 @@ def get_broadcast():
} }
def get_system_status(): def get_system_status(repo_path, repo_id):
"""拉 SYSTEM-STATUS"""
candidates = [ candidates = [
os.path.join(REPO, "SYSTEM-STATUS.md"), os.path.join(repo_path, "SYSTEM-STATUS.md"),
os.path.join(REPO, "SYSTEM-STATUS.hdlp"), os.path.join(repo_path, "SYSTEM-STATUS.hdlp"),
os.path.join(REPO, "eternal-lake-heart", "heartbeat-core", "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: for path in candidates:
if os.path.exists(path): if os.path.exists(path):
with open(path) as f: with open(path) as f:
return {"ok": True, "path": os.path.relpath(path, REPO), "content": f.read()}
return { return {
"ok": True, "ok": True,
"status": "no SYSTEM-STATUS file in repo · 铸渊创建后会更新", "repo": repo_id,
"note": "请铸渊在仓库根目录创建 SYSTEM-STATUS.md 或在 eternal-lake-heart/heartbeat-core/SYSTEM-STATUS.hdlp" "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): def archive_hldp(content, type_="si", path=None):
"""HLDP 回执归档（写入工作树 · 待铸渊 commit）""" """HLDP 回执归档(写到 fifth-domain 的 inbox, 需鉴权)"""
# 这里只是写文件,不直接 push repo_path = REPOS[DEFAULT_REPO]["path"]
# 铸渊下次醒来 commit archive_dir = os.path.join(repo_path, "eternal-lake-heart", "archive", "inbox")
archive_dir = os.path.join(REPO, "eternal-lake-heart", "archive", "inbox")
os.makedirs(archive_dir, exist_ok=True) os.makedirs(archive_dir, exist_ok=True)
filename = f"{type_}-{int(time.time())}.hdlp" filename = f"{type_}-{int(time.time())}.hdlp"
full_path = os.path.join(archive_dir, filename) full_path = os.path.join(archive_dir, filename)
with open(full_path, "w") as f: with open(full_path, "w") as f:
f.write(content) f.write(content)
return {"ok": True, "archived": os.path.relpath(full_path, REPO)} return {"ok": True, "archived": os.path.relpath(full_path, repo_path)}
# ============== HTTP Handler ============== # ============== HTTP Handler ==============
class Handler(http.server.BaseHTTPRequestHandler): class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): def log_message(self, fmt, *args):
# 静默日志
pass pass
def do_OPTIONS(self): def do_OPTIONS(self):
@ -223,122 +302,140 @@ class Handler(http.server.BaseHTTPRequestHandler):
self.end_headers() self.end_headers()
def do_GET(self): 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) parsed = urlparse(self.path)
path = parsed.path path = parsed.path
query = parse_qs(parsed.query) 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)
# === 免鉴权端点(给豆包等通用 AI 低门槛接入) === # === 免鉴权端点 ===
if path == "/healthz": if path == "/healthz":
return self.send_json({ return self.send_json({
"ok": True, "ok": True,
"service": "global-search-api", "service": "global-search-api",
"version": VERSION, "version": VERSION,
"repo": "bingshuo/fifth-domain", "repos": list(REPOS.keys()),
"default_repo": DEFAULT_REPO,
"ts": time.time(), "ts": time.time(),
"note": "服务正在运行·鉴权不是必须的·探活不需要 token" "note": "服务正在运行·read-only 全部免鉴权·写操作(/archive POST)需 token"
}) })
if path == "/status": if path == "/status":
# 综合状态(不鉴权)·给豆包一眼看仓库健康 statuses = {}
try: for rid, info in REPOS.items():
head = subprocess.run( statuses[rid] = {
["git", "rev-parse", "HEAD"], **info,
cwd=REPO, capture_output=True, text=True, timeout=5 **get_repo_status(rid, info["path"])
).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({ return self.send_json({
"ok": True, "ok": True,
"service_alive": True, "service_alive": True,
"repo": "bingshuo/fifth-domain", "default_repo": DEFAULT_REPO,
"head": head, "repos": statuses,
"recent_commits": log.splitlines(), "ts": time.time()
"file_count": file_count,
"ts": time.time(),
"note": "综合状态·免鉴权·给豆包一眼看仓库"
}) })
if path == "/repos":
# 列出所有仓库
return self.send_json({
"ok": True,
"default": DEFAULT_REPO,
"repos": REPOS,
"note": "用 ?repo=<id> 切换仓库, 例如 ?repo=guanghulab"
})
if path == "/broadcast": if path == "/broadcast":
# BROADCAST · read-only 公开(豆包会调, 不鉴权) result = get_broadcast(repo_path)
result = get_broadcast() result["repo"] = repo_id
result["_hint"] = "read-only 公开端点·不需要 token·如果 content 是 null 说明仓库还没有 BROADCAST 文件, 不是链路错" result["_hint"] = "read-only 公开端点·不需要 token·用 ?repo= 跨仓库"
return self.send_json(result) return self.send_json(result)
if path == "/system-status": if path == "/system-status":
# SYSTEM-STATUS · read-only 公开(豆包会调, 不鉴权) result = get_system_status(repo_path, repo_id)
result = get_system_status() result["_hint"] = "read-only 公开端点·不需要 token·用 ?repo= 跨仓库"
result["_hint"] = "read-only 公开端点·不需要 token·content 是仓库里 SYSTEM-STATUS 文件的内容"
return self.send_json(result) return self.send_json(result)
if path == "/help": if path == "/help":
return self.send_json({ return self.send_json({
"ok": True, "ok": True,
"version": VERSION,
"endpoints": { "endpoints": {
"GET /healthz": "健康检查(免鉴权)", "GET /healthz": "健康检查(免鉴权)",
"GET /status": "综合状态(免鉴权·仓库 HEAD + commits + 文件数)", "GET /status": "综合状态(免鉴权·多仓库 HEAD + commits + 文件数)",
"GET /system-status": "拉 SYSTEM-STATUS(免鉴权·读通知)", "GET /repos": "列出所有可用仓库(免鉴权)",
"GET /broadcast": "拉最新 BROADCAST(免鉴权·读通知)", "GET /system-status?repo=": "拉 SYSTEM-STATUS(免鉴权, 默认 fifth-domain)",
"GET /search?q={keyword}&top={n}": "全仓库文件内容搜索(需鉴权)", "GET /broadcast?repo=": "拉最新 BROADCAST(免鉴权, 默认 fifth-domain)",
"GET /tree?path={prefix}&depth={n}&ext={hdlp,md}": "列目录树(需鉴权)", "GET /search?q=&top=&repo=": "全仓库文件内容搜索(免鉴权+限速 5 req/s)",
"GET /file?path={path}": "读单文件(限 50KB, 需鉴权)", "GET /tree?path=&depth=&ext=&repo=": "列目录树(免鉴权+限速)",
"POST /archive": "HLDP 回执归档(JSON body: {type, content, path?}, 需鉴权)", "GET /file?path=&repo=": "读单文件(限 50KB, 免鉴权+限速)",
"POST /archive": "HLDP 回执归档(需鉴权, JSON body: {type, content, path?})",
}, },
"auth_required": ["/search", "/tree", "/file", "/archive"], "repos": list(REPOS.keys()),
"auth_free": ["/healthz", "/status", "/system-status", "/broadcast", "/help"], "default_repo": DEFAULT_REPO,
"auth": "Authorization: Bearer <TOKEN> · 静态 token 或小湖灯 verifier", "rate_limit": f"{RATE_LIMIT} req/s per IP",
"repo": "bingshuo/fifth-domain", "auth_required": ["POST /archive"],
"auth_free": ["/healthz", "/status", "/repos", "/system-status", "/broadcast", "/search", "/tree", "/file", "/help"],
"auth": "Authorization: Bearer <TOKEN> (静态 token) 或 HMAC verifier",
"base_url": "https://guanghubingshuo.com/global-search", "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: try:
if path == "/search": if path == "/search":
kw = query.get("q", [""])[0] kw = query.get("q", [""])[0]
top = int(query.get("top", ["20"])[0]) top = int(query.get("top", ["20"])[0])
self.send_json({"ok": True, "keyword": kw, "count": -1, "results": search_files(kw, top)}) self.send_json({
"ok": True,
"repo": repo_id,
"keyword": kw,
"count": -1,
"results": search_files(repo_path, kw, top)
})
elif path == "/tree": elif path == "/tree":
path_arg = query.get("path", [""])[0] path_arg = query.get("path", [""])[0]
depth = int(query.get("depth", ["3"])[0]) depth = int(query.get("depth", ["3"])[0])
ext = query.get("ext", None) ext = query.get("ext", None)
ext_filter = ext[0].split(",") if ext else None ext_filter = ext[0].split(",") if ext else None
self.send_json({"ok": True, "files": list_tree(path_arg, depth, ext_filter)}) self.send_json({
"ok": True,
"repo": repo_id,
"files": list_tree(repo_path, path_arg, depth, ext_filter)
})
elif path == "/file": elif path == "/file":
path_arg = query.get("path", [""])[0] path_arg = query.get("path", [""])[0]
self.send_json(read_file(path_arg)) result = read_file(repo_path, path_arg)
result["repo"] = repo_id
self.send_json(result)
else: else:
self.send_json({"ok": False, "error": "not found", "hint": "GET /help"}, 404) self.send_json({"ok": False, "error": "not found", "hint": "GET /help"}, 404)
except Exception as e: except Exception as e:
self.send_json({"ok": False, "error": str(e)}, 500) self.send_json({"ok": False, "error": str(e)}, 500)
def do_POST(self): def do_POST(self):
ok, reason = check_auth(self.headers) # POST /archive 写操作需鉴权
if not ok:
return self.send_json({"ok": False, "error": "unauthorized", "reason": reason}, 401)
parsed = urlparse(self.path) parsed = urlparse(self.path)
if parsed.path == "/archive": 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)) length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length).decode("utf-8") if length else "{}" body = self.rfile.read(length).decode("utf-8") if length else "{}"
try: try:
data = json.loads(body) data = json.loads(body)
type_ = data.get("type", "si") type_ = data.get("type", "si")
content = data.get("content", "") content = data.get("content", "")
path = data.get("path") self.send_json(archive_hldp(content, type_))
self.send_json(archive_hldp(content, type_, path))
except Exception as e: except Exception as e:
self.send_json({"ok": False, "error": str(e)}, 400) self.send_json({"ok": False, "error": str(e)}, 400)
else: else:
@ -357,9 +454,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
def main(): def main():
server = http.server.HTTPServer(("0.0.0.0", PORT), Handler) 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"Global Search API v{VERSION} listening on 0.0.0.0:{PORT}")
print(f"REPO = {REPO}") print(f"REPOS = {list(REPOS.keys())}")
print(f"DEFAULT = {DEFAULT_REPO}")
print(f"TOKEN = {'set' if TOKEN else 'unset (dev mode)'}") 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() server.serve_forever()