Initial commit: company-kb framework

- 完整的知识库框架（P0-P5 路线图）
- 12篇治理文档
- 工具脚本（kb-init.sh, kb-lint-fm.py, feishu-sync.py, kb-bot）
- 7个 Claude 命令
- 示例项目和测试项目（wanniu-l1）
- 契约文件（meta/kb-contract.yaml）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tools/feishu-pull.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""feishu-pull.py —— 飞书半自动单向导出（飞书 → docs/）
+
+MVP 形态：半自动单向。飞书只作「文档来源」，不直接写 Git。
+凭据走环境变量（绝不入库）：
+    FEISHU_APP_ID, FEISHU_APP_SECRET
+
+用法:
+    # 干跑（默认）：只打印将要导出什么，不落盘
+    python3 tools/feishu-pull.py <代号> --doc <飞书文档token或url>
+
+    # 实际写入 projects/<代号>/docs/
+    python3 tools/feishu-pull.py <代号> --doc <token> --write
+
+AI 自动补齐 frontmatter：title(H1回填) / ingested(今天) / source(飞书文档) / source_link。
+去重：写前查 meta/ingest-manifest.json 的内容 hash。
+
+注意：本文件是接口骨架。飞书 SDK 调用留 TODO，接入时填 _fetch_doc()。
+凭据缺失时优雅报错，不泄露。
+"""
+import sys
+import os
+import json
+import hashlib
+import argparse
+import datetime
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+MANIFEST = os.path.join(ROOT, "meta", "ingest-manifest.json")
+
+
+def load_manifest():
+    if os.path.isfile(MANIFEST):
+        with open(MANIFEST, encoding="utf-8") as f:
+            return json.load(f)
+    return {"_comment": "", "entries": {}}
+
+
+def save_manifest(m):
+    with open(MANIFEST, "w", encoding="utf-8") as f:
+        json.dump(m, f, ensure_ascii=False, indent=2)
+
+
+def content_hash(text):
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _fetch_doc(token):
+    """接入飞书时实现：用 tenant_access_token 拉文档，返回 (标题, markdown 正文, 原始链接)。
+    现为占位，避免误导：明确报未接入。"""
+    app_id = os.environ.get("FEISHU_APP_ID")
+    app_secret = os.environ.get("FEISHU_APP_SECRET")
+    if not app_id or not app_secret:
+        raise RuntimeError(
+            "缺凭据：请设置环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET（不要写进代码或 .env 入库）"
+        )
+    # TODO: 接入 lark-oapi SDK：
+    #   1) POST /open-apis/auth/v3/tenant_access_token/internal
+    #   2) GET  /open-apis/docx/v1/documents/{token}/raw_content
+    raise NotImplementedError(
+        "飞书 SDK 调用尚未接入（P2 阶段实现）。当前仅提供 frontmatter 自动补齐与去重骨架。"
+    )
+
+
+def build_page(title, body, source_link):
+    today = datetime.date.today().isoformat()
+    fm = (
+        "---\n"
+        "type: doc\n"
+        f'title: "{title}"\n'
+        "status: seed\n"
+        f"created: {today}\n"
+        f"ingested: {today}\n"
+        "source: 飞书文档\n"
+        f"source_link: {source_link}\n"
+        "tags: []\n"
+        "related: []\n"
+        "---\n\n"
+        f"# {title}\n\n"
+        f"{body}\n"
+    )
+    return fm
+
+
+def slugify(title):
+    keep = "".join(c if c.isalnum() or c in " -_" else "" for c in title)
+    return "-".join(keep.split())[:60] or "untitled"
+
+
+def main():
+    ap = argparse.ArgumentParser(description="飞书半自动单向导出")
+    ap.add_argument("code", help="项目代号（英文/拼音）")
+    ap.add_argument("--doc", required=True, help="飞书文档 token 或 URL")
+    ap.add_argument("--write", action="store_true", help="实际写盘（默认干跑）")
+    args = ap.parse_args()
+
+    outdir = os.path.join(ROOT, "projects", args.code, "docs")
+
+    try:
+        title, body, link = _fetch_doc(args.doc)
+    except (RuntimeError, NotImplementedError) as e:
+        print(f"[飞书导出未就绪] {e}", file=sys.stderr)
+        sys.exit(2)
+
+    page = build_page(title, body, link)
+    h = content_hash(body)
+    manifest = load_manifest()
+
+    if h in manifest.get("entries", {}):
+        print(f"[去重] 内容已存在于 {manifest['entries'][h]}，跳过。")
+        return
+
+    fname = slugify(title) + ".md"
+    dest = os.path.join(outdir, fname)
+    rel = os.path.relpath(dest, ROOT)
+
+    if not args.write:
+        print(f"[干跑] 将写入: {rel}\n--- frontmatter 预览 ---\n{page[:400]}...")
+        print("加 --write 实际落盘。")
+        return
+
+    os.makedirs(outdir, exist_ok=True)
+    with open(dest, "w", encoding="utf-8") as f:
+        f.write(page)
+    manifest.setdefault("entries", {})[h] = rel
+    save_manifest(manifest)
+    print(f"[已导出] {rel}")
+
+
+if __name__ == "__main__":
+    main()

tools/feishu-render.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""feishu-render.py —— 知识库 → 飞书呈现（单向 md→飞书，不回写）
+
+两种载体：
+  --bitable   扫 projects 下页的 frontmatter，每页一条记录 upsert 到多维表格（结构型呈现，可筛选/看板）
+  --docs 代号  把某项目的 _index + 决策同步为飞书云文档（叙事型呈现）
+
+飞书只读呈现，写路径仍是 Obsidian/Git，防双写冲突。
+凭据由 lark-cli 自管，不入库。字段严格对齐 meta/kb-contract.yaml。
+
+用法:
+    python3 tools/feishu-render.py --bitable [--dry-run]
+    python3 tools/feishu-render.py --docs <代号> [--dry-run]
+
+注意：本文件是接口骨架。bitable app/table 的创建与 record 写入留 TODO，
+接入时填 _bitable_upsert()；docs 同步复用 lark-doc-manager。
+"""
+import sys
+import os
+import re
+import glob
+import argparse
+
+try:
+    import yaml
+except ImportError:
+    print("ERROR: 需要 PyYAML。请运行: pip install pyyaml", file=sys.stderr)
+    sys.exit(2)
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+FM_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
+
+# 呈现给飞书多维表格的字段（对齐 kb-contract.yaml）
+BITABLE_FIELDS = ["type", "title", "status", "source", "created", "source_link"]
+
+
+def parse_fm(fp):
+    text = open(fp, encoding="utf-8").read()
+    m = FM_RE.match(text)
+    if not m:
+        return None
+    try:
+        data = yaml.safe_load(m.group(1))
+        return data if isinstance(data, dict) else None
+    except yaml.YAMLError:
+        return None
+
+
+def collect_pages():
+    """扫 projects 下所有知识页，返回 (相对路径, frontmatter) 列表。"""
+    rows = []
+    for fp in glob.glob(os.path.join(ROOT, "projects", "**", "*.md"), recursive=True):
+        fm = parse_fm(fp)
+        if fm and fm.get("type"):
+            rel = os.path.relpath(fp, ROOT)
+            project = rel.split(os.sep)[1] if os.sep in rel else ""
+            rows.append((rel, project, fm))
+    return rows
+
+
+def _bitable_upsert(rows, dry_run):
+    """接入时实现：通用 api 写多维表格记录。
+    唯一键 = 页相对路径（防重复插入，做 upsert）。
+    命令参考:
+      lark-cli api POST /open-apis/bitable/v1/apps/{app}/tables/{table}/records \
+        --data '{"fields": {...}}'
+    """
+    print(f"[bitable] 采集到 {len(rows)} 页知识：")
+    for rel, project, fm in rows:
+        fields = {k: fm.get(k, "") for k in BITABLE_FIELDS}
+        fields["项目"] = project
+        fields["页路径"] = rel
+        if dry_run:
+            print(f"  - {project}/{fm.get('type')}: {fm.get('title', rel)}")
+        else:
+            # TODO: lark-cli api POST .../records（含 app_token/table_id 配置）
+            pass
+    if not dry_run:
+        raise NotImplementedError(
+            "Bitable 写入尚未接入（P2-d）。需先创建多维表并配置 app_token/table_id。"
+            "当前 --dry-run 可预览将同步的记录。"
+        )
+
+
+def _docs_sync(code, dry_run):
+    idx = os.path.join(ROOT, "projects", code, "_index.md")
+    if not os.path.isfile(idx):
+        print(f"ERROR: 找不到 projects/{code}/_index.md", file=sys.stderr)
+        sys.exit(1)
+    print(f"[docs] 将同步 projects/{code}/ 的 _index + 决策为飞书云文档")
+    if dry_run:
+        print("  （dry-run：实际同步走 lark-doc-manager / lark-cli docs +create/+update）")
+        return
+    raise NotImplementedError(
+        "云文档同步走 lark-doc-manager skill（已激活）。接入时在此调用其创建/更新流程。"
+    )
+
+
+def main():
+    ap = argparse.ArgumentParser(description="知识库 → 飞书呈现")
+    g = ap.add_mutually_exclusive_group(required=True)
+    g.add_argument("--bitable", action="store_true", help="同步全库到多维表格")
+    g.add_argument("--docs", metavar="代号", help="同步某项目为云文档")
+    ap.add_argument("--dry-run", action="store_true", help="只预览不写飞书")
+    args = ap.parse_args()
+
+    if args.bitable:
+        _bitable_upsert(collect_pages(), args.dry_run)
+    else:
+        _docs_sync(args.docs, args.dry_run)
+
+
+if __name__ == "__main__":
+    main()

tools/feishu-sync.py

@@ -0,0 +1,576 @@
+#!/usr/bin/env python3
+"""feishu-sync.py —— 飞书文件夹递归镜像到本地（飞书 → 本地目录，单向）
+
+把飞书云盘里一个文件夹（含所有子文件夹）完整拉到本地：
+  - docx 云文档  → markdown 正文 + 图片下载到本地 assets/ + 图链接改本地路径 + 补 frontmatter
+  - sheet/bitable → 导出 xlsx（在线表格）
+  - file 上传件   → 原样下载（pdf/xmind/zip/视频…）
+  - mindnote 在线思维导图 → 跳过（拉它旁边导出的 .pdf/.xmind 即可）
+  - shortcut 快捷方式 → 跳过（只是指向别处的链接，无实际内容）
+
+凭据由 lark-cli 自管（lark-cli auth login），本脚本不碰任何密钥。
+
+用法:
+    # 干跑（默认）：只列出将要拉什么，不落盘
+    python3 tools/feishu-sync.py --folder <飞书文件夹token> --out <本地目录>
+
+    # 实际写盘
+    python3 tools/feishu-sync.py --folder <token> --out <本地目录> --write
+
+去重：每篇内容算 sha256 记进 <out>/.feishu-sync-manifest.json，没变过下次跳过。
+"""
+import argparse
+import datetime
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+
+# 飞书导出 markdown 里图片用的临时下载域名（会过期，需替换成本地路径）
+STREAM_HOST = "internal-api-drive-stream.feishu.cn"
+
+# 垃圾文件名：飞书/Mac 同步产生，不拉
+JUNK_NAMES = {".DS_Store", "DS_Store", ".ds_store", "Thumbs.db",
+              "desktop.ini", ".localized"}
+
+# 大文件扩展名（视频/音频/压缩包/大 PPT）：--skip-big 时跳过。
+# 这类无法转 markdown、体积大；飞书列表不返回体积，按扩展名判断最稳，
+# 避免每个文件多发一次探测请求。
+BIG_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".wmv", ".flv", ".m4v",
+            ".mp3", ".wav", ".zip", ".rar", ".7z", ".iso",
+            ".pptx", ".ppt"}
+# 每篇文档 URL 前缀按类型拼
+DOC_URL = {
+    "docx": "/docx/",
+    "doc": "/doc/",
+    "sheet": "/sheets/",
+    "bitable": "/base/",
+    "mindnote": "/mindnotes/",
+    "slides": "/slides/",
+}
+
+
+def find_lark_cli():
+    """定位 lark-cli，PATH 找不到就试常见位置。"""
+    p = shutil.which("lark-cli")
+    if p:
+        return p
+    for cand in [
+        os.path.expanduser("~/.npm-global/bin/lark-cli"),
+        os.path.expanduser("~/.nvm/versions/node/v22.17.1/bin/lark-cli"),
+    ]:
+        if os.path.isfile(cand):
+            return cand
+    # 最后从 nvm 目录里扫一个
+    nvm = os.path.expanduser("~/.nvm/versions/node")
+    if os.path.isdir(nvm):
+        for v in sorted(os.listdir(nvm), reverse=True):
+            cand = os.path.join(nvm, v, "bin", "lark-cli")
+            if os.path.isfile(cand):
+                return cand
+    sys.exit("找不到 lark-cli，请先安装并 `lark-cli auth login`。")
+
+
+LARK = find_lark_cli()
+
+
+def lark(args, binary_out=None, timeout=60, cwd=None):
+    """跑 lark-cli，返回解析后的 JSON（binary_out 模式返回 CompletedProcess）。
+
+    cwd：下载/导出类命令要求 --output 是「相对当前目录」的路径，故传 cwd=目标目录，
+    配合 --output ./文件名 使用（lark-cli 拒绝绝对路径）。
+    """
+    cmd = [LARK] + args
+    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, cwd=cwd)
+    if binary_out is not None:
+        return r
+    if not r.stdout.strip():
+        raise RuntimeError(f"lark-cli 无输出: {' '.join(args)}\n{r.stderr[:300]}")
+    try:
+        return json.loads(r.stdout)
+    except json.JSONDecodeError:
+        raise RuntimeError(f"lark-cli 输出非 JSON: {' '.join(args)}\n{r.stdout[:300]}")
+
+
+def list_folder(folder_token):
+    """列一个文件夹里的直接子项（自动翻页）。返回 [{name,type,token,created_time,url}...]
+
+    folder_token 为空或 "ROOT" 时列云盘根目录（不传 --folder-token，飞书返回全部根目录项）。
+    """
+    is_root = not folder_token or folder_token == "ROOT"
+    items = []
+    page_token = None
+    while True:
+        args = ["drive", "files", "list", "--page-size", "200", "--as", "user", "--json"]
+        if not is_root:
+            args += ["--folder-token", folder_token]
+        if page_token:
+            args += ["--page-token", page_token]
+        res = lark(args)
+        data = res.get("data", {})
+        items.extend(data.get("files", []))
+        if data.get("has_more") and data.get("next_page_token"):
+            page_token = data["next_page_token"]
+        else:
+            break
+    return items
+
+
+def list_bitable_tables(app_token):
+    """列一个多维表格的所有子表。返回 [{table_id, name}...]。失败返回空列表。"""
+    tables = []
+    page_token = None
+    while True:
+        params = {"page_size": 100}
+        if page_token:
+            params["page_token"] = page_token
+        try:
+            res = lark(["api", "GET",
+                        f"/open-apis/bitable/v1/apps/{app_token}/tables",
+                        "--params", json.dumps(params), "--as", "user", "--json"])
+        except RuntimeError:
+            return tables
+        data = res.get("data", {})
+        for it in data.get("items", []):
+            tables.append({"table_id": it.get("table_id"), "name": it.get("name", "")})
+        if data.get("has_more") and data.get("page_token"):
+            page_token = data["page_token"]
+        else:
+            break
+    return tables
+
+
+def safe_name(name, token, ext=""):
+    """把飞书文档名清成安全文件名；空名用 token。"""
+    name = (name or "").strip()
+    if not name:
+        name = token
+    name = re.sub(r'[/\\:*?"<>|]', "_", name)
+    name = name.strip(". ")[:120] or token
+    if ext and not name.lower().endswith(ext.lower()):
+        name += ext
+    return name
+
+
+def fetch_docx(token):
+    """拉一篇 docx：返回 (markdown正文, [媒体 dict 按出现顺序])。
+
+    媒体 dict: {"token": file_token, "mime": mime}。同时覆盖两种飞书导出形态：
+      - 图片    → <img src="file_token" mime="image/..">
+      - 视频/文件嵌入 → <figure><source token="file_token" mime="video/..">
+    两者在 markdown 里分别渲染成 ![](过期链接) 和 <figure><source href=过期链接>，
+    必须一起处理，否则视频类的过期链接会残留（违背「长久」）。
+    """
+    md_res = lark(["docs", "+fetch", "--doc", token,
+                   "--doc-format", "markdown", "--detail", "simple",
+                   "--as", "user", "--json"], timeout=120)
+    md = md_res.get("data", {}).get("document", {}).get("content", "")
+
+    xml_res = lark(["docs", "+fetch", "--doc", token,
+                    "--doc-format", "xml", "--detail", "full",
+                    "--as", "user", "--json"], timeout=120)
+    xml = xml_res.get("data", {}).get("document", {}).get("content", "")
+
+    # 按出现顺序扫描 <img> 与 <source>（figure 内），保持与 md 中链接顺序一致
+    media = []
+    for m in re.finditer(r'<(img|source)\b([^>]*)>', xml):
+        tag, attrs = m.group(1), m.group(2)
+        # img 的 file_token 在 src=；source 的在 token=
+        if tag == "img":
+            tm = re.search(r'\bsrc="([^"]+)"', attrs)
+        else:
+            tm = re.search(r'\btoken="([^"]+)"', attrs)
+        if not tm:
+            continue
+        mm = re.search(r'\bmime="([^"]+)"', attrs)
+        mime = mm.group(1) if mm else ("image/png" if tag == "img" else "")
+        media.append({"token": tm.group(1), "mime": mime})
+    return md, media
+
+
+def download_media(file_token, out_dir, basename):
+    """下一张文档内图片，返回本地文件名（含扩展名），失败返回 None。
+
+    lark-cli 的 --output 只认「相对当前目录」的路径，故 cwd=out_dir + --output ./basename。
+    """
+    os.makedirs(out_dir, exist_ok=True)
+    r = lark(["docs", "+media-download", "--token", file_token,
+              "--output", "./" + basename, "--overwrite", "--as", "user", "--json"],
+             timeout=120, cwd=out_dir)
+    saved = r.get("data", {}).get("saved_path")
+    if saved and os.path.isfile(saved):
+        return os.path.basename(saved)
+    # 兜底：目录里找 basename.*
+    for f in os.listdir(out_dir):
+        if f.startswith(os.path.basename(basename)):
+            return f
+    return None
+
+
+def localize_images(md, media, assets_dir, assets_rel, doc_slug, write):
+    """把 md 里会过期的飞书 stream 链接处理掉，保证不残留死链（长久性铁律）。
+
+    - 图片 (mime=image/*)      → 下载到本地 assets/，链接改本地相对路径
+    - 视频/其它 (mime=video/* 等) → 不下载，整段换成 `> 📎 [媒体已省略：类型，见原文]`
+    - 任何形态的过期链接都在最后兜底清除，绝不残留。
+
+    media 与 md 中过期链接按出现顺序一一对应（fetch_docx 已保证顺序）。
+    返回 (新md, 下载成功的图片数, 媒体总数)。
+    """
+    img_got = 0
+
+    # 处理每个媒体：图片下载、非图片省略。按序消费 media，逐个替换 md 中第一个未处理的过期链接。
+    def consume_next_link(new_text, replacement):
+        """把 new_text 中第一个出现的过期 stream 链接（含其外层包装）替换掉。
+
+        覆盖飞书导出的三种媒体形态；alt 用非贪婪匹配，兼容 alt 里含 ] [ 等字符。
+        """
+        host = re.escape(STREAM_HOST)
+        pats = [
+            # 形态1：markdown 图片 ![alt](stream_url)；alt 非贪婪，兼容 alt 内含 ]
+            r'!\[.*?\]\(https://' + host + r'/[^)]+\)',
+            # 形态2：<figure><source ... href="stream_url" .../></figure>
+            r'<figure\b[^>]*>\s*<source\b[^>]*\bhref="https://'
+            + host + r'/[^"]+"[^>]*/?>\s*</figure>',
+            # 形态3：HTML <img ... href="stream_url" .../>（表格单元格内常见）
+            r'<img\b[^>]*\bhref="https://' + host + r'/[^"]+"[^>]*/?>',
+        ]
+        best = None
+        for p in pats:
+            m = re.search(p, new_text)
+            if m and (best is None or m.start() < best.start()):
+                best = m
+        if best is None:
+            return new_text, False
+        return new_text[:best.start()] + replacement + new_text[best.end():], True
+
+    new_md = md
+    for i, item in enumerate(media):
+        mime = (item.get("mime") or "").lower()
+        if mime.startswith("image/"):
+            ext = "." + mime.split("/", 1)[1].split("+")[0] if "/" in mime else ".png"
+            fname = f"{doc_slug}-img{i+1}"
+            if write:
+                got = download_media(item["token"], assets_dir, fname)
+                rep = f"![]({assets_rel}/{got})" if got else None
+            else:
+                rep = f"![]({assets_rel}/{fname}{ext})"
+            if rep:
+                new_md, ok = consume_next_link(new_md, rep)
+                if ok and write:
+                    img_got += 1
+            # 下载失败则不消费，留给兜底清理
+        else:
+            kind = mime.split("/", 1)[0] if "/" in mime else "媒体"
+            note = f"> 📎 [已省略{kind}媒体，原文见 source_link]"
+            new_md, _ = consume_next_link(new_md, note)
+
+    # 兜底：清除任何仍残留的过期 stream 链接（各种包装形态），绝不留死链。
+    # 顺序从「包装最完整」到「最裸」，逐层收网；最后一条裸 URL catch-all 确保零残留。
+    NOTE = "> 📎 [已省略媒体，原文见 source_link]"
+    H = re.escape(STREAM_HOST)
+    # 形态A：<figure><source href="stream"...></figure>
+    new_md = re.sub(
+        r'<figure\b[^>]*>\s*<source\b[^>]*\bhref="https://' + H
+        + r'/[^"]+"[^>]*/?>\s*</figure>', NOTE, new_md)
+    # 形态B：HTML <img ... src/href="stream" ...>（含表格 <td> 里的）
+    new_md = re.sub(
+        r'<img\b[^>]*\b(?:src|href)="https://' + H + r'/[^"]*"[^>]*/?>',
+        NOTE, new_md)
+    # 形态C：markdown ![alt](stream)。alt 用 [\s\S]*? 非贪婪，允许 alt 内含 ] 等特殊字符
+    new_md = re.sub(
+        r'!\[[\s\S]*?\]\(https://' + H + r'/[^)]+\)', NOTE, new_md)
+    # 形态D：兜底 catch-all——任何裸露的 stream URL（连同可能的 markdown 链接括号）
+    new_md = re.sub(r'\(https://' + H + r'/[^)]+\)', "(见原文)", new_md)
+    new_md = re.sub(r'https://' + H + r'/\S+', "〔媒体链接已省略〕", new_md)
+
+    return new_md, img_got, len(media)
+
+
+def build_frontmatter(title, source_link, created_date):
+    today = datetime.date.today().isoformat()
+    title_esc = title.replace('"', "'")
+    return (
+        "---\n"
+        "type: doc\n"
+        f'title: "{title_esc}"\n'
+        "source: 飞书文档\n"
+        f"source_link: {source_link}\n"
+        f"created: {created_date}\n"
+        f"ingested: {today}\n"
+        "status: seed\n"
+        "tags: []\n"
+        "related: []\n"
+        "---\n\n"
+    )
+
+
+def ts_to_date(ts):
+    try:
+        return datetime.datetime.fromtimestamp(int(ts)).date().isoformat()
+    except (ValueError, TypeError):
+        return datetime.date.today().isoformat()
+
+
+def load_manifest(path):
+    if os.path.isfile(path):
+        try:
+            return json.load(open(path, encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            pass
+    return {"_comment": "feishu-sync 去重台账：key=内容sha256, value=本地相对路径", "entries": {}}
+
+
+def save_manifest(path, m):
+    json.dump(m, open(path, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
+
+
+# ── 统计 ──
+STATS = {"docx": 0, "file": 0, "export": 0, "csv": 0, "skip": 0, "skipbig": 0, "dedup": 0, "img": 0, "fail": 0}
+
+
+class ProgressList(list):
+    """记录每条处理结果的列表；LIVE=True 时 append 即实时打印并 flush，
+    让同事在长时间同步中能看到逐条动态，而不是一片死寂。"""
+    LIVE = False
+
+    def append(self, line):
+        super().append(line)
+        if ProgressList.LIVE:
+            print(line, flush=True)
+
+
+PLAN = ProgressList()  # 干跑记录待办；写盘时实时打印进度
+
+
+def handle_item(item, rel_path, out_root, host, manifest, write, skip_big=False):
+    typ = item.get("type")
+    name = item.get("name", "")
+    token = item.get("token")
+    created = ts_to_date(item.get("created_time"))
+    local_dir = os.path.join(out_root, rel_path) if rel_path else out_root
+
+    # 垃圾文件直接跳过（.DS_Store 等）
+    if name.strip() in JUNK_NAMES:
+        STATS["skip"] += 1
+        return
+
+    if typ == "shortcut":
+        STATS["skip"] += 1
+        PLAN.append(f"  ⏭️  跳过快捷方式: {rel_path}/{name}")
+        return
+    if typ == "mindnote":
+        STATS["skip"] += 1
+        PLAN.append(f"  ⏭️  跳过在线思维导图(拉旁边导出件): {rel_path}/{name}")
+        return
+
+    if typ == "docx":
+        slug = safe_name(name, token).rsplit(".", 1)[0]
+        fname = slug + ".md"
+        dest = os.path.join(local_dir, fname)
+        rel_dest = os.path.relpath(dest, out_root)
+        src_link = f"https://{host}{DOC_URL.get('docx','/docx/')}{token}"
+        if not write:
+            PLAN.append(f"  📄 docx → {rel_dest}  (+图片到 assets/)")
+            STATS["docx"] += 1
+            return
+        try:
+            md, media = fetch_docx(token)
+            h = hashlib.sha256(md.encode("utf-8")).hexdigest()
+            if h in manifest["entries"]:
+                STATS["dedup"] += 1
+                PLAN.append(f"  ♻️  未变跳过: {rel_dest}")
+                return
+            assets_dir = os.path.join(local_dir, "assets")
+            new_md, got, total = localize_images(
+                md, media, assets_dir, "assets", slug, write=True)
+            STATS["img"] += got
+            title = name or slug
+            # H1 回填 title：若正文首行是 # 标题，用它
+            m = re.match(r"^#\s+(.+)", new_md.strip())
+            if m:
+                title = m.group(1).strip()
+            os.makedirs(local_dir, exist_ok=True)
+            page = build_frontmatter(title, src_link, created) + new_md.strip() + "\n"
+            open(dest, "w", encoding="utf-8").write(page)
+            manifest["entries"][h] = rel_dest
+            STATS["docx"] += 1
+            PLAN.append(f"  ✅ {rel_dest}  (图 {got}/{total})")
+        except Exception as e:
+            STATS["fail"] += 1
+            PLAN.append(f"  ❌ docx 失败 {rel_dest}: {str(e)[:120]}")
+        return
+
+    if typ == "sheet":
+        base_name = safe_name(name, token)
+        fname = base_name + ".xlsx"
+        dest = os.path.join(local_dir, fname)
+        rel_dest = os.path.relpath(dest, out_root)
+        if not write:
+            PLAN.append(f"  📊 sheet → {rel_dest} (导出 xlsx)")
+            STATS["export"] += 1
+            return
+        try:
+            os.makedirs(local_dir, exist_ok=True)
+            # lark-cli 的 --output-dir 也只认 cwd 内的相对路径，故 cwd=目标目录 + 相对 "."
+            lark(["drive", "+export", "--token", token, "--doc-type", "sheet",
+                  "--file-extension", "xlsx", "--output-dir", ".",
+                  "--file-name", base_name,
+                  "--overwrite", "--as", "user", "--json"],
+                 timeout=180, cwd=local_dir)
+            STATS["export"] += 1
+            PLAN.append(f"  ✅ {rel_dest}")
+        except Exception as e:
+            STATS["fail"] += 1
+            PLAN.append(f"  ❌ 导出失败 {rel_dest}: {str(e)[:120]}")
+        return
+
+    if typ == "bitable":
+        # 双轨：.base 原生存档（保真、可导回）+ 每个子表一份 csv（可读、可检索）
+        base_name = safe_name(name, token)
+        rel_base = os.path.relpath(os.path.join(local_dir, base_name + ".base"), out_root)
+        if not write:
+            PLAN.append(f"  🗂️  bitable → {rel_base} (.base 存档 + 各子表 csv)")
+            STATS["export"] += 1
+            return
+        try:
+            os.makedirs(local_dir, exist_ok=True)
+            # ① .base 原生格式存档
+            lark(["drive", "+export", "--token", token, "--doc-type", "bitable",
+                  "--file-extension", "base", "--output-dir", ".",
+                  "--file-name", base_name,
+                  "--overwrite", "--as", "user", "--json"],
+                 timeout=180, cwd=local_dir)
+            STATS["export"] += 1
+            PLAN.append(f"  ✅ {rel_base}")
+            # ② 每个子表导一份 csv（可读检索）
+            tables = list_bitable_tables(token)
+            for t in tables:
+                sub_id = t.get("table_id")
+                sub_name = safe_name(t.get("name") or sub_id, sub_id)
+                csv_fname = f"{base_name}-{sub_name}"
+                rel_csv = os.path.relpath(
+                    os.path.join(local_dir, csv_fname + ".csv"), out_root)
+                try:
+                    lark(["drive", "+export", "--token", token,
+                          "--doc-type", "bitable", "--file-extension", "csv",
+                          "--sub-id", sub_id, "--output-dir", ".",
+                          "--file-name", csv_fname,
+                          "--overwrite", "--as", "user", "--json"],
+                         timeout=180, cwd=local_dir)
+                    STATS["csv"] += 1
+                    PLAN.append(f"     ↳ csv {rel_csv}")
+                except Exception as e:
+                    STATS["fail"] += 1
+                    PLAN.append(f"     ↳ ❌ csv 失败 {rel_csv}: {str(e)[:80]}")
+        except Exception as e:
+            STATS["fail"] += 1
+            PLAN.append(f"  ❌ bitable 失败 {rel_base}: {str(e)[:120]}")
+        return
+
+    if typ == "file":
+        fname = safe_name(name, token)
+        dest = os.path.join(local_dir, fname)
+        rel_dest = os.path.relpath(dest, out_root)
+        # 按扩展名跳过大文件（视频等）；飞书列表不返回体积，靠扩展名判断，避免逐个多发请求
+        if skip_big and os.path.splitext(fname)[1].lower() in BIG_EXTS:
+            STATS["skipbig"] += 1
+            PLAN.append(f"  ⏭️  跳过大文件(--skip-big): {rel_dest}")
+            return
+        if not write:
+            PLAN.append(f"  📎 file → {rel_dest} (原样下载)")
+            STATS["file"] += 1
+            return
+        try:
+            os.makedirs(local_dir, exist_ok=True)
+            # lark-cli 的 --output 只认「cwd 内的相对路径」，所以 cwd=目标目录 + 相对文件名
+            lark(["drive", "+download", "--file-token", token,
+                  "--output", "./" + fname, "--overwrite", "--as", "user", "--json"],
+                 timeout=300, cwd=local_dir)
+            STATS["file"] += 1
+            PLAN.append(f"  ✅ {rel_dest}")
+        except Exception as e:
+            STATS["fail"] += 1
+            PLAN.append(f"  ❌ 下载失败 {rel_dest}: {str(e)[:120]}")
+        return
+
+    # 其它未知类型
+    STATS["skip"] += 1
+    PLAN.append(f"  ⏭️  未知类型 {typ}: {rel_path}/{name}")
+
+
+def walk(folder_token, rel_path, out_root, host, manifest, write, skip_big=False, depth=0):
+    if depth > 20:
+        PLAN.append(f"  ⚠️ 目录过深(>20层)，停在 {rel_path}")
+        return
+    items = list_folder(folder_token)
+    for it in items:
+        if it.get("type") == "folder":
+            sub = os.path.join(rel_path, safe_name(it.get("name"), it.get("token"))) if rel_path \
+                else safe_name(it.get("name"), it.get("token"))
+            PLAN.append(f"📁 {sub}/")
+            walk(it["token"], sub, out_root, host, manifest, write, skip_big, depth + 1)
+        else:
+            handle_item(it, rel_path, out_root, host, manifest, write, skip_big)
+
+
+def main():
+    ap = argparse.ArgumentParser(description="飞书文件夹递归镜像到本地")
+    ap.add_argument("--folder", required=True, help="飞书文件夹 token")
+    ap.add_argument("--out", required=True, help="本地输出目录")
+    ap.add_argument("--host", default="ncnc1z9jg8c4.feishu.cn",
+                    help="飞书租户域名(拼 source_link 用)")
+    ap.add_argument("--write", action="store_true", help="实际写盘(默认干跑)")
+    ap.add_argument("--skip-big", action="store_true",
+                    help="跳过大文件(视频/压缩包/PPT 等，见 BIG_EXTS)，只拉可转 markdown 的文档")
+    args = ap.parse_args()
+
+    out_root = os.path.abspath(args.out)
+    manifest_path = os.path.join(out_root, ".feishu-sync-manifest.json")
+    manifest = load_manifest(manifest_path)
+
+    mode = "写盘" if args.write else "干跑(不落盘)"
+    print(f"=== feishu-sync [{mode}] ===")
+    print(f"飞书文件夹: {args.folder}")
+    print(f"本地输出:   {out_root}\n")
+
+    if args.write:
+        os.makedirs(out_root, exist_ok=True)
+        ProgressList.LIVE = True  # 写盘时逐条实时打印进度，避免长时间无输出
+        print("正在扫描飞书并逐个同步，下面会实时显示进度（文件多时需十几分钟，请耐心等待）：\n", flush=True)
+
+    try:
+        walk(args.folder, "", out_root, args.host, manifest, args.write, args.skip_big)
+    except Exception as e:
+        # 顶层兜底：飞书未登录/登录过期/网络异常时，给人话而非一屏 Traceback
+        msg = str(e)
+        if any(k in msg for k in ("access_token", "unauthor", "登录", "auth",
+                                   "not configured", "identity", "401", "403")):
+            print("\n⚠️  飞书未登录或登录已过期。请先运行 `lark-cli auth login "
+                  "--domain drive,docs --recommend` 扫码登录，再重试。", file=sys.stderr)
+        else:
+            print(f"\n⚠️  同步中断：{msg[:200]}\n"
+                  "   若反复出现，请把本提示截图求助。", file=sys.stderr)
+        sys.exit(2)
+
+    # 干跑：最后一次性打印全部计划；写盘：已在处理时实时打印过，不再重复
+    if not args.write:
+        print("\n".join(PLAN))
+    print("\n=== 汇总 ===")
+    print(f"云文档 docx: {STATS['docx']}   在线表格/bitable导出: {STATS['export']}   "
+          f"bitable子表csv: {STATS['csv']}   上传文件: {STATS['file']}")
+    print(f"跳过(思维导图/快捷方式): {STATS['skip']}   跳过大文件: {STATS['skipbig']}   "
+          f"去重跳过: {STATS['dedup']}   图片下载: {STATS['img']}   失败: {STATS['fail']}")
+
+    if args.write:
+        save_manifest(manifest_path, manifest)
+        print(f"\n台账已更新: {manifest_path}")
+    else:
+        print("\n以上为干跑预览。加 --write 实际落盘。")
+
+
+if __name__ == "__main__":
+    main()

tools/kb-bot/README.md

@@ -0,0 +1,43 @@
+# kb-bot —— 飞书群知识库问答机器人
+
+> 群里 @机器人 提问 → 查知识库 → 带溯源回复。**长连接，无需公网服务器。**
+> 底座：lark-cli（bot 身份自管凭据，不入库）。详见 `docs/08-飞书双向前端方案.md`。
+
+## 组件
+
+| 文件 | 作用 |
+|---|---|
+| `subscribe.sh` | 起 WebSocket 长连接，把 @机器人 群消息事件落到 `events/` |
+| `handle.py` | 轮询 `events/`，抽问题→调 `/kb-ask`→`lark-cli` 回复 |
+| `send.sh` | 手动发消息（验证发消息通路用） |
+
+## 分步验证
+
+```bash
+# P2-a 发消息：查群 id 后手动发一条
+lark-cli im +chat-search --query "你的群名"
+bash tools/kb-bot/send.sh oc_xxxxx "**kb-bot 上线测试**"
+
+# P2-b 收消息：起长连接，去群里 @机器人 说句话，看 events/ 是否落文件
+bash tools/kb-bot/subscribe.sh        # 前台观察
+
+# P2-c 闭环：另开一个终端跑消费者
+python3 tools/kb-bot/handle.py --watch
+```
+
+## 常驻部署（验证通过后）
+
+跑在任一常开机器（Mac mini / VPS 皆可，长连接不挑落点）：
+
+```bash
+nohup bash tools/kb-bot/subscribe.sh > /tmp/kb-bot-sub.log 2>&1 &
+nohup python3 tools/kb-bot/handle.py --watch > /tmp/kb-bot-handle.log 2>&1 &
+```
+
+稳定后再包成 launchd(macOS)/systemd(Linux) 常驻服务。
+
+## 铁律（继承知识库）
+
+- 机器人**只读**知识库，绝不写/删。
+- 答案强制带 `source_link` 溯源；无据回「知识库暂无」，**绝不编造**（由 `/kb-ask` 保证）。
+- `events/` 与 `.seen-msg-ids` 是运行时产物，已 gitignore。

tools/kb-bot/handle.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""handle.py —— 消费飞书群消息事件，调 /kb-ask 回答，回复到群。
+
+轮询 events/ 目录（subscribe.sh 落的事件文件）：
+  过滤 @机器人 的消息 → 抽问题 → CC headless 跑 /kb-ask → lark-cli 回复。
+
+用法:
+    python3 tools/kb-bot/handle.py            # 轮询一轮已有事件后退出
+    python3 tools/kb-bot/handle.py --watch    # 常驻轮询
+
+铁律：只读知识库、答案带溯源、无据报 Gap、绝不编造（由 /kb-ask 保证）。
+凭据由 lark-cli 自管，不在本脚本。
+"""
+import sys
+import os
+import re
+import json
+import time
+import subprocess
+
+BOT_DIR = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.abspath(os.path.join(BOT_DIR, "..", ".."))
+EVENTS_DIR = os.path.join(BOT_DIR, "events")
+DONE_DIR = os.path.join(BOT_DIR, "events", ".done")
+SEEN_FILE = os.path.join(BOT_DIR, ".seen-msg-ids")
+
+os.environ["PATH"] = os.path.expanduser("~/.npm-global/bin") + ":" + os.environ.get("PATH", "")
+
+
+def load_seen():
+    if os.path.isfile(SEEN_FILE):
+        return set(open(SEEN_FILE, encoding="utf-8").read().split())
+    return set()
+
+
+def mark_seen(mid):
+    with open(SEEN_FILE, "a", encoding="utf-8") as f:
+        f.write(mid + "\n")
+
+
+def extract(event):
+    """从事件里取 (message_id, chat_id, 纯文本问题)；非群 @ 消息返回 None。"""
+    # --compact 扁平结构；兼容原始嵌套结构
+    mid = event.get("message_id") or event.get("event", {}).get("message", {}).get("message_id")
+    chat_id = event.get("chat_id") or event.get("event", {}).get("message", {}).get("chat_id")
+    text = event.get("text") or ""
+    if not text:
+        content = event.get("content") or event.get("event", {}).get("message", {}).get("content", "")
+        if isinstance(content, str) and content:
+            try:
+                text = json.loads(content).get("text", "")
+            except json.JSONDecodeError:
+                text = content
+    if not (mid and chat_id and text):
+        return None
+    # 剥离 @_user_1 等 @ 标记
+    q = re.sub(r"@_?\w+", "", text).strip()
+    if not q:
+        return None
+    return mid, chat_id, q
+
+
+def ask_kb(question):
+    """调 CC headless 跑 /kb-ask，返回答案文本。"""
+    try:
+        r = subprocess.run(
+            ["claude", "-p", f"/kb-ask {question}"],
+            cwd=ROOT, capture_output=True, text=True, timeout=180,
+        )
+        return (r.stdout or "").strip() or "（知识库查询无输出，请稍后重试）"
+    except FileNotFoundError:
+        return "（CC 未安装，无法查询）"
+    except subprocess.TimeoutExpired:
+        return "（查询超时，请缩小问题范围）"
+
+
+def reply(chat_id, mid, answer):
+    subprocess.run(
+        ["lark-cli", "im", "+messages-reply",
+         "--message-id", mid, "--markdown", answer, "--as", "bot"],
+        capture_output=True, text=True,
+    )
+
+
+def process_once(seen):
+    if not os.path.isdir(EVENTS_DIR):
+        return
+    os.makedirs(DONE_DIR, exist_ok=True)
+    for fn in sorted(os.listdir(EVENTS_DIR)):
+        fp = os.path.join(EVENTS_DIR, fn)
+        if not fn.endswith(".json") or not os.path.isfile(fp):
+            continue
+        try:
+            event = json.load(open(fp, encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            continue
+        parsed = extract(event)
+        if parsed:
+            mid, chat_id, q = parsed
+            if mid not in seen:
+                answer = ask_kb(q)
+                reply(chat_id, mid, answer)
+                mark_seen(mid)
+                seen.add(mid)
+                print(f"[kb-bot] answered {mid}: {q[:40]}")
+        os.replace(fp, os.path.join(DONE_DIR, fn))
+
+
+def main():
+    seen = load_seen()
+    watch = "--watch" in sys.argv[1:]
+    while True:
+        process_once(seen)
+        if not watch:
+            break
+        time.sleep(3)
+
+
+if __name__ == "__main__":
+    main()

tools/kb-bot/send.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# send.sh —— 手动发一条 markdown 消息到飞书群（P2-a 验证发消息通路）
+#
+# 用法:
+#   bash tools/kb-bot/send.sh <chat-id> "markdown 内容"
+#   # chat-id 用 lark-cli im +chat-search --query "群名" 查
+#
+# 加 --dry-run 只打印不发。
+set -euo pipefail
+export PATH="$HOME/.npm-global/bin:$PATH"
+
+CHAT_ID="${1:-}"
+CONTENT="${2:-}"
+
+if [ -z "$CHAT_ID" ] || [ -z "$CONTENT" ]; then
+  echo "用法: bash tools/kb-bot/send.sh <chat-id> \"markdown 内容\"" >&2
+  echo "提示: 先用 lark-cli im +chat-search --query \"群名\" 查 chat-id" >&2
+  exit 1
+fi
+
+lark-cli im +messages-send \
+  --chat-id "$CHAT_ID" \
+  --markdown "$CONTENT" \
+  --as bot

tools/kb-bot/subscribe.sh

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+# subscribe.sh —— 飞书群消息长连接订阅（WebSocket，无需公网回调）
+#
+# 起一个常驻长连接，把 @机器人 的群消息事件逐条落到 events/ 目录，
+# 由 handle.py 消费。单实例锁由 lark-cli 保证（勿加 --force）。
+#
+# 用法:
+#   bash tools/kb-bot/subscribe.sh            # 前台跑
+#   nohup bash tools/kb-bot/subscribe.sh &    # 后台常驻
+#
+# 依赖: lark-cli 已装且 bot 身份可用（lark-cli auth status）。
+set -euo pipefail
+
+BOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+EVENTS_DIR="$BOT_DIR/events"
+mkdir -p "$EVENTS_DIR"
+
+export PATH="$HOME/.npm-global/bin:$PATH"
+
+if ! command -v lark-cli >/dev/null 2>&1; then
+  echo "ERROR: lark-cli 不在 PATH。请先安装并 lark-cli auth login。" >&2
+  exit 2
+fi
+
+echo "[kb-bot] 订阅飞书群消息事件 → $EVENTS_DIR"
+echo "[kb-bot] 只收 im.message.receive_v1；Ctrl-C 停止。"
+
+# --output-dir: 每个事件写成独立 JSON 文件，handle.py 轮询消费
+# --compact:    扁平化，提取 text，去噪
+# --quiet:      抑制 stderr 状态噪音（避免日志泄露）
+exec lark-cli event +subscribe \
+  --event-types im.message.receive_v1 \
+  --compact \
+  --quiet \
+  --output-dir "$EVENTS_DIR"

tools/kb-init.sh

@@ -0,0 +1,80 @@
+#!/usr/bin/env bash
+# kb-init.sh —— 幂等建项目骨架
+# 用法:
+#   bash tools/kb-init.sh <代号>        # 建 projects/<代号>/ 骨架 + _index.md(type: project, status: seed)
+#   bash tools/kb-init.sh --stats       # 打印源计数（更新 meta/stats.md 参考值）
+# 约束: <代号> 用英文/拼音，禁纯中文目录名（Claude Code 会把中文编码为 - 导致碰撞）。
+set -euo pipefail
+
+# 定位库根（本脚本在 tools/ 下）
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+
+usage() { grep '^#' "$0" | sed 's/^# \{0,1\}//'; exit "${1:-0}"; }
+
+# ── --stats: 红线观察点 ──
+if [[ "${1:-}" == "--stats" ]]; then
+  proj=$(find "$ROOT/projects" -mindepth 1 -maxdepth 1 -type d ! -name '_example' 2>/dev/null | wc -l | tr -d ' ')
+  pages=$(grep -rl '^type:' "$ROOT/projects" "$ROOT/entities" "$ROOT/concepts" "$ROOT/questions" 2>/dev/null | wc -l | tr -d ' ')
+  srcs=$(grep -rh '^source_link:' "$ROOT/projects" 2>/dev/null | sort -u | wc -l | tr -d ' ')
+  echo "项目数: $proj  |  知识页数: $pages  |  外部源数(去重): $srcs"
+  exit 0
+fi
+
+CODE="${1:-}"
+[[ -z "$CODE" || "$CODE" == "-h" || "$CODE" == "--help" ]] && usage 0
+
+# 中文目录名保护
+if echo "$CODE" | LC_ALL=C grep -q '[^ -~]'; then
+  echo "错误: 代号 '$CODE' 含非 ASCII 字符。请用英文/拼音（禁纯中文目录名）。" >&2
+  exit 1
+fi
+
+PDIR="$ROOT/projects/$CODE"
+if [[ -d "$PDIR" ]]; then
+  echo "项目已存在: projects/$CODE （幂等：不覆盖，仅补齐缺失子目录）"
+fi
+
+mkdir -p "$PDIR"/{docs,decisions,conversations,assets}
+for sub in docs assets; do [[ -f "$PDIR/$sub/.gitkeep" ]] || touch "$PDIR/$sub/.gitkeep"; done
+
+INDEX="$PDIR/_index.md"
+if [[ -f "$INDEX" ]]; then
+  echo "保留已有 _index.md（幂等）: projects/$CODE/_index.md"
+else
+  TODAY="$(date +%F)"
+  cat > "$INDEX" <<EOF
+---
+type: project
+title: "$CODE"
+description:
+status: seed
+created: $TODAY
+ingested: $TODAY
+tags: []
+related: []
+---
+
+# $CODE
+
+## 概览
+一句话说明本项目是什么、当前在做什么。
+
+## 进展
+-
+
+## 关键决策
+（见 decisions/，用 [[slug]] 链接）
+
+## 需求方 / 参与人
+-
+
+## 资料索引
+- docs/ ：外部汇入文档
+- conversations/ ：聊天沉淀
+- decisions/ ：决策记录
+EOF
+  echo "已创建: projects/$CODE/_index.md (type: project, status: seed)"
+fi
+
+echo "骨架就绪: projects/$CODE/{docs,decisions,conversations,assets}"
+echo "下一步: 丢资料 + 一句说明，或运行 kb-lint-fm.py 校验。"

tools/kb-lint-fm.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""kb-lint-fm.py —— frontmatter 校验器
+
+单一真相源：读 meta/kb-contract.yaml 判定 type 白名单 / 必填矩阵 / enum，
+绝不硬编码 type 或字段清单。
+
+用法:
+    python3 tools/kb-lint-fm.py [路径...]        # 默认扫全库
+    python3 tools/kb-lint-fm.py projects/foo     # 只扫某目录/文件
+
+退出码:
+    0 = 全部通过
+    1 = 有错误（缺硬必填 / 未激活 type / enum 非法 / 派生必填缺失）
+    2 = 运行环境问题（缺 PyYAML / 缺契约文件）
+供 CC 判定：非 0 即需处理。
+"""
+import sys
+import os
+import re
+import fnmatch
+
+try:
+    import yaml
+except ImportError:
+    print("ERROR: 需要 PyYAML。请运行: pip install pyyaml", file=sys.stderr)
+    sys.exit(2)
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+CONTRACT_PATH = os.path.join(ROOT, "meta", "kb-contract.yaml")
+
+FM_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
+
+
+def load_contract():
+    if not os.path.isfile(CONTRACT_PATH):
+        print(f"ERROR: 缺契约文件 {CONTRACT_PATH}", file=sys.stderr)
+        sys.exit(2)
+    with open(CONTRACT_PATH, encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def is_exempt(rel, patterns):
+    for pat in patterns:
+        if fnmatch.fnmatch(rel, pat) or fnmatch.fnmatch(rel, pat.rstrip("/") + "/*"):
+            return True
+    # 目录型 glob（docs/**）也匹配 docs/ 下任意层
+    for pat in patterns:
+        if pat.endswith("/**") and rel.startswith(pat[:-2]):
+            return True
+    return False
+
+
+def parse_frontmatter(text):
+    m = FM_RE.match(text)
+    if not m:
+        return None, "缺 frontmatter 块（--- 开头）"
+    try:
+        data = yaml.safe_load(m.group(1))
+    except yaml.YAMLError as e:
+        return None, f"frontmatter YAML 解析失败: {e}"
+    if not isinstance(data, dict):
+        return None, "frontmatter 不是键值映射"
+    return data, None
+
+
+def collect_md(paths):
+    files = []
+    for p in paths:
+        ap = os.path.abspath(p)
+        if os.path.isfile(ap) and ap.endswith(".md"):
+            files.append(ap)
+        elif os.path.isdir(ap):
+            for dp, _, fns in os.walk(ap):
+                for fn in fns:
+                    if fn.endswith(".md"):
+                        files.append(os.path.join(dp, fn))
+    return sorted(set(files))
+
+
+def check_derived(data, rules):
+    """返回派生必填字段名列表"""
+    req = []
+    t = data.get("type")
+    src = data.get("source")
+    for rule in rules or []:
+        w = rule.get("when", {})
+        ok = True
+        if "type_in" in w and t not in w["type_in"]:
+            ok = False
+        if "source_in" in w and src not in w["source_in"]:
+            ok = False
+        if ok:
+            req.extend(rule.get("require", []))
+    return req
+
+
+def main():
+    contract = load_contract()
+    types = {t["name"]: t for t in contract.get("types", [])}
+    active = {n for n, t in types.items() if t.get("active")}
+    fields = contract.get("fields", {})
+    enums = contract.get("enums", {})
+    exempt = contract.get("frontmatter_exempt", [])
+    source_exempt_types = set(contract.get("source_exempt_types", []))
+    derived = contract.get("derived_rules", [])
+
+    always_required = [f for f, spec in fields.items()
+                       if spec.get("required") == "always"]
+
+    targets = sys.argv[1:] or [ROOT]
+    files = collect_md(targets)
+
+    errors = 0
+    warnings = 0
+    checked = 0
+
+    for fp in files:
+        rel = os.path.relpath(fp, ROOT)
+        if is_exempt(rel, exempt):
+            continue
+        checked += 1
+        with open(fp, encoding="utf-8") as f:
+            text = f.read()
+
+        data, err = parse_frontmatter(text)
+        if err:
+            print(f"[ERROR] {rel}: {err}")
+            errors += 1
+            continue
+
+        # type 硬必填 + 白名单
+        t = data.get("type")
+        if not t:
+            print(f"[ERROR] {rel}: 缺硬必填字段 type")
+            errors += 1
+            continue
+        if t not in types:
+            print(f"[ERROR] {rel}: 未知 type '{t}'（不在契约中，扩展需改 kb-contract.yaml）")
+            errors += 1
+        elif t not in active:
+            print(f"[ERROR] {rel}: type '{t}' 已登记但未激活（改契约 active: true 再用）")
+            errors += 1
+
+        # 其余硬必填
+        for f_ in always_required:
+            if f_ == "type":
+                continue
+            if data.get(f_) in (None, ""):
+                print(f"[ERROR] {rel}: 缺硬必填字段 {f_}")
+                errors += 1
+
+        # 派生必填（project 型豁免溯源）
+        req = check_derived(data, derived)
+        if t in source_exempt_types:
+            req = [r for r in req if r not in ("source", "source_link")]
+        for f_ in req:
+            if data.get(f_) in (None, ""):
+                print(f"[ERROR] {rel}: 派生必填缺失 {f_}（type={t} + source 条件触发）")
+                errors += 1
+
+        # enum 校验
+        for f_, allowed in enums.items():
+            v = data.get(f_)
+            if v is not None and v not in allowed:
+                print(f"[ERROR] {rel}: 字段 {f_}='{v}' 不在 enum {allowed}")
+                errors += 1
+
+        # 软提示：建议但非必填的 title
+        if not data.get("title"):
+            print(f"[WARN]  {rel}: 建议补 title（AI 可自动=H1 或文件名）")
+            warnings += 1
+
+    print(f"\n检查 {checked} 页 | 错误 {errors} | 警告 {warnings}")
+    sys.exit(1 if errors else 0)
+
+
+if __name__ == "__main__":
+    main()