dc-docs/skills/dingtalk-feishu-collector/scripts/step3_fetch_feishu_docs.py

"""Step 3: 拉取飞书文档内容"""
import argparse, json, os, subprocess, sys, re
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from paths import LINKS_DIR, DOCS_DIR, find_lark_cli

def lark_api(lark, path, params=None, timeout=30):
    cmd = [lark, "api", "GET", path, "--format", "json"]
    if params:
        cmd += ["--params", json.dumps(params, ensure_ascii=False)]
    r = subprocess.run(cmd, capture_output=True, timeout=timeout)
    out = r.stdout.decode("utf-8", errors="replace")
    err = r.stderr.decode("utf-8", errors="replace")
    if r.returncode != 0:
        return None, err or out
    try:
        data = json.loads(out)
    except Exception as e:
        return None, f"JSON解析失败: {e}"
    if data.get("ok") is False or data.get("code", 0) not in (0, None):
        return None, data.get("error") or data
    return data, None

def resolve_wiki(node_id, lark):
    data, _ = lark_api(lark, "/open-apis/wiki/v2/spaces/get_node", params={"token": node_id}, timeout=15)
    if not data:
        return "", ""
    node = data.get("data",{}).get("node",{})
    return node.get("obj_token",""), node.get("title","")

def extract_text(obj):
    if not obj: return ""
    return "".join(e.get("text_run",{}).get("content","") for e in obj.get("elements",[])).strip()

def blocks_to_md(blocks, doc_id):
    lines = [f"# {doc_id}\n", f"Source: https://feishu.cn/docx/{doc_id}\n"]
    for b in blocks:
        bt = b.get("block_type",0)
        if bt==2: t=extract_text(b.get("text",{})); [lines.append(t)] if t else None
        elif bt in range(3,10): t=extract_text(b.get(f"heading{bt-2}",{}) or b.get("text",{})); lines.append(f"{'#'*(bt-2)} {t}") if t else None
        elif bt==10: t=extract_text(b.get("bullet",{}) or b.get("text",{})); lines.append(f"- {t}") if t else None
        elif bt==11: t=extract_text(b.get("ordered",{}) or b.get("text",{})); lines.append(f"1. {t}") if t else None
        elif bt==14: t=extract_text(b.get("quote",{}) or b.get("text",{})); lines.append(f"> {t}") if t else None
        elif bt==15: t=extract_text(b.get("code",{}) or b.get("text",{})); lines.append(f"```\n{t}\n```") if t else None
        else:
            for k in ["text","heading1","heading2","heading3","bullet","ordered","quote","code"]:
                if k in b: t=extract_text(b[k]); [lines.append(t)] if t else None; break
    return "\n\n".join(lines)

def fetch_doc(doc_id, doc_type, lark):
    real_id, wiki_title = (doc_id, "")
    if doc_type=="wiki":
        real_id, wiki_title = resolve_wiki(doc_id, lark)
    if not real_id:
        return f"# {doc_id}\n\n> wiki解析失败"
    data, error = lark_api(lark, f"/open-apis/docx/v1/documents/{real_id}/blocks", timeout=30)
    if error:
        return f"# {doc_id}\n\n> 获取失败: {error}"
    blocks = data.get("data",{}).get("items",[]) if data else []
    md = blocks_to_md(blocks, doc_id) if blocks else f"# {doc_id}\n\n> 空文档"
    if wiki_title:
        md = md.replace(f"# {doc_id}", f"# {wiki_title}", 1)
    return md

def should_fetch(doc_id, existing, docs_dir):
    old = existing.get(doc_id)
    if not old:
        return True
    if old.get("content_length", 0) <= 200:
        return True
    path = os.path.join(docs_dir, f"{doc_id}.md")
    if not os.path.isfile(path):
        return True
    with open(path, "r", encoding="utf-8") as f:
        head = f.read(500)
    return any(marker in head for marker in ["获取失败", "wiki解析失败", "NoneType", "空文档"])

def is_error_markdown(md):
    head = md[:1000]
    return any(marker in head for marker in ["获取失败", "wiki解析失败", "NoneType", "空文档", "forBidden", "permission denied"])

def main():
    p = argparse.ArgumentParser(); p.add_argument("--incremental",action="store_true"); args = p.parse_args()
    links_path = os.path.join(LINKS_DIR,"all_feishu_links.json")
    if not os.path.isfile(links_path): print("ERROR: 先运行 step2"); sys.exit(1)
    with open(links_path,"r",encoding="utf-8") as f: all_links = json.load(f).get("links",[])
    print(f"链接: {len(all_links)} 个")

    content_path = os.path.join(LINKS_DIR,"all_feishu_content.json")
    existing = {}
    if os.path.isfile(content_path):
        with open(content_path,"r",encoding="utf-8") as f:
            for d in json.load(f).get("documents",[]): existing[d.get("doc_id","")] = d
    print(f"已拉取: {len(existing)}")

    lark = find_lark_cli()
    os.makedirs(DOCS_DIR, exist_ok=True)
    new_count = 0
    for link in all_links:
        did = link.get("doc_id","")
        if not did or not should_fetch(did, existing, DOCS_DIR): continue
        print(f"  {did} ({link.get('sender','?')})")
        md = fetch_doc(did, link.get("type","docx"), lark)
        fetched = not is_error_markdown(md)
        old = existing.get(did)
        if old and old.get("content_length", 0) > 200 and not fetched:
            print("    WARN: 获取失败，保留已有正文")
            continue
        with open(os.path.join(DOCS_DIR,f"{did}.md"),"w",encoding="utf-8") as f: f.write(md)
        title = ""
        for line in md.split("\n"):
            if line.startswith("# "): title = line[2:].strip(); break
        existing[did] = {"url":link.get("url",""),"doc_id":did,"title":title,"sender":link.get("sender",""),"group":link.get("group",""),"time":link.get("time",""),"content_length":len(md),"fetched":fetched}
        new_count += 1

    with open(content_path,"w",encoding="utf-8") as f:
        json.dump({"date":datetime.now().strftime("%Y-%m-%d"),"total":len(existing),"documents":list(existing.values())},f,ensure_ascii=False,indent=2)
    print(f"\n新增 {new_count}，总计 {len(existing)} -> {DOCS_DIR}")

if __name__ == "__main__": main()