company-kb/tools/feishu-sync.py

#!/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()