From 2fdb56d7a6d5ce61eff1abc64ac00a226363bbae Mon Sep 17 00:00:00 2001
From: Ct201314 <1195214305@qq.com>
Date: Fri, 12 Jun 2026 18:00:53 +0800
Subject: [PATCH] =?UTF-8?q?feat(skills):=20=E8=A1=A5=E5=85=85=20gitlink-kb?=
 =?UTF-8?q?=20=E9=85=8D=E5=A5=97=E8=84=9A=E6=9C=AC=E4=B8=8E=E6=B5=8B?=
 =?UTF-8?q?=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

依审阅意见，将 SKILL.md 方式A引用的脚本（scripts/）与单元测试（tests/）一并纳入，使 PR 自包含可运行。
---
 skills/gitlink-kb/scripts/glapi.py | 241 ++++++++++++++++++++++++
 skills/gitlink-kb/scripts/kb.py    | 293 +++++++++++++++++++++++++++++
 skills/gitlink-kb/tests/test_kb.py | 130 +++++++++++++
 3 files changed, 664 insertions(+)
 create mode 100644 skills/gitlink-kb/scripts/glapi.py
 create mode 100644 skills/gitlink-kb/scripts/kb.py
 create mode 100644 skills/gitlink-kb/tests/test_kb.py

diff --git a/skills/gitlink-kb/scripts/glapi.py b/skills/gitlink-kb/scripts/glapi.py
new file mode 100644
index 0000000..41db09d
--- /dev/null
+++ b/skills/gitlink-kb/scripts/glapi.py
@@ -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}")
diff --git a/skills/gitlink-kb/scripts/kb.py b/skills/gitlink-kb/scripts/kb.py
new file mode 100644
index 0000000..1176121
--- /dev/null
+++ b/skills/gitlink-kb/scripts/kb.py
@@ -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())
diff --git a/skills/gitlink-kb/tests/test_kb.py b/skills/gitlink-kb/tests/test_kb.py
new file mode 100644
index 0000000..8e4444c
--- /dev/null
+++ b/skills/gitlink-kb/tests/test_kb.py
@@ -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"]))