feat(skills): 补充 gitlink-kb 配套脚本与测试
依审阅意见，将 SKILL.md 方式A引用的脚本（scripts/）与单元测试（tests/）一并纳入，使 PR 自包含可运行。
This commit is contained in:
parent
340cf543f4
commit
2fdb56d7a6
|
|
@ -0,0 +1,241 @@
|
|||
"""GitLink 公开 API 共享客户端。
|
||||
|
||||
供 gitlink-skills-pack 下各 Skill 的脚本复用。仅依赖 Python 标准库，
|
||||
无需第三方包，便于在受限环境或 Agent 沙箱中运行。
|
||||
|
||||
数据全部来自 GitLink 平台公开接口（https://www.gitlink.org.cn/api），
|
||||
默认无需 token；如需访问私有仓库，可传入 token。
|
||||
|
||||
所有方法均为只读，不修改任何远程数据。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
API_BASE = "https://www.gitlink.org.cn/api"
|
||||
USER_AGENT = "gitlink-skills-pack/1.0 (+https://www.gitlink.org.cn)"
|
||||
DEFAULT_TIMEOUT = 30
|
||||
COMMIT_PAGE_SIZE = 50 # GitLink commits 接口每页硬上限
|
||||
|
||||
|
||||
class GitLinkError(RuntimeError):
|
||||
"""API 调用中不可恢复的错误。"""
|
||||
|
||||
|
||||
class GitLinkClient:
|
||||
"""GitLink 公开数据接口客户端。
|
||||
|
||||
带可选文件缓存：同一资源重复读取不重复打网，对平台友好。
|
||||
"""
|
||||
|
||||
def __init__(self, base: str = API_BASE, token: str | None = None,
|
||||
timeout: int = DEFAULT_TIMEOUT, cache_dir: Path | None = None) -> None:
|
||||
self.base = base.rstrip("/")
|
||||
self.token = token
|
||||
self.timeout = timeout
|
||||
self.cache_dir = cache_dir
|
||||
if self.cache_dir:
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 底层请求
|
||||
# ------------------------------------------------------------------
|
||||
def _cache_path(self, url: str) -> Path | None:
|
||||
if not self.cache_dir:
|
||||
return None
|
||||
safe = urllib.parse.quote(url, safe="")
|
||||
return self.cache_dir / f"{safe}.json"
|
||||
|
||||
def get(self, path: str, query: dict[str, Any] | None = None) -> Any:
|
||||
"""GET 请求，返回解析后的 JSON（dict/list）或 None。"""
|
||||
url = f"{self.base}/{path.lstrip('/')}"
|
||||
if query:
|
||||
url = f"{url}?{urllib.parse.urlencode(query)}"
|
||||
|
||||
cache_path = self._cache_path(url)
|
||||
if cache_path and cache_path.exists():
|
||||
return json.loads(cache_path.read_text(encoding="utf-8"))
|
||||
|
||||
headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
|
||||
if self.token:
|
||||
headers["Authorization"] = f"Bearer {self.token}"
|
||||
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||||
raw = resp.read().decode("utf-8", errors="replace")
|
||||
except urllib.error.HTTPError as exc:
|
||||
raise GitLinkError(f"HTTP {exc.code}: {url}") from exc
|
||||
except urllib.error.URLError as exc:
|
||||
raise GitLinkError(f"网络错误: {url} -> {exc.reason}") from exc
|
||||
|
||||
text = raw.strip()
|
||||
if not text or text in ("null", "{}", "[]"):
|
||||
data: Any = None
|
||||
elif text[0] in "{[":
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise GitLinkError(f"响应非 JSON: {url}") from exc
|
||||
else:
|
||||
raise GitLinkError(f"响应非 JSON（可能是 HTML）: {url}")
|
||||
|
||||
if cache_path is not None:
|
||||
cache_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||||
return data
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 资源访问（高层封装）
|
||||
# ------------------------------------------------------------------
|
||||
def repo_info(self, owner: str, repo: str) -> dict[str, Any]:
|
||||
"""仓库元信息。"""
|
||||
data = self.get(f"{owner}/{repo}.json")
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def issues(self, owner: str, repo: str, limit: int = 50,
|
||||
page: int = 1) -> list[dict[str, Any]]:
|
||||
"""Issue 列表。"""
|
||||
data = self.get(f"{owner}/{repo}/issues.json", {"page": page, "limit": limit})
|
||||
return _extract_list(data, ("issues",))
|
||||
|
||||
def issue_detail(self, owner: str, repo: str, number: int) -> dict[str, Any]:
|
||||
"""单个 Issue 详情（含完整字段）。"""
|
||||
data = self.get(f"{owner}/{repo}/issues/{number}.json")
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def pulls(self, owner: str, repo: str, limit: int = 50,
|
||||
page: int = 1) -> list[dict[str, Any]]:
|
||||
"""PR 列表。"""
|
||||
data = self.get(f"{owner}/{repo}/pulls.json", {"page": page, "limit": limit})
|
||||
return _extract_list(data, ("issues", "pulls"))
|
||||
|
||||
def contributors(self, owner: str, repo: str) -> list[dict[str, Any]]:
|
||||
"""贡献者列表。"""
|
||||
data = self.get(f"{owner}/{repo}/contributors.json")
|
||||
return _extract_list(data, ("list",))
|
||||
|
||||
def commits(self, owner: str, repo: str, max_pages: int = 4) -> list[dict[str, Any]]:
|
||||
"""提交列表（按需翻页，每页 50 条，以 total_count 为终止依据）。"""
|
||||
out: list[dict[str, Any]] = []
|
||||
total: int | None = None
|
||||
for page in range(1, max(1, max_pages) + 1):
|
||||
data = self.get(f"{owner}/{repo}/commits.json",
|
||||
{"page": page, "limit": COMMIT_PAGE_SIZE})
|
||||
if total is None and isinstance(data, dict):
|
||||
total = _safe_int(data.get("total_count")) or None
|
||||
page_items = _extract_list(data, ("commits",))
|
||||
if not page_items:
|
||||
break
|
||||
out.extend(page_items)
|
||||
if total is not None and len(out) >= total:
|
||||
break
|
||||
return out
|
||||
|
||||
def list_dir(self, owner: str, repo: str, path: str = "",
|
||||
ref: str = "master") -> list[dict[str, Any]]:
|
||||
"""列出目录下的条目（文件与子目录）。
|
||||
|
||||
返回的每个 entry 含 name / path / type(file|dir) / sha / size，
|
||||
文件类型的 entry 还可能直接带明文 content。
|
||||
"""
|
||||
data = self.get(f"{owner}/{repo}/sub_entries.json",
|
||||
{"filepath": path, "ref": ref})
|
||||
# 查询目录时 entries 为 list；查询单文件时 entries 为单个 dict。
|
||||
# 统一归一化为 list，便于下游处理。
|
||||
if isinstance(data, dict):
|
||||
entries = data.get("entries")
|
||||
if isinstance(entries, dict):
|
||||
return [entries]
|
||||
if isinstance(entries, list):
|
||||
return entries
|
||||
return _extract_list(data, ("entries",))
|
||||
|
||||
def file_content(self, owner: str, repo: str, filepath: str,
|
||||
ref: str = "master") -> str | None:
|
||||
"""读取单个文件的文本内容。
|
||||
|
||||
GitLink 的 sub_entries 接口对单文件查询会在 entries 中返回明文 content，
|
||||
据此取出。文件不存在或无内容时返回 None。
|
||||
"""
|
||||
entries = self.list_dir(owner, repo, filepath, ref)
|
||||
target = filepath.rsplit("/", 1)[-1]
|
||||
for entry in entries:
|
||||
if entry.get("type") == "file" and entry.get("name") == target:
|
||||
content = entry.get("content")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
# 回退：部分情况下单文件查询 entries 仅一项
|
||||
if len(entries) == 1 and entries[0].get("type") == "file":
|
||||
content = entries[0].get("content")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
return None
|
||||
|
||||
def readme(self, owner: str, repo: str, ref: str = "master") -> str | None:
|
||||
"""读取仓库 README（自动 base64 解码）。"""
|
||||
data = self.get(f"{owner}/{repo}/readme.json", {"ref": ref})
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
content = data.get("content")
|
||||
if not isinstance(content, str):
|
||||
return None
|
||||
# 注意：GitLink 的 readme.json 虽然 encoding 标为 base64，
|
||||
# 实测 content 多为明文 Markdown。先探测明文特征，命中则直接返回；
|
||||
# 否则再尝试 base64 解码。
|
||||
stripped = content.lstrip()
|
||||
if stripped.startswith(("#", "<", "[", "-", "*", "本", "这", "项")) or "\n" in content[:200]:
|
||||
return content
|
||||
try:
|
||||
raw = base64.b64decode(content.encode("ascii", "ignore"))
|
||||
decoded = raw.decode("utf-8", errors="replace")
|
||||
# 解码结果若不像文本（大量替换符），回退为原文
|
||||
if decoded.count("\ufffd") > len(decoded) * 0.1:
|
||||
return content
|
||||
return decoded
|
||||
except (ValueError, TypeError):
|
||||
return content
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# 辅助
|
||||
# ----------------------------------------------------------------------------
|
||||
|
||||
def _extract_list(payload: Any, keys: tuple[str, ...]) -> list[Any]:
|
||||
"""从可能嵌套的响应中提取第一个匹配键的列表。"""
|
||||
if isinstance(payload, list):
|
||||
return payload
|
||||
if isinstance(payload, dict):
|
||||
for key in keys:
|
||||
value = payload.get(key)
|
||||
if isinstance(value, list):
|
||||
return value
|
||||
return []
|
||||
|
||||
|
||||
def _safe_int(value: Any, default: int = 0) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def split_owner_repo(slug: str) -> tuple[str, str]:
|
||||
"""把 'owner/repo' 或完整 URL 解析为 (owner, repo)。"""
|
||||
s = slug.strip()
|
||||
if s.startswith("http"):
|
||||
parts = urllib.parse.urlparse(s).path.strip("/").split("/")
|
||||
if len(parts) >= 2:
|
||||
return parts[0], parts[1].replace(".git", "")
|
||||
raise GitLinkError(f"无法从 URL 解析 owner/repo: {slug}")
|
||||
if "/" in s:
|
||||
owner, repo = s.split("/", 1)
|
||||
return owner, repo.replace(".git", "")
|
||||
raise GitLinkError(f"格式应为 owner/repo: {slug}")
|
||||
|
|
@ -0,0 +1,293 @@
|
|||
"""gitlink-kb：仓库知识库问答助手。
|
||||
|
||||
把一个 GitLink 仓库的文档（README、docs/ 目录、各类 Markdown）索引起来，
|
||||
支持关键词检索、文档地图生成与 FAQ 提取，让仓库沉淀的知识可被快速查询。
|
||||
|
||||
数据来自 GitLink 公开 API（只读），无需登录。
|
||||
|
||||
用法：
|
||||
python kb.py --owner Gitlink --repo gitlink-cli --query "如何安装"
|
||||
python kb.py --owner Gitlink --repo gitlink-cli --map
|
||||
python kb.py --owner Gitlink --repo gitlink-cli --faq
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from glapi import GitLinkClient, GitLinkError, split_owner_repo
|
||||
|
||||
# Windows 控制台默认 GBK，直接打印含 emoji 的 Markdown 会抛 UnicodeEncodeError。
|
||||
# 重配置 stdout 为 UTF-8，确保跨平台正常输出。
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 文档类文件扩展名
|
||||
DOC_EXTS = (".md", ".markdown", ".rst", ".txt")
|
||||
# 优先索引的文档目录
|
||||
DOC_DIRS = ["", "docs", "doc", ".gitlink", "wiki"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 文档解析
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def split_sections(markdown: str) -> list[dict[str, Any]]:
|
||||
"""按 Markdown 标题切分为段落，每段含标题、层级、正文。"""
|
||||
sections: list[dict[str, Any]] = []
|
||||
current = {"title": "（开头）", "level": 0, "lines": []}
|
||||
for line in markdown.splitlines():
|
||||
m = re.match(r"^(#{1,6})\s+(.*)", line)
|
||||
if m:
|
||||
if current["lines"] or current["title"] != "（开头）":
|
||||
sections.append(current)
|
||||
current = {"title": m.group(2).strip(), "level": len(m.group(1)), "lines": []}
|
||||
else:
|
||||
current["lines"].append(line)
|
||||
if current["lines"] or current["title"] != "（开头）":
|
||||
sections.append(current)
|
||||
for s in sections:
|
||||
s["body"] = "\n".join(s["lines"]).strip()
|
||||
del s["lines"]
|
||||
return sections
|
||||
|
||||
|
||||
def collect_docs(owner: str, repo: str, ref: str,
|
||||
client: GitLinkClient, max_files: int = 20) -> list[dict[str, Any]]:
|
||||
"""收集仓库中的文档文件及其内容。"""
|
||||
docs: list[dict[str, Any]] = []
|
||||
|
||||
# README 优先
|
||||
readme = client.readme(owner, repo, ref)
|
||||
if readme:
|
||||
docs.append({"path": "README", "content": readme})
|
||||
|
||||
seen = {"readme", "readme.md"}
|
||||
for d in DOC_DIRS:
|
||||
if len(docs) >= max_files:
|
||||
break
|
||||
try:
|
||||
entries = client.list_dir(owner, repo, d, ref)
|
||||
except GitLinkError:
|
||||
continue
|
||||
for e in entries:
|
||||
if len(docs) >= max_files:
|
||||
break
|
||||
name = str(e.get("name", ""))
|
||||
if e.get("type") != "file" or not name.lower().endswith(DOC_EXTS):
|
||||
continue
|
||||
path = f"{d}/{name}" if d else name
|
||||
if path.lower() in seen:
|
||||
continue
|
||||
seen.add(path.lower())
|
||||
# entries 里可能已带 content，否则单独取
|
||||
content = e.get("content")
|
||||
if not content:
|
||||
content = client.file_content(owner, repo, path, ref)
|
||||
if content:
|
||||
docs.append({"path": path, "content": content})
|
||||
return docs
|
||||
|
||||
|
||||
def build_index(docs: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""把文档切分为可检索的段落索引。"""
|
||||
index: list[dict[str, Any]] = []
|
||||
for doc in docs:
|
||||
for sec in split_sections(doc["content"]):
|
||||
if not sec["body"] and sec["title"] == "（开头）":
|
||||
continue
|
||||
index.append({
|
||||
"doc": doc["path"],
|
||||
"title": sec["title"],
|
||||
"level": sec["level"],
|
||||
"body": sec["body"],
|
||||
})
|
||||
return index
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 检索
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _tokenize_query(query: str) -> list[str]:
|
||||
"""把查询拆为关键词（英文按词，中文按字/词粗切）。"""
|
||||
tokens = re.findall(r"[A-Za-z0-9_]+", query.lower())
|
||||
# 中文：粗略按 2-gram 补充
|
||||
zh = re.findall(r"[\u4e00-\u9fff]+", query)
|
||||
for seg in zh:
|
||||
tokens.append(seg)
|
||||
for i in range(len(seg) - 1):
|
||||
tokens.append(seg[i:i + 2])
|
||||
return [t for t in tokens if t]
|
||||
|
||||
|
||||
def search(index: list[dict[str, Any]], query: str, top: int = 5) -> list[dict[str, Any]]:
|
||||
"""在索引中检索与查询最相关的段落（基于关键词命中计分）。"""
|
||||
tokens = _tokenize_query(query)
|
||||
if not tokens:
|
||||
return []
|
||||
scored: list[tuple[int, dict[str, Any]]] = []
|
||||
for sec in index:
|
||||
haystack = (sec["title"] + "\n" + sec["body"]).lower()
|
||||
score = 0
|
||||
for t in tokens:
|
||||
score += haystack.count(t.lower())
|
||||
# 标题命中加权
|
||||
title_low = sec["title"].lower()
|
||||
for t in tokens:
|
||||
if t.lower() in title_low:
|
||||
score += 5
|
||||
if score > 0:
|
||||
scored.append((score, sec))
|
||||
scored.sort(key=lambda x: x[0], reverse=True)
|
||||
results = []
|
||||
for score, sec in scored[:top]:
|
||||
snippet = sec["body"][:300].strip()
|
||||
results.append({
|
||||
"doc": sec["doc"], "title": sec["title"],
|
||||
"score": score, "snippet": snippet,
|
||||
})
|
||||
return results
|
||||
|
||||
|
||||
def extract_faq(index: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||||
"""从文档中提取 FAQ（标题形似问题，或 Q/问 开头的段落）。"""
|
||||
faq: list[dict[str, str]] = []
|
||||
for sec in index:
|
||||
title = sec["title"]
|
||||
is_question = (
|
||||
"?" in title or "？" in title
|
||||
or title.lower().startswith(("q:", "q ", "how", "what", "why", "when", "如何", "怎么", "为什么", "是否"))
|
||||
)
|
||||
if is_question and sec["body"]:
|
||||
faq.append({"question": title, "answer": sec["body"][:400].strip(), "doc": sec["doc"]})
|
||||
return faq
|
||||
|
||||
|
||||
def build_map(index: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
||||
"""按文档归类标题，生成文档地图。"""
|
||||
doc_map: dict[str, list[dict[str, Any]]] = {}
|
||||
for sec in index:
|
||||
if sec["title"] == "（开头）":
|
||||
continue
|
||||
doc_map.setdefault(sec["doc"], []).append(
|
||||
{"title": sec["title"], "level": sec["level"]})
|
||||
return doc_map
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 渲染
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def render_search(owner: str, repo: str, query: str,
|
||||
results: list[dict[str, Any]]) -> str:
|
||||
lines = [f"# 知识库检索 — {owner}/{repo}", "", f"查询：**{query}**", ""]
|
||||
if not results:
|
||||
lines += ["未找到相关内容。建议换个关键词，或确认仓库是否有相关文档。", ""]
|
||||
return "\n".join(lines)
|
||||
for i, r in enumerate(results, 1):
|
||||
lines += [
|
||||
f"## {i}. {r['title']} `{r['doc']}`（相关度 {r['score']}）",
|
||||
"",
|
||||
r["snippet"] + ("…" if len(r["snippet"]) >= 300 else ""),
|
||||
"",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_map(owner: str, repo: str, doc_map: dict[str, list[dict[str, Any]]]) -> str:
|
||||
lines = [f"# 文档地图 — {owner}/{repo}", "",
|
||||
f"共索引 {len(doc_map)} 个文档。", ""]
|
||||
for doc, secs in doc_map.items():
|
||||
lines.append(f"## 📄 {doc}")
|
||||
lines.append("")
|
||||
for s in secs:
|
||||
indent = " " * max(0, s["level"] - 1)
|
||||
lines.append(f"{indent}- {s['title']}")
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_faq(owner: str, repo: str, faq: list[dict[str, str]]) -> str:
|
||||
lines = [f"# 常见问题（FAQ）— {owner}/{repo}", ""]
|
||||
if not faq:
|
||||
lines += ["未从文档中识别出 FAQ 条目。", ""]
|
||||
return "\n".join(lines)
|
||||
for item in faq:
|
||||
lines += [f"### ❓ {item['question']}", "", item["answer"], "",
|
||||
f"<sub>来源：{item['doc']}</sub>", ""]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
p = argparse.ArgumentParser(prog="gitlink-kb", description="仓库知识库问答助手")
|
||||
p.add_argument("--owner", help="仓库所有者")
|
||||
p.add_argument("--repo", help="仓库名称")
|
||||
p.add_argument("--slug", help="owner/repo 或完整 URL")
|
||||
p.add_argument("--ref", default="master", help="分支或标签，默认 master")
|
||||
p.add_argument("--query", help="检索关键词/问题")
|
||||
p.add_argument("--map", action="store_true", help="生成文档地图")
|
||||
p.add_argument("--faq", action="store_true", help="提取 FAQ")
|
||||
p.add_argument("--max-files", type=int, default=20, help="最多索引的文档数")
|
||||
p.add_argument("--format", choices=["markdown", "json"], default="markdown")
|
||||
p.add_argument("--output", type=Path, help="输出文件")
|
||||
args = p.parse_args(argv)
|
||||
|
||||
if args.slug:
|
||||
owner, repo = split_owner_repo(args.slug)
|
||||
elif args.owner and args.repo:
|
||||
owner, repo = args.owner, args.repo
|
||||
else:
|
||||
print("错误：请用 --owner/--repo 或 --slug 指定仓库。", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
client = GitLinkClient()
|
||||
try:
|
||||
docs = collect_docs(owner, repo, args.ref, client, max_files=args.max_files)
|
||||
index = build_index(docs)
|
||||
except GitLinkError as exc:
|
||||
print(f"采集失败：{exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if args.query:
|
||||
results = search(index, args.query)
|
||||
out = (json.dumps({"query": args.query, "results": results}, ensure_ascii=False, indent=2)
|
||||
if args.format == "json" else render_search(owner, repo, args.query, results))
|
||||
elif args.map:
|
||||
doc_map = build_map(index)
|
||||
out = (json.dumps(doc_map, ensure_ascii=False, indent=2)
|
||||
if args.format == "json" else render_map(owner, repo, doc_map))
|
||||
elif args.faq:
|
||||
faq = extract_faq(index)
|
||||
out = (json.dumps({"faq": faq}, ensure_ascii=False, indent=2)
|
||||
if args.format == "json" else render_faq(owner, repo, faq))
|
||||
else:
|
||||
# 默认输出索引概况
|
||||
summary = {"owner": owner, "repo": repo,
|
||||
"indexed_docs": len({s["doc"] for s in index}),
|
||||
"sections": len(index)}
|
||||
out = (json.dumps(summary, ensure_ascii=False, indent=2)
|
||||
if args.format == "json"
|
||||
else f"# 知识库索引 — {owner}/{repo}\n\n已索引 {summary['indexed_docs']} 个文档、"
|
||||
f"{summary['sections']} 个段落。\n\n用 `--query <问题>` 检索、`--map` 看文档地图、`--faq` 提取常见问题。")
|
||||
|
||||
if args.output:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(out, encoding="utf-8")
|
||||
print(f"已写入 {args.output}")
|
||||
else:
|
||||
print(out)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,130 @@
|
|||
"""gitlink-kb 单元测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||||
|
||||
import pytest
|
||||
|
||||
from kb import (
|
||||
split_sections, build_index, search, extract_faq, build_map,
|
||||
collect_docs, _tokenize_query,
|
||||
)
|
||||
|
||||
DOC = """# 项目标题
|
||||
|
||||
简介段落。
|
||||
|
||||
## 安装
|
||||
|
||||
使用 pip 安装这个工具。
|
||||
|
||||
## 如何配置？
|
||||
|
||||
先创建配置文件，然后运行 init。
|
||||
|
||||
### 子配置
|
||||
|
||||
细节内容。
|
||||
"""
|
||||
|
||||
|
||||
class TestSplitSections:
|
||||
def test_splits_by_heading(self):
|
||||
secs = split_sections(DOC)
|
||||
titles = [s["title"] for s in secs]
|
||||
assert "安装" in titles
|
||||
assert "如何配置？" in titles
|
||||
|
||||
def test_level_recorded(self):
|
||||
secs = split_sections(DOC)
|
||||
sub = next(s for s in secs if s["title"] == "子配置")
|
||||
assert sub["level"] == 3
|
||||
|
||||
def test_body_captured(self):
|
||||
secs = split_sections(DOC)
|
||||
install = next(s for s in secs if s["title"] == "安装")
|
||||
assert "pip" in install["body"]
|
||||
|
||||
|
||||
class TestTokenize:
|
||||
def test_english(self):
|
||||
assert "install" in _tokenize_query("how to install")
|
||||
|
||||
def test_chinese_bigram(self):
|
||||
tokens = _tokenize_query("如何安装")
|
||||
assert "如何" in tokens or "安装" in tokens
|
||||
|
||||
|
||||
class TestSearch:
|
||||
def setup_method(self):
|
||||
self.index = build_index([{"path": "README", "content": DOC}])
|
||||
|
||||
def test_finds_install(self):
|
||||
results = search(self.index, "安装")
|
||||
assert results
|
||||
assert any("安装" in r["title"] for r in results)
|
||||
|
||||
def test_english_query(self):
|
||||
results = search(self.index, "pip")
|
||||
assert results
|
||||
assert "pip" in results[0]["snippet"]
|
||||
|
||||
def test_no_match(self):
|
||||
assert search(self.index, "zzzznotexist") == []
|
||||
|
||||
def test_empty_query(self):
|
||||
assert search(self.index, "") == []
|
||||
|
||||
def test_title_weighted(self):
|
||||
# 标题命中应排在前面
|
||||
results = search(self.index, "配置")
|
||||
assert results
|
||||
assert "配置" in results[0]["title"]
|
||||
|
||||
|
||||
class TestFaq:
|
||||
def test_extracts_question(self):
|
||||
index = build_index([{"path": "README", "content": DOC}])
|
||||
faq = extract_faq(index)
|
||||
assert any("配置" in f["question"] for f in faq)
|
||||
|
||||
def test_no_question(self):
|
||||
index = build_index([{"path": "x", "content": "# Title\n\nbody"}])
|
||||
assert extract_faq(index) == []
|
||||
|
||||
|
||||
class TestMap:
|
||||
def test_builds_map(self):
|
||||
index = build_index([{"path": "README", "content": DOC}])
|
||||
doc_map = build_map(index)
|
||||
assert "README" in doc_map
|
||||
titles = [s["title"] for s in doc_map["README"]]
|
||||
assert "安装" in titles
|
||||
|
||||
|
||||
class TestCollectDocs:
|
||||
class FakeClient:
|
||||
def readme(self, owner, repo, ref):
|
||||
return "# README\n\n内容"
|
||||
|
||||
def list_dir(self, owner, repo, path, ref):
|
||||
if path == "docs":
|
||||
return [{"name": "guide.md", "type": "file", "content": "# 指南\n\n步骤"}]
|
||||
return []
|
||||
|
||||
def file_content(self, owner, repo, filepath, ref):
|
||||
return None
|
||||
|
||||
def test_collects_readme_and_docs(self):
|
||||
docs = collect_docs("o", "r", "master", self.FakeClient())
|
||||
paths = {d["path"] for d in docs}
|
||||
assert "README" in paths
|
||||
assert any("guide" in p for p in paths)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(pytest.main([__file__, "-v"]))
|
||||
Loading…
Reference in New Issue