316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""
|
||
云文档集成：飞书、语雀、腾讯文档的文档创建/更新。
|
||
统一以 Markdown 为中间格式，由各平台 API 写入。
|
||
|
||
扩展建议：可增加「月度财务明细表」自动导出——每月在飞书/腾讯文档生成表格，
|
||
插入当月发票等附件预览链接，供财务查看（需对接财务记录与附件列表）。
|
||
"""
|
||
from typing import Any, Dict, List, Tuple
|
||
|
||
import httpx
|
||
|
||
FEISHU_BASE = "https://open.feishu.cn"
|
||
YUQUE_BASE = "https://www.yuque.com/api/v2"
|
||
|
||
|
||
async def get_feishu_tenant_token(app_id: str, app_secret: str) -> str:
|
||
"""获取飞书 tenant_access_token。"""
|
||
async with httpx.AsyncClient() as client:
|
||
r = await client.post(
|
||
f"{FEISHU_BASE}/open-apis/auth/v3/tenant_access_token/internal",
|
||
json={"app_id": app_id, "app_secret": app_secret},
|
||
timeout=10.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(data.get("msg", "飞书鉴权失败"))
|
||
return data["tenant_access_token"]
|
||
|
||
|
||
def _feishu_text_block_elements(md: str) -> List[Dict[str, Any]]:
|
||
"""将 Markdown 转为飞书文本块 elements（按行拆成 textRun，简单实现）。"""
|
||
elements: List[Dict[str, Any]] = []
|
||
for line in md.split("\n"):
|
||
line = line.rstrip()
|
||
if not line:
|
||
elements.append({"type": "textRun", "text_run": {"text": "\n"}})
|
||
else:
|
||
elements.append({"type": "textRun", "text_run": {"text": line + "\n"}})
|
||
if not elements:
|
||
elements.append({"type": "textRun", "text_run": {"text": " "}})
|
||
return elements
|
||
|
||
|
||
async def feishu_create_doc(
|
||
token: str, title: str, body_md: str, folder_token: str = ""
|
||
) -> Tuple[str, str]:
|
||
"""
|
||
创建飞书文档并写入内容。返回 (document_id, url)。
|
||
使用 docx/v1：创建文档后向根块下添加子块写入 Markdown 文本。
|
||
"""
|
||
async with httpx.AsyncClient() as client:
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
# 1. 创建文档
|
||
create_body: Dict[str, Any] = {"title": title[:50] or "未命名文档"}
|
||
if folder_token:
|
||
create_body["folder_token"] = folder_token
|
||
r = await client.post(
|
||
f"{FEISHU_BASE}/open-apis/docx/v1/documents",
|
||
headers=headers,
|
||
json=create_body,
|
||
timeout=15.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(data.get("msg", "飞书创建文档失败"))
|
||
doc = data.get("data", {})
|
||
document_id = doc.get("document", {}).get("document_id")
|
||
if not document_id:
|
||
raise RuntimeError("飞书未返回 document_id")
|
||
url = doc.get("document", {}).get("url", "")
|
||
# 2. 根块 ID 即 document_id（飞书约定）
|
||
block_id = document_id
|
||
# 3. 添加子块（内容）
|
||
elements = _feishu_text_block_elements(body_md)
|
||
# 单块有长度限制，分批写入多块
|
||
chunk_size = 3000
|
||
for i in range(0, len(elements), chunk_size):
|
||
chunk = elements[i : i + chunk_size]
|
||
body_json = {"children": [{"block_type": "text", "text": {"elements": chunk}}], "index": -1}
|
||
r3 = await client.post(
|
||
f"{FEISHU_BASE}/open-apis/docx/v1/documents/{document_id}/blocks/{block_id}/children",
|
||
headers=headers,
|
||
json=body_json,
|
||
timeout=15.0,
|
||
)
|
||
r3.raise_for_status()
|
||
res = r3.json()
|
||
if res.get("code") != 0:
|
||
raise RuntimeError(res.get("msg", "飞书写入块失败"))
|
||
# 下一批挂在刚创建的块下
|
||
new_items = res.get("data", {}).get("children", [])
|
||
if new_items:
|
||
block_id = new_items[0].get("block_id", block_id)
|
||
return document_id, url or f"https://feishu.cn/docx/{document_id}"
|
||
|
||
|
||
async def feishu_update_doc(token: str, document_id: str, body_md: str) -> str:
|
||
"""
|
||
更新飞书文档内容：获取现有块并批量更新首个文本块，或追加新块。
|
||
返回文档 URL。
|
||
"""
|
||
async with httpx.AsyncClient() as client:
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
r = await client.get(
|
||
f"{FEISHU_BASE}/open-apis/docx/v1/documents/{document_id}/blocks",
|
||
headers=headers,
|
||
params={"document_id": document_id},
|
||
timeout=10.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(data.get("msg", "飞书获取块失败"))
|
||
items = data.get("data", {}).get("items", [])
|
||
elements = _feishu_text_block_elements(body_md)
|
||
if items:
|
||
first_id = items[0].get("block_id")
|
||
if first_id:
|
||
# 批量更新：只更新第一个块的内容
|
||
update_body = {
|
||
"requests": [
|
||
{
|
||
"request_type": "blockUpdate",
|
||
"block_id": first_id,
|
||
"update_text": {"elements": elements},
|
||
}
|
||
]
|
||
}
|
||
r2 = await client.patch(
|
||
f"{FEISHU_BASE}/open-apis/docx/v1/documents/{document_id}/blocks/batch_update",
|
||
headers=headers,
|
||
json=update_body,
|
||
timeout=15.0,
|
||
)
|
||
r2.raise_for_status()
|
||
if r2.json().get("code") != 0:
|
||
# 若 PATCH 不支持该块类型，则追加新块
|
||
pass
|
||
else:
|
||
return f"https://feishu.cn/docx/{document_id}"
|
||
# 无块或更新失败：在根下追加子块
|
||
block_id = document_id
|
||
for i in range(0, len(elements), 3000):
|
||
chunk = elements[i : i + 3000]
|
||
body_json = {"children": [{"block_type": "text", "text": {"elements": chunk}}], "index": -1}
|
||
r3 = await client.post(
|
||
f"{FEISHU_BASE}/open-apis/docx/v1/documents/{document_id}/blocks/{block_id}/children",
|
||
headers=headers,
|
||
json=body_json,
|
||
timeout=15.0,
|
||
)
|
||
r3.raise_for_status()
|
||
res = r3.json()
|
||
if res.get("data", {}).get("children"):
|
||
block_id = res["data"]["children"][0].get("block_id", block_id)
|
||
return f"https://feishu.cn/docx/{document_id}"
|
||
|
||
|
||
# --------------- 语雀 ---------------
|
||
|
||
|
||
async def yuque_create_doc(
|
||
token: str, repo_id_or_namespace: str, title: str, body_md: str
|
||
) -> Tuple[str, str]:
|
||
"""
|
||
在语雀知识库创建文档。repo_id_or_namespace 可为 repo_id 或 namespace（如 user/repo）。
|
||
返回 (doc_id, url)。
|
||
"""
|
||
async with httpx.AsyncClient() as client:
|
||
headers = {
|
||
"X-Auth-Token": token,
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "OpsCore-CloudDoc/1.0",
|
||
}
|
||
# 若为 namespace 需先解析为 repo_id（语雀 API 创建文档用 repo_id）
|
||
repo_id = repo_id_or_namespace
|
||
if "/" in repo_id_or_namespace:
|
||
r_repo = await client.get(
|
||
f"{YUQUE_BASE}/repos/{repo_id_or_namespace}",
|
||
headers=headers,
|
||
timeout=10.0,
|
||
)
|
||
if r_repo.status_code == 200 and r_repo.json().get("data"):
|
||
repo_id = str(r_repo.json()["data"]["id"])
|
||
r = await client.post(
|
||
f"{YUQUE_BASE}/repos/{repo_id}/docs",
|
||
headers=headers,
|
||
json={
|
||
"title": title[:100] or "未命名",
|
||
"body": body_md,
|
||
"format": "markdown",
|
||
},
|
||
timeout=15.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
doc = data.get("data", {})
|
||
doc_id = str(doc.get("id", ""))
|
||
url = doc.get("url", "")
|
||
if not url and doc.get("slug"):
|
||
url = f"https://www.yuque.com/{doc.get('namespace', '').replace('/', '/')}/{doc.get('slug', '')}"
|
||
return doc_id, url or ""
|
||
|
||
|
||
async def yuque_update_doc(
|
||
token: str, repo_id_or_namespace: str, doc_id: str, title: str, body_md: str
|
||
) -> str:
|
||
"""更新语雀文档。返回文档 URL。"""
|
||
async with httpx.AsyncClient() as client:
|
||
headers = {
|
||
"X-Auth-Token": token,
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "OpsCore-CloudDoc/1.0",
|
||
}
|
||
r = await client.put(
|
||
f"{YUQUE_BASE}/repos/{repo_id_or_namespace}/docs/{doc_id}",
|
||
headers=headers,
|
||
json={
|
||
"title": title[:100] or "未命名",
|
||
"body": body_md,
|
||
"format": "markdown",
|
||
},
|
||
timeout=15.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
doc = data.get("data", {})
|
||
return doc.get("url", "") or f"https://www.yuque.com/docs/{doc_id}"
|
||
|
||
|
||
async def yuque_list_docs(token: str, repo_id_or_namespace: str) -> List[Dict[str, Any]]:
|
||
"""获取知识库文档列表。"""
|
||
async with httpx.AsyncClient() as client:
|
||
headers = {
|
||
"X-Auth-Token": token,
|
||
"User-Agent": "OpsCore-CloudDoc/1.0",
|
||
}
|
||
r = await client.get(
|
||
f"{YUQUE_BASE}/repos/{repo_id_or_namespace}/docs",
|
||
headers=headers,
|
||
timeout=10.0,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
return data.get("data", [])
|
||
|
||
|
||
# --------------- 腾讯文档（占位） ---------------
|
||
|
||
|
||
async def tencent_create_doc(client_id: str, client_secret: str, title: str, body_md: str) -> Tuple[str, str]:
|
||
"""
|
||
腾讯文档需 OAuth 用户授权与文件创建 API，此处返回占位。
|
||
正式接入需在腾讯开放平台创建应用并走 OAuth 流程。
|
||
"""
|
||
raise RuntimeError(
|
||
"腾讯文档 Open API 需在开放平台配置 OAuth 并获取用户授权；当前版本请先用飞书或语雀推送。"
|
||
)
|
||
|
||
|
||
# --------------- 统一入口 ---------------
|
||
|
||
|
||
class CloudDocManager:
|
||
"""统一封装：读取配置并执行创建/更新，支持增量（有 cloud_doc_id 则更新）。"""
|
||
|
||
def __init__(self, credentials: Dict[str, Dict[str, str]]):
|
||
self.credentials = credentials
|
||
|
||
async def push_markdown(
|
||
self,
|
||
platform: str,
|
||
title: str,
|
||
body_md: str,
|
||
existing_doc_id: str | None = None,
|
||
extra: Dict[str, str] | None = None,
|
||
) -> Tuple[str, str]:
|
||
"""
|
||
将 Markdown 推送到指定平台。若 existing_doc_id 存在则更新，否则创建。
|
||
返回 (cloud_doc_id, url)。
|
||
extra: 平台相关参数，如 yuque 的 default_repo。
|
||
"""
|
||
extra = extra or {}
|
||
if platform == "feishu":
|
||
cred = self.credentials.get("feishu") or {}
|
||
app_id = (cred.get("app_id") or "").strip()
|
||
app_secret = (cred.get("app_secret") or "").strip()
|
||
if not app_id or not app_secret:
|
||
raise RuntimeError("请先在设置中配置飞书 App ID 与 App Secret")
|
||
token = await get_feishu_tenant_token(app_id, app_secret)
|
||
if existing_doc_id:
|
||
url = await feishu_update_doc(token, existing_doc_id, body_md)
|
||
return existing_doc_id, url
|
||
return await feishu_create_doc(token, title, body_md)
|
||
|
||
if platform == "yuque":
|
||
cred = self.credentials.get("yuque") or {}
|
||
token = (cred.get("token") or "").strip()
|
||
default_repo = (cred.get("default_repo") or extra.get("repo") or "").strip()
|
||
if not token:
|
||
raise RuntimeError("请先在设置中配置语雀 Personal Access Token")
|
||
if not default_repo:
|
||
raise RuntimeError("请先在设置中配置语雀默认知识库（namespace，如 user/repo）")
|
||
if existing_doc_id:
|
||
url = await yuque_update_doc(token, default_repo, existing_doc_id, title, body_md)
|
||
return existing_doc_id, url
|
||
return await yuque_create_doc(token, default_repo, title, body_md)
|
||
|
||
if platform == "tencent":
|
||
await tencent_create_doc("", "", title, body_md)
|
||
return "", ""
|
||
|
||
raise RuntimeError(f"不支持的平台: {platform}")
|