452 lines
15 KiB
Python
452 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
飞书云文档操作工具类
|
|
包含文档创建、内容块管理、图片上传等功能
|
|
"""
|
|
|
|
import os
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
|
|
|
|
class FeishuDocx:
|
|
"""飞书云文档操作工具类（含图片上传）"""
|
|
|
|
BASE_URL = "https://open.feishu.cn/open-apis"
|
|
DEFAULT_APP_ID = "cli_a9f29dca82b9dbef"
|
|
DEFAULT_APP_SECRET = "sDfhjG7QT1S4gfHiMVYSygmPQPN1R2Ho"
|
|
|
|
def __init__(self, app_id: str = None, app_secret: str = None):
|
|
self.app_id = app_id or os.environ.get("FEISHU_APP_ID") or self.DEFAULT_APP_ID
|
|
self.app_secret = app_secret or os.environ.get("FEISHU_APP_SECRET") or self.DEFAULT_APP_SECRET
|
|
self._token = None
|
|
self._token_expires = None
|
|
|
|
@property
|
|
def token(self) -> str:
|
|
"""获取或刷新 access token"""
|
|
if self._token and self._token_expires and datetime.now() < self._token_expires:
|
|
return self._token
|
|
|
|
url = f"{self.BASE_URL}/auth/v3/tenant_access_token/internal"
|
|
response = requests.post(url, json={
|
|
"app_id": self.app_id,
|
|
"app_secret": self.app_secret
|
|
})
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"获取 token 失败: {data}")
|
|
|
|
self._token = data["tenant_access_token"]
|
|
self._token_expires = datetime.now() + timedelta(seconds=data.get("expire", 7200) - 60)
|
|
return self._token
|
|
|
|
@property
|
|
def headers(self) -> dict:
|
|
"""获取请求头"""
|
|
return {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# ==================== 文档管理 ====================
|
|
|
|
def set_document_permission(self, document_id: str, editable: bool = True) -> bool:
|
|
"""
|
|
设置文档权限
|
|
|
|
Args:
|
|
document_id: 文档ID
|
|
editable: True=组织内可编辑, False=组织内只读
|
|
"""
|
|
url = f"{self.BASE_URL}/drive/v1/permissions/{document_id}/public"
|
|
payload = {
|
|
"external_access_entity": "open",
|
|
"security_entity": "anyone_can_view",
|
|
"comment_entity": "anyone_can_view",
|
|
"share_entity": "anyone",
|
|
"link_share_entity": "tenant_editable" if editable else "tenant_readable",
|
|
"invite_external": False
|
|
}
|
|
response = requests.patch(url, headers=self.headers, params={"type": "docx"}, json=payload)
|
|
return response.json().get("code") == 0
|
|
|
|
def create_document(self, title: str, folder_token: str = None, editable: bool = True) -> dict:
|
|
"""
|
|
创建文档（自动设置为组织内可编辑）
|
|
|
|
Args:
|
|
title: 文档标题
|
|
folder_token: 文件夹token (可选)
|
|
editable: 是否设置为组织内可编辑 (默认True)
|
|
|
|
Returns:
|
|
dict: {"document_id": "xxx", "title": "xxx", "url": "xxx"}
|
|
"""
|
|
url = f"{self.BASE_URL}/docx/v1/documents"
|
|
payload = {"title": title}
|
|
if folder_token:
|
|
payload["folder_token"] = folder_token
|
|
|
|
response = requests.post(url, headers=self.headers, json=payload)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"创建文档失败: {data}")
|
|
|
|
doc = data["data"]["document"]
|
|
document_id = doc["document_id"]
|
|
|
|
if editable:
|
|
self.set_document_permission(document_id, editable=True)
|
|
|
|
return {
|
|
"document_id": document_id,
|
|
"title": doc["title"],
|
|
"url": f"https://feishu.cn/docx/{document_id}"
|
|
}
|
|
|
|
def get_document_blocks(self, document_id: str, page_size: int = 500) -> List[Dict]:
|
|
"""获取文档所有内容块"""
|
|
url = f"{self.BASE_URL}/docx/v1/documents/{document_id}/blocks"
|
|
all_blocks = []
|
|
page_token = None
|
|
|
|
while True:
|
|
params = {"page_size": page_size}
|
|
if page_token:
|
|
params["page_token"] = page_token
|
|
|
|
response = requests.get(url, headers=self.headers, params=params)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"获取内容块失败: {data}")
|
|
|
|
all_blocks.extend(data["data"].get("items", []))
|
|
|
|
if not data["data"].get("has_more"):
|
|
break
|
|
page_token = data["data"]["page_token"]
|
|
|
|
return all_blocks
|
|
|
|
# ==================== 内容块操作 ====================
|
|
|
|
def create_blocks(self, document_id: str, blocks: list, parent_id: str = None) -> List[Dict]:
|
|
"""
|
|
创建内容块
|
|
|
|
Args:
|
|
document_id: 文档ID
|
|
blocks: 内容块列表
|
|
parent_id: 父块ID (默认为文档根块)
|
|
"""
|
|
parent_id = parent_id or document_id
|
|
url = f"{self.BASE_URL}/docx/v1/documents/{document_id}/blocks/{parent_id}/children"
|
|
|
|
response = requests.post(
|
|
url,
|
|
headers=self.headers,
|
|
params={"document_revision_id": -1},
|
|
json={"children": blocks}
|
|
)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"创建内容块失败: {data}")
|
|
|
|
return data["data"]["children"]
|
|
|
|
# ==================== 图片上传 (三步流程) ====================
|
|
|
|
def create_empty_image_block(self, document_id: str, parent_id: str = None) -> str:
|
|
"""
|
|
Step 1: 创建空图片块
|
|
|
|
Args:
|
|
document_id: 文档ID
|
|
parent_id: 父块ID (默认为文档根块)
|
|
|
|
Returns:
|
|
block_id: 图片块ID
|
|
"""
|
|
parent_id = parent_id or document_id
|
|
url = f"{self.BASE_URL}/docx/v1/documents/{document_id}/blocks/{parent_id}/children"
|
|
|
|
payload = {
|
|
"children": [{
|
|
"block_type": 27,
|
|
"image": {}
|
|
}]
|
|
}
|
|
|
|
response = requests.post(
|
|
url,
|
|
headers=self.headers,
|
|
params={"document_revision_id": -1},
|
|
json=payload
|
|
)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"创建图片块失败: {data}")
|
|
|
|
return data["data"]["children"][0]["block_id"]
|
|
|
|
def upload_image_to_block(self, file_path: str, block_id: str) -> str:
|
|
"""
|
|
Step 2: 上传图片文件
|
|
|
|
Args:
|
|
file_path: 本地图片路径
|
|
block_id: 图片块ID (从 Step 1 获取)
|
|
|
|
Returns:
|
|
file_token: 图片token
|
|
"""
|
|
url = f"{self.BASE_URL}/drive/v1/medias/upload_all"
|
|
file_name = os.path.basename(file_path)
|
|
file_size = os.path.getsize(file_path)
|
|
|
|
# 根据文件扩展名确定 MIME 类型
|
|
ext = os.path.splitext(file_path)[1].lower()
|
|
mime_types = {
|
|
'.png': 'image/png',
|
|
'.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg',
|
|
'.gif': 'image/gif',
|
|
'.webp': 'image/webp',
|
|
'.bmp': 'image/bmp',
|
|
}
|
|
mime_type = mime_types.get(ext, 'image/png')
|
|
|
|
with open(file_path, 'rb') as f:
|
|
files = {'file': (file_name, f, mime_type)}
|
|
data = {
|
|
'file_name': file_name,
|
|
'parent_type': 'docx_image',
|
|
'parent_node': block_id,
|
|
'size': str(file_size)
|
|
}
|
|
response = requests.post(
|
|
url,
|
|
headers={"Authorization": f"Bearer {self.token}"},
|
|
files=files,
|
|
data=data
|
|
)
|
|
|
|
result = response.json()
|
|
if result.get("code") != 0:
|
|
raise Exception(f"上传图片失败: {result}")
|
|
|
|
return result["data"]["file_token"]
|
|
|
|
def bind_image_to_block(self, document_id: str, block_id: str, file_token: str) -> bool:
|
|
"""
|
|
Step 3: 绑定图片到图片块 (关键步骤!)
|
|
|
|
Args:
|
|
document_id: 文档ID
|
|
block_id: 图片块ID
|
|
file_token: 图片token (从 Step 2 获取)
|
|
|
|
Returns:
|
|
bool: 是否成功
|
|
"""
|
|
url = f"{self.BASE_URL}/docx/v1/documents/{document_id}/blocks/{block_id}"
|
|
|
|
payload = {
|
|
"replace_image": {
|
|
"token": file_token
|
|
}
|
|
}
|
|
|
|
response = requests.patch(
|
|
url,
|
|
headers=self.headers,
|
|
params={"document_revision_id": -1},
|
|
json=payload
|
|
)
|
|
data = response.json()
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"绑定图片失败: {data}")
|
|
|
|
return True
|
|
|
|
def insert_image(self, document_id: str, file_path: str, parent_id: str = None) -> dict:
|
|
"""
|
|
插入图片到文档 (封装三步流程)
|
|
|
|
Args:
|
|
document_id: 文档ID
|
|
file_path: 本地图片路径
|
|
parent_id: 父块ID (默认为文档根块)
|
|
|
|
Returns:
|
|
dict: {"block_id": "xxx", "file_token": "xxx"}
|
|
"""
|
|
# Step 1: 创建空图片块
|
|
block_id = self.create_empty_image_block(document_id, parent_id)
|
|
|
|
# Step 2: 上传图片
|
|
file_token = self.upload_image_to_block(file_path, block_id)
|
|
|
|
# Step 3: 绑定图片
|
|
self.bind_image_to_block(document_id, block_id, file_token)
|
|
|
|
return {"block_id": block_id, "file_token": file_token}
|
|
|
|
def check_image_status(self, document_id: str) -> List[Dict]:
|
|
"""
|
|
检查文档中图片块的状态
|
|
|
|
Returns:
|
|
List[Dict]: 图片块状态列表
|
|
"""
|
|
blocks = self.get_document_blocks(document_id)
|
|
images = []
|
|
|
|
for block in blocks:
|
|
if block.get("block_type") == 27:
|
|
image_data = block.get("image", {})
|
|
images.append({
|
|
"block_id": block.get("block_id"),
|
|
"token": image_data.get("token", ""),
|
|
"width": image_data.get("width", 0),
|
|
"height": image_data.get("height", 0),
|
|
"is_valid": bool(image_data.get("token"))
|
|
})
|
|
|
|
return images
|
|
|
|
# ==================== 内容块构建器 ====================
|
|
|
|
@staticmethod
|
|
def heading1(text: str) -> dict:
|
|
"""一级标题"""
|
|
return {"block_type": 3, "heading1": {"elements": [{"text_run": {"content": text}}]}}
|
|
|
|
@staticmethod
|
|
def heading2(text: str) -> dict:
|
|
"""二级标题"""
|
|
return {"block_type": 4, "heading2": {"elements": [{"text_run": {"content": text}}]}}
|
|
|
|
@staticmethod
|
|
def heading3(text: str) -> dict:
|
|
"""三级标题"""
|
|
return {"block_type": 5, "heading3": {"elements": [{"text_run": {"content": text}}]}}
|
|
|
|
@staticmethod
|
|
def text(content: str) -> dict:
|
|
"""文本段落"""
|
|
return {"block_type": 2, "text": {"elements": [{"text_run": {"content": content}}]}}
|
|
|
|
@staticmethod
|
|
def bullet(content: str) -> dict:
|
|
"""无序列表项"""
|
|
return {"block_type": 12, "bullet": {"elements": [{"text_run": {"content": content}}]}}
|
|
|
|
@staticmethod
|
|
def ordered(content: str) -> dict:
|
|
"""有序列表项"""
|
|
return {"block_type": 13, "ordered": {"elements": [{"text_run": {"content": content}}]}}
|
|
|
|
@staticmethod
|
|
def code(content: str, language: int = 1) -> dict:
|
|
"""代码块 (language: 1=JSON, 4=Python, 等)"""
|
|
return {"block_type": 14, "code": {"elements": [{"text_run": {"content": content}}], "language": language}}
|
|
|
|
@staticmethod
|
|
def divider() -> dict:
|
|
"""分割线"""
|
|
return {"block_type": 22, "divider": {}}
|
|
|
|
@staticmethod
|
|
def todo(content: str, done: bool = False) -> dict:
|
|
"""待办事项"""
|
|
return {"block_type": 17, "todo": {"elements": [{"text_run": {"content": content}}], "style": {"done": done}}}
|
|
|
|
# ==================== 高级功能 ====================
|
|
|
|
def create_meeting_minutes(self, title: str, content: dict) -> dict:
|
|
"""
|
|
创建会议纪要
|
|
|
|
Args:
|
|
title: 文档标题
|
|
content: {
|
|
"summary": "会议摘要",
|
|
"points": [{"title": "...", "items": ["..."]}],
|
|
"todos": [{"assignee": "...", "task": "..."}]
|
|
}
|
|
|
|
Returns:
|
|
dict: 文档信息
|
|
"""
|
|
doc = self.create_document(title)
|
|
document_id = doc["document_id"]
|
|
|
|
blocks = []
|
|
|
|
if content.get("summary"):
|
|
blocks.append(self.heading2("会议摘要"))
|
|
blocks.append(self.text(content["summary"]))
|
|
blocks.append(self.divider())
|
|
|
|
if content.get("points"):
|
|
blocks.append(self.heading2("小结"))
|
|
for point in content["points"]:
|
|
blocks.append(self.heading3(point["title"]))
|
|
for item in point.get("items", []):
|
|
blocks.append(self.bullet(item))
|
|
blocks.append(self.divider())
|
|
|
|
if content.get("todos"):
|
|
blocks.append(self.heading2("待办事项"))
|
|
for todo_item in content["todos"]:
|
|
text = f"{todo_item['assignee']}: {todo_item['task']}"
|
|
blocks.append(self.todo(text))
|
|
|
|
self.create_blocks(document_id, blocks)
|
|
return doc
|
|
|
|
|
|
# ==================== 使用示例 ====================
|
|
if __name__ == "__main__":
|
|
docx = FeishuDocx()
|
|
|
|
# 示例1: 创建文档并插入图片
|
|
print("=== 示例1: 创建带图片的文档 ===")
|
|
doc = docx.create_document("图片上传测试文档")
|
|
print(f"文档创建成功: {doc['url']}")
|
|
|
|
# 如果有测试图片，取消注释以下代码
|
|
# result = docx.insert_image(doc["document_id"], "/path/to/image.png")
|
|
# print(f"图片插入成功: {result}")
|
|
|
|
# 示例2: 创建会议纪要
|
|
print("\n=== 示例2: 创建会议纪要 ===")
|
|
doc = docx.create_meeting_minutes(
|
|
"测试会议纪要",
|
|
{
|
|
"summary": "本次会议讨论了项目进度...",
|
|
"points": [{"title": "进度汇报", "items": ["任务A已完成", "任务B进行中"]}],
|
|
"todos": [{"assignee": "张三", "task": "完成文档编写"}]
|
|
}
|
|
)
|
|
print(f"会议纪要已创建: {doc['url']}")
|
|
|
|
# 示例3: 检查图片状态
|
|
print("\n=== 示例3: 检查图片状态 ===")
|
|
images = docx.check_image_status(doc["document_id"])
|
|
if images:
|
|
for img in images:
|
|
status = "✅ 有效" if img["is_valid"] else "❌ 空"
|
|
print(f"图片: {status}, 尺寸: {img['width']}x{img['height']}")
|
|
else:
|
|
print("文档中没有图片")
|